model = ChatOpenAI(model=“gpt-4.1-mini”)

class State(TypedDict): topic: str joke: str poem: str

def write_joke(state: State): topic = state[“topic”] joke_response = model.invoke( [{“role”: “user”, “content”: f"Write a joke about {topic}"}] ) return {“joke”: joke_response.content}

def write_poem(state: State): topic = state[“topic”] poem_response = model.invoke( [{“role”: “user”, “content”: f"Write a short poem about {topic}"}] ) return {“poem”: poem_response.content}

graph = ( StateGraph(State) .add_node(write_joke) .add_node(write_poem) # write both the joke and the poem concurrently .add_edge(START, “write_joke”) .add_edge(START, “write_poem”) .compile() )

The “messages” stream mode returns a tuple of (message_chunk, metadata)#

where message_chunk is the token streamed by the LLM and metadata is a dictionary#

with information about the graph node where the LLM was called and other information#

for msg, metadata in graph.stream( {“topic”: “cats”}, stream_mode=“messages”, # [!code highlight] ): # Filter the streamed tokens by the langgraph_node field in the metadata # to only include the tokens from the write_poem node if msg.content and metadata[“langgraph_node”] == “write_poem”: print(msg.content, end="|", flush=True)

</Accordion>

## Stream custom data

To send **custom user-defined data** from inside a LangGraph node or tool, follow these steps:

1. Use [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer) to access the stream writer and emit custom data.
2. Set `stream_mode="custom"` when calling `.stream()` or `.astream()` to get the custom data in the stream. You can combine multiple modes (e.g., `["updates", "custom"]`), but at least one must be `"custom"`.

<span class="callout-start" data-callout-type="warning"></span>
**No [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer) in async for Python \< 3.11**
In async code running on Python \< 3.11, [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer) will not work.
Instead, add a `writer` parameter to your node or tool and pass it manually.
See [Async with Python \< 3.11](#async) for usage examples.
<span class="callout-end"></span>

<span class="tab-group-start"></span>
<span class="tab-start" data-tab-title="node"></span>
```python
  from typing import TypedDict
  from langgraph.config import get_stream_writer
  from langgraph.graph import StateGraph, START

  class State(TypedDict):
      query: str
      answer: str

  def node(state: State):
      # Get the stream writer to send custom data
      writer = get_stream_writer()
      # Emit a custom key-value pair (e.g., progress update)
      writer({"custom_key": "Generating custom data inside node"})
      return {"answer": "some data"}

  graph = (
      StateGraph(State)
      .add_node(node)
      .add_edge(START, "node")
      .compile()
  )

  inputs = {"query": "example"}

  # Set stream_mode="custom" to receive the custom data in the stream
  for chunk in graph.stream(inputs, stream_mode="custom"):
      print(chunk)
  ```
<span class="tab-end"></span>

<span class="tab-start" data-tab-title="tool"></span>
```python
  from langchain.tools import tool
  from langgraph.config import get_stream_writer

  @tool
  def query_database(query: str) -> str:
      """Query the database."""
      # Access the stream writer to send custom data
      writer = get_stream_writer()  # [!code highlight]
      # Emit a custom key-value pair (e.g., progress update)
      writer({"data": "Retrieved 0/100 records", "type": "progress"})  # [!code highlight]
      # perform query
      # Emit another custom key-value pair
      writer({"data": "Retrieved 100/100 records", "type": "progress"})
      return "some-answer"


  graph = ... # define a graph that uses this tool

  # Set stream_mode="custom" to receive the custom data in the stream
  for chunk in graph.stream(inputs, stream_mode="custom"):
      print(chunk)
  ```
<span class="tab-end"></span>
<span class="tab-group-end"></span>

## Use with any LLM

You can use `stream_mode="custom"` to stream data from **any LLM API** — even if that API does **not** implement the LangChain chat model interface.

This lets you integrate raw LLM clients or external services that provide their own streaming interfaces, making LangGraph highly flexible for custom setups.

```python
from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
  """Example node that calls an arbitrary model and streams the output"""
  # Get the stream writer to send custom data
  writer = get_stream_writer()  # [!code highlight]
  # Assume you have a streaming client that yields chunks
  # Generate LLM tokens using your custom streaming client
  for chunk in your_custom_streaming_client(state["topic"]):
      # Use the writer to send custom data to the stream
      writer({"custom_llm_chunk": chunk})  # [!code highlight]
  return {"result": "completed"}

graph = (
  StateGraph(State)
  .add_node(call_arbitrary_model)
  # Add other nodes and edges as needed
  .compile()
)
# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(
  {"topic": "cats"},
  stream_mode="custom",  # [!code highlight]

):
  # The chunk will contain the custom data streamed from the llm
  print(chunk)

      if delta.role is not None:
          role = delta.role

      if delta.content:
          yield {"role": role, "content": delta.content}

  return response

  function_name = tool_call["function"]["name"]
  if function_name != "get_items":
      raise ValueError(f"Tool {function_name} not supported")

  function_arguments = tool_call["function"]["arguments"]
  arguments = json.loads(function_arguments)

  function_response = await get_items(**arguments)
  tool_message = {
      "tool_call_id": tool_call["id"],
      "role": "tool",
      "name": function_name,
      "content": function_response,
  }
  return {"messages": [tool_message]}

Let's invoke the graph with an [`AIMessage`](https://reference.langchain.com/python/langchain-core/messages/ai/AIMessage) that includes a tool call:

```python
inputs = {
    "messages": [
        {
            "content": None,
            "role": "assistant",
            "tool_calls": [
                {
                    "id": "1",
                    "function": {
                        "arguments": '{"place":"bedroom"}',
                        "name": "get_items",
                    },
                    "type": "function",
                }
            ],
        }
    ]
}

async for chunk in graph.astream(
    inputs,
    stream_mode="custom",
):
    print(chunk["content"], end="|", flush=True)

Disable streaming for specific chat models#

If your application mixes models that support streaming with those that do not, you may need to explicitly disable streaming for models that do not support it.

Set streaming=False when initializing the model.

    from langchain.chat_models import init_chat_model

    model = init_chat_model(
        "claude-sonnet-4-5-20250929",
        # Set streaming=False to disable streaming for the chat model
        streaming=False  # [!code highlight]
    )
    ```
  <span class="tab-end"></span>

  <span class="tab-start" data-tab-title="Chat model interface"></span>
```python
    from langchain_openai import ChatOpenAI

    # Set streaming=False to disable streaming for the chat model
    model = ChatOpenAI(model="o1-preview", streaming=False)
    ```
  <span class="tab-end"></span>
<span class="tab-group-end"></span>

<span class="callout-start" data-callout-type="note"></span>
  Not all chat model integrations support the `streaming` parameter. If your model doesn't support it, use `disable_streaming=True` instead. This parameter is available on all chat models via the base class.
<span class="callout-end"></span>

<a id="async" />

### Async with Python \< 3.11

In Python versions \< 3.11, [asyncio tasks](https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task) do not support the `context` parameter.
This limits LangGraph ability to automatically propagate context, and affects LangGraph's streaming mechanisms in two key ways:

1. You **must** explicitly pass [`RunnableConfig`](https://python.langchain.com/docs/concepts/runnables/#runnableconfig) into async LLM calls (e.g., `ainvoke()`), as callbacks are not automatically propagated.
2. You **cannot** use [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer) in async nodes or tools — you must pass a `writer` argument directly.

<Accordion title="Extended example: async LLM call with manual config">
  ```python
  from typing import TypedDict
  from langgraph.graph import START, StateGraph
  from langchain.chat_models import init_chat_model

  model = init_chat_model(model="gpt-4.1-mini")

  class State(TypedDict):
      topic: str
      joke: str

  # Accept config as an argument in the async node function
  async def call_model(state, config):
      topic = state["topic"]
      print("Generating joke...")
      # Pass config to model.ainvoke() to ensure proper context propagation
      joke_response = await model.ainvoke(  # [!code highlight]
          [{"role": "user", "content": f"Write a joke about {topic}"}],
          config,
      )
      return {"joke": joke_response.content}

  graph = (
      StateGraph(State)
      .add_node(call_model)
      .add_edge(START, "call_model")
      .compile()
  )

  # Set stream_mode="messages" to stream LLM tokens
  async for chunk, metadata in graph.astream(
      {"topic": "ice cream"},
      stream_mode="messages",  # [!code highlight]
  ):
      if chunk.content:
          print(chunk.content, end="|", flush=True)

class State(TypedDict): topic: str joke: str

Add writer as an argument in the function signature of the async node or tool#

LangGraph will automatically pass the stream writer to the function#

async def generate_joke(state: State, writer: StreamWriter): # [!code highlight] writer({“custom_key”: “Streaming custom data while generating a joke”}) return {“joke”: f"This is a joke about {state[’topic’]}”}

graph = ( StateGraph(State) .add_node(generate_joke) .add_edge(START, “generate_joke”) .compile() )

Set stream_mode=“custom” to receive the custom data in the stream # [!code highlight]#

async for chunk in graph.astream( {“topic”: “ice cream”}, stream_mode=“custom”, ): print(chunk)

</Accordion>

***

<span class="callout-start" data-callout-type="note"></span>
[Edit this page on GitHub](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/streaming.mdx) or [file an issue](https://github.com/langchain-ai/docs/issues/new/choose).
<span class="callout-end"></span>

<span class="callout-start" data-callout-type="note"></span>
[Connect these docs](/use-these-docs) to Claude, VSCode, and more via MCP for real-time answers.
<span class="callout-end"></span>