Streaming

For new applications, we recommend event streaming—the typed-projection API introduced in Deep Agents v0.6. Event streaming gives you separate iterators per projection (subagents, messages, tool calls, values) so you can consume them independently instead of branching on stream_mode chunks.

Deep Agents build on LangGraph’s streaming infrastructure with first-class support for subagent streams. When a deep agent delegates work to subagents, you can stream updates from each subagent independently—tracking progress, LLM tokens, and tool calls in real time. What’s possible with deep agent streaming:

Stream subagent progress—track each subagent’s execution as it runs in parallel.
Stream LLM tokens—stream tokens from the main agent and each subagent.
Stream tool calls—see tool calls and results from within subagent execution.
Stream custom updates—emit user-defined signals from inside subagent nodes.

Enable subgraph streaming

Deep Agents use LangGraph’s subgraph streaming to surface events from subagent execution. To receive subagent events, enable stream_subgraphs when streaming.

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
):
    if namespace:
        # Subagent event - namespace identifies the source
        print(f"[subagent: {namespace}]")
    else:
        # Main agent event
        print("[main agent]")
    print(data)

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
):
    if namespace:
        # Subagent event - namespace identifies the source
        print(f"[subagent: {namespace}]")
    else:
        # Main agent event
        print("[main agent]")
    print(data)

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="anthropic:claude-sonnet-5",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
):
    if namespace:
        # Subagent event - namespace identifies the source
        print(f"[subagent: {namespace}]")
    else:
        # Main agent event
        print("[main agent]")
    print(data)

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
):
    if namespace:
        # Subagent event - namespace identifies the source
        print(f"[subagent: {namespace}]")
    else:
        # Main agent event
        print("[main agent]")
    print(data)

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
):
    if namespace:
        # Subagent event - namespace identifies the source
        print(f"[subagent: {namespace}]")
    else:
        # Main agent event
        print("[main agent]")
    print(data)

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
):
    if namespace:
        # Subagent event - namespace identifies the source
        print(f"[subagent: {namespace}]")
    else:
        # Main agent event
        print("[main agent]")
    print(data)

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
):
    if namespace:
        # Subagent event - namespace identifies the source
        print(f"[subagent: {namespace}]")
    else:
        # Main agent event
        print("[main agent]")
    print(data)

Namespaces

When subgraphs is enabled, each streaming event includes a namespace that identifies which agent produced it. The namespace is a path of node names and task IDs that represents the agent hierarchy.

Namespace	Source
`()` (empty)	Main agent
`("tools:abc123",)`	A subagent spawned by the main agent’s `task` tool call `abc123`
`("tools:abc123", "model_request:def456")`	The model request node inside a subagent

Use namespaces to route events to the correct UI component:

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Plan my vacation"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Check if this event came from a subagent
    is_subagent = any(
        segment.startswith("tools:") for segment in namespace
    )

    if is_subagent:
        # Extract the tool call ID from the namespace
        tool_call_id = next(
            s.split(":")[1] for s in namespace if s.startswith("tools:")
        )
        print(f"Subagent {tool_call_id}: {data}")
    else:
        print(f"Main agent: {data}")

Subagent progress

Use stream_mode="updates" to track subagent progress as each step completes. This is useful for showing which subagents are active and what work they’ve completed.

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt=(
        "You are a project coordinator with no research knowledge. "
        "For every user request, you must call the task() tool with "
        "subagent_type set to researcher. Never answer research questions yourself. "
        "Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Main agent updates (empty namespace)
    if not namespace:
        for node_name, data in data.items():
            if node_name == "tools":
                # Subagent results returned to main agent
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        print(f"\nSubagent complete: {msg.name}")
                        print(f"  Result: {str(msg.content)[:200]}...")
            else:
                print(f"[main agent] step: {node_name}")

    # Subagent updates (non-empty namespace)
    else:
        for node_name, data in data.items():
            print(f"  [{namespace[0]}] step: {node_name}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt=(
        "You are a project coordinator with no research knowledge. "
        "For every user request, you must call the task() tool with "
        "subagent_type set to researcher. Never answer research questions yourself. "
        "Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Main agent updates (empty namespace)
    if not namespace:
        for node_name, data in data.items():
            if node_name == "tools":
                # Subagent results returned to main agent
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        print(f"\nSubagent complete: {msg.name}")
                        print(f"  Result: {str(msg.content)[:200]}...")
            else:
                print(f"[main agent] step: {node_name}")

    # Subagent updates (non-empty namespace)
    else:
        for node_name, data in data.items():
            print(f"  [{namespace[0]}] step: {node_name}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="anthropic:claude-sonnet-5",
    system_prompt=(
        "You are a project coordinator with no research knowledge. "
        "For every user request, you must call the task() tool with "
        "subagent_type set to researcher. Never answer research questions yourself. "
        "Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Main agent updates (empty namespace)
    if not namespace:
        for node_name, data in data.items():
            if node_name == "tools":
                # Subagent results returned to main agent
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        print(f"\nSubagent complete: {msg.name}")
                        print(f"  Result: {str(msg.content)[:200]}...")
            else:
                print(f"[main agent] step: {node_name}")

    # Subagent updates (non-empty namespace)
    else:
        for node_name, data in data.items():
            print(f"  [{namespace[0]}] step: {node_name}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    system_prompt=(
        "You are a project coordinator with no research knowledge. "
        "For every user request, you must call the task() tool with "
        "subagent_type set to researcher. Never answer research questions yourself. "
        "Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Main agent updates (empty namespace)
    if not namespace:
        for node_name, data in data.items():
            if node_name == "tools":
                # Subagent results returned to main agent
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        print(f"\nSubagent complete: {msg.name}")
                        print(f"  Result: {str(msg.content)[:200]}...")
            else:
                print(f"[main agent] step: {node_name}")

    # Subagent updates (non-empty namespace)
    else:
        for node_name, data in data.items():
            print(f"  [{namespace[0]}] step: {node_name}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    system_prompt=(
        "You are a project coordinator with no research knowledge. "
        "For every user request, you must call the task() tool with "
        "subagent_type set to researcher. Never answer research questions yourself. "
        "Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Main agent updates (empty namespace)
    if not namespace:
        for node_name, data in data.items():
            if node_name == "tools":
                # Subagent results returned to main agent
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        print(f"\nSubagent complete: {msg.name}")
                        print(f"  Result: {str(msg.content)[:200]}...")
            else:
                print(f"[main agent] step: {node_name}")

    # Subagent updates (non-empty namespace)
    else:
        for node_name, data in data.items():
            print(f"  [{namespace[0]}] step: {node_name}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    system_prompt=(
        "You are a project coordinator with no research knowledge. "
        "For every user request, you must call the task() tool with "
        "subagent_type set to researcher. Never answer research questions yourself. "
        "Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Main agent updates (empty namespace)
    if not namespace:
        for node_name, data in data.items():
            if node_name == "tools":
                # Subagent results returned to main agent
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        print(f"\nSubagent complete: {msg.name}")
                        print(f"  Result: {str(msg.content)[:200]}...")
            else:
                print(f"[main agent] step: {node_name}")

    # Subagent updates (non-empty namespace)
    else:
        for node_name, data in data.items():
            print(f"  [{namespace[0]}] step: {node_name}")

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    system_prompt=(
        "You are a project coordinator with no research knowledge. "
        "For every user request, you must call the task() tool with "
        "subagent_type set to researcher. Never answer research questions yourself. "
        "Keep your final response to one sentence."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches topics thoroughly",
            "system_prompt": (
                "You are a thorough researcher. Research the given topic "
                "and provide a concise summary in 2-3 sentences."
            ),
        },
    ],
)

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
    stream_mode="updates",
    subgraphs=True,
):
    # Main agent updates (empty namespace)
    if not namespace:
        for node_name, data in data.items():
            if node_name == "tools":
                # Subagent results returned to main agent
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        print(f"\nSubagent complete: {msg.name}")
                        print(f"  Result: {str(msg.content)[:200]}...")
            else:
                print(f"[main agent] step: {node_name}")

    # Subagent updates (non-empty namespace)
    else:
        for node_name, data in data.items():
            print(f"  [{namespace[0]}] step: {node_name}")

Output

[main agent] step: model_request
  [tools:call_abc123] step: model_request
  [tools:call_abc123] step: tools
  [tools:call_abc123] step: model_request

Subagent complete: task
  Result: ## AI Safety Report...
[main agent] step: model_request

LLM tokens

Use stream_mode="messages" to stream individual tokens from both the main agent and subagents. Each message event includes metadata that identifies the source agent.

current_source = ""

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="messages",
    subgraphs=True,
):
    token, metadata = data

    # Check if this event came from a subagent (namespace contains "tools:")
    is_subagent = any(s.startswith("tools:") for s in namespace)

    if is_subagent:
        # Token from a subagent
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        if subagent_ns != current_source:
            print(f"\n\n--- [subagent: {subagent_ns}] ---")
            current_source = subagent_ns
        if token.content:
            print(token.content, end="", flush=True)
    else:
        # Token from the main agent
        if "main" != current_source:
            print("\n\n--- [main agent] ---")
            current_source = "main"
        if token.content:
            print(token.content, end="", flush=True)

print()

Tool calls

When subagents use tools, you can stream tool call events to display what each subagent is doing. Tool call chunks appear in the messages stream mode.

from langchain.messages import AIMessageChunk, ToolMessage

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research recent quantum computing advances"}]},
    stream_mode="messages",
    subgraphs=True,
):
    token, metadata = data

    # Identify source: "main" or the subagent namespace segment
    is_subagent = any(s.startswith("tools:") for s in namespace)
    source = next((s for s in namespace if s.startswith("tools:")), "main") if is_subagent else "main"

    # Tool call chunks (streaming tool invocations)
    if isinstance(token, AIMessageChunk) and token.tool_call_chunks:
        for tc in token.tool_call_chunks:
            if tc.get("name"):
                print(f"\n[{source}] Tool call: {tc['name']}")
            # Args stream in chunks - write them incrementally
            if tc.get("args"):
                print(tc["args"], end="", flush=True)

    # Tool results
    if isinstance(token, ToolMessage):
        print(f"\n[{source}] Tool result [{token.name}]: {str(token.content)[:150]}")

    # Regular AI content (skip tool call messages)
    if (
        isinstance(token, AIMessageChunk)
        and token.content
        and not token.tool_call_chunks
    ):
        print(token.content, end="", flush=True)

print()

Custom updates

Use get_stream_writer inside your subagent tools to emit custom progress events:

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.

    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
):
    custom_event_count += 1
    is_subagent = any(s.startswith("tools:") for s in namespace)
    if is_subagent:
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        print(f"[{subagent_ns}]", data)
    else:
        print("[main]", data)

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.

    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
):
    custom_event_count += 1
    is_subagent = any(s.startswith("tools:") for s in namespace)
    if is_subagent:
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        print(f"[{subagent_ns}]", data)
    else:
        print("[main]", data)

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.

    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="anthropic:claude-sonnet-5",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
):
    custom_event_count += 1
    is_subagent = any(s.startswith("tools:") for s in namespace)
    if is_subagent:
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        print(f"[{subagent_ns}]", data)
    else:
        print("[main]", data)

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.

    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
):
    custom_event_count += 1
    is_subagent = any(s.startswith("tools:") for s in namespace)
    if is_subagent:
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        print(f"[{subagent_ns}]", data)
    else:
        print("[main]", data)

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.

    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
):
    custom_event_count += 1
    is_subagent = any(s.startswith("tools:") for s in namespace)
    if is_subagent:
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        print(f"[{subagent_ns}]", data)
    else:
        print("[main]", data)

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.

    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
):
    custom_event_count += 1
    is_subagent = any(s.startswith("tools:") for s in namespace)
    if is_subagent:
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        print(f"[{subagent_ns}]", data)
    else:
        print("[main]", data)

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent


@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.

    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()

    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)

    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)

    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )


agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
):
    custom_event_count += 1
    is_subagent = any(s.startswith("tools:") for s in namespace)
    if is_subagent:
        subagent_ns = next(s for s in namespace if s.startswith("tools:"))
        print(f"[{subagent_ns}]", data)
    else:
        print("[main]", data)

Output

[tools:call_abc123] {'status': 'starting', 'topic': 'customer satisfaction trends', 'progress': 0}
[tools:call_abc123] {'status': 'analyzing', 'progress': 50}
[tools:call_abc123] {'status': 'complete', 'progress': 100}

Stream multiple modes

Combine multiple stream modes to get a complete picture of agent execution:

# Skip internal middleware steps - only show meaningful node names
INTERESTING_NODES = {"model", "tools"}

last_source = ""
mid_line = False  # True when we've written tokens without a trailing newline

for namespace, mode, data in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze the impact of remote work on team productivity"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
):
    is_subagent = any(s.startswith("tools:") for s in namespace)
    source = "subagent" if is_subagent else "main"

    if mode == "updates":
        for node_name in data:
            if node_name not in INTERESTING_NODES:
                continue
            if mid_line:
                print()
                mid_line = False
            print(f"[{source}] step: {node_name}")

    elif mode == "messages":
        token, metadata = data
        if token.content:
            # Print a header when the source changes
            if source != last_source:
                if mid_line:
                    print()
                    mid_line = False
                print(f"\n[{source}] ", end="")
                last_source = source
            print(token.content, end="", flush=True)
            mid_line = True

    elif mode == "custom":
        if mid_line:
            print()
            mid_line = False
        print(f"[{source}] custom event:", data)

print()

Common patterns

Track subagent lifecycle

Monitor when subagents start, run, and complete:

active_subagents = {}

for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research the latest AI safety developments"}]},
    stream_mode="updates",
    subgraphs=True,
):
    for node_name, data in data.items():
        # ─── Phase 1: Detect subagent starting ────────────────────────
        # When the main agent's model node contains task tool calls,
        # a subagent has been spawned.
        if not namespace and node_name == "model":
            for msg in data.get("messages", []):
                for tc in getattr(msg, "tool_calls", []):
                    if tc["name"] == "task":
                        active_subagents[tc["id"]] = {
                            "type": tc["args"].get("subagent_type"),
                            "description": tc["args"].get("description", "")[:80],
                            "status": "pending",
                        }
                        print(
                            f'[lifecycle] PENDING  → subagent "{tc["args"].get("subagent_type")}" '
                            f'({tc["id"]})'
                        )

        # ─── Phase 2: Detect subagent running ─────────────────────────
        # When we receive events from a tools:UUID namespace, that
        # subagent is actively executing.
        if namespace and namespace[0].startswith("tools:"):
            pregel_id = namespace[0].split(":")[1]
            # Check if any pending subagent needs to be marked running.
            # Note: the pregel task ID differs from the tool_call_id,
            # so we mark any pending subagent as running on first subagent event.
            for sub_id, sub in active_subagents.items():
                if sub["status"] == "pending":
                    sub["status"] = "running"
                    print(
                        f'[lifecycle] RUNNING  → subagent "{sub["type"]}" '
                        f"(pregel: {pregel_id})"
                    )
                    break

        # ─── Phase 3: Detect subagent completing ──────────────────────
        # When the main agent's tools node returns a tool message,
        # the subagent has completed and returned its result.
        if not namespace and node_name == "tools":
            for msg in data.get("messages", []):
                if msg.type == "tool":
                    sub = active_subagents.get(msg.tool_call_id)
                    if sub:
                        sub["status"] = "complete"
                        print(
                            f'[lifecycle] COMPLETE → subagent "{sub["type"]}" '
                            f"({msg.tool_call_id})"
                        )
                        print(f"  Result preview: {str(msg.content)[:120]}...")

# Print final state
print("\n--- Final subagent states ---")
for sub_id, sub in active_subagents.items():
    print(f"  {sub['type']}: {sub['status']}")

Handle human-in-the-loop interrupts

When interrupt_on is configured, the updates stream may contain __interrupt__ entries that pause execution for human approval. Detect them by checking for the __interrupt__ key in update chunks. Resume with the same flat {"decisions": [...]} payload you use with .invoke():

from langchain_core.utils.uuid import uuid7

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

for mode, data in agent.stream(
    {"messages": [{"role": "user", "content": "Delete temp.txt"}]},
    stream_mode=["messages", "updates"],
    config=config,
):
    if mode == "updates" and isinstance(data, dict):
        if "__interrupt__" in data:
            for interrupt_obj in data["__interrupt__"]:
                print(f"Approval needed: {interrupt_obj.value}")

See Human-in-the-loop: Handle interrupts with streaming for the complete stream-resume loop pattern.

Stream chunk shapes

With subgraphs=True, chunk shape depends on whether you pass one stream mode or several:

`stream_mode`	Chunk shape
Single mode (for example `"updates"`)	`(namespace, data)`
Multiple modes (for example `["messages", "updates"]`)	`(namespace, mode, data)`

Without subgraphs, a single mode yields data directly, and multiple modes yield (mode, data).

# Single mode + subgraphs
for namespace, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing"}]},
    stream_mode="updates",
    subgraphs=True,
):
    print(namespace)  # () for main agent, ("tools:<id>",) for subagent
    print(data)

# Multiple modes + subgraphs
for namespace, mode, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
):
    print(mode)       # "updates", "messages", or "custom"
    print(namespace)  # () for main agent, ("tools:<id>",) for subagent
    print(data)

LangGraph 1.1 and later also support a unified version="v2" StreamPart dict format. See Stream output format (v2). Deep Agents examples on this page use the tuple format, which matches the LangGraph version Deep Agents currently depends on.

Human-in-the-loop—Configure approval workflows for sensitive tool operations
Subagents—Configure and use subagents with Deep Agents
Frontend streaming—Build React UIs with useStream for Deep Agents
LangChain Event Streaming—General streaming concepts with LangChain agents

Connect these docs to your agent of choice via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Deployment

Execution environment

Context management

Delegation

Steering

Middleware

Frontend

Protocols

Enable subgraph streaming

Namespaces

Subagent progress

LLM tokens

Tool calls

Custom updates

Stream multiple modes

Common patterns

Track subagent lifecycle

Handle human-in-the-loop interrupts

Stream chunk shapes

​Enable subgraph streaming

​Namespaces

​Subagent progress

​LLM tokens

​Tool calls

​Custom updates

​Stream multiple modes

​Common patterns

​Track subagent lifecycle

​Handle human-in-the-loop interrupts

​Stream chunk shapes

​Related

Enable subgraph streaming

Namespaces

Subagent progress

LLM tokens

Tool calls

Custom updates

Stream multiple modes

Common patterns

Track subagent lifecycle

Handle human-in-the-loop interrupts

Stream chunk shapes

Related