```
### 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.
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain.chat_models import init_chat_model
model = init_chat_model(model="gpt-5.4-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 in graph.astream(
{"topic": "ice cream"},
stream_mode="messages", # [!code highlight]
version="v2", # [!code highlight]
):
if chunk["type"] == "messages":
message_chunk, metadata = chunk["data"]
if message_chunk.content:
print(message_chunk.content, end="|", flush=True)
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import TypedDict
from langgraph.types import StreamWriter
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",
version="v2",
):
if chunk["type"] == "custom":
print(chunk["data"])
```
***
[Connect these docs](/use-these-docs) to Claude, VSCode, and more via MCP for real-time answers.
[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).