Stream with the Functional API

Stream from a Functional API entrypoint in two ways. Custom data is a side channel: you push progress with a writer, then read custom chunks. Values are state snapshots the runtime records, including the entrypoint return value. You do not call a writer for values. For concepts, see the Functional API overview. For messages and other typed projections, see event streaming.

Emit custom data

The Functional API uses the same stream as the Graph API. Emit with a writer, then consume custom chunks. For stream modes, see Streaming.

from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.config import get_stream_writer
from langgraph.func import entrypoint

checkpointer = InMemorySaver()


@entrypoint(checkpointer=checkpointer)
def main(inputs: dict) -> int:
    writer = get_stream_writer()
    writer("Started processing")
    result = inputs["x"] * 2
    writer(f"Result is {result}")
    return result


config = {"configurable": {"thread_id": str(uuid7())}}

for chunk in main.stream({"x": 5}, config, stream_mode="custom"):
    print(chunk)
# Started processing
# Result is 10

View example trace

Open a public LangSmith run for this example.

Import get_stream_writer from langgraph.config.
Call it inside the entrypoint to get a writer.
Emit custom data before and after the computation.
Call stream(..., stream_mode="custom") and iterate the chunks.

Async with Python < 3.11 In async code on Python < 3.11, get_stream_writer does not work. Use the injectable StreamWriter parameter instead. See Async with Python < 3.11.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint
from langgraph.types import StreamWriter

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
async def main(inputs: dict, writer: StreamWriter) -> int:
    writer("Started processing")
    return inputs["x"] * 2

Stream values

Reuse the main entrypoint from Emit custom data. Iterate stream.values for state snapshots. For messages and other projections, see event streaming.

values_config = {"configurable": {"thread_id": str(uuid7())}}
stream = main.stream_events({"x": 5}, config=values_config, version="v3")
for snapshot in stream.values:
    print(snapshot)
# 10

Get started

Capabilities

Patterns

Build with the Graph API

Deployment

Frontend

Advanced

Stream with the Functional API

Emit custom data

View example trace

Stream values

See also

​Emit custom data

View example trace

​Stream values

​See also

Emit custom data

Stream values

See also