> ## Documentation Index
> Fetch the complete documentation index at: https://langchain-5e9cc07a-preview-docsby-1791319236-3be7a15.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming

> Stream real-time updates from deep agent runs and subagent execution

<Tip>
  For new applications, we recommend [event streaming](/oss/python/deepagents/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.
</Tip>

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**](#subagent-progress)—track each subagent's execution as it runs in parallel.
* [**Stream LLM tokens**](#llm-tokens)—stream tokens from the main agent and each subagent.
* [**Stream tool calls**](#tool-calls)—see tool calls and results from within subagent execution.
* [**Stream custom updates**](#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.

<CodeGroup>
  ```python Google theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
      stream_mode="updates",
      subgraphs=True,  # [!code highlight]
      version="v2",  # [!code highlight]
  ):
      if chunk["type"] == "updates":
          if chunk["ns"]:
              # Subagent event - namespace identifies the source
              print(f"[subagent: {chunk['ns']}]")
          else:
              # Main agent event
              print("[main agent]")
          print(chunk["data"])
  ```

  ```python OpenAI theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
      stream_mode="updates",
      subgraphs=True,  # [!code highlight]
      version="v2",  # [!code highlight]
  ):
      if chunk["type"] == "updates":
          if chunk["ns"]:
              # Subagent event - namespace identifies the source
              print(f"[subagent: {chunk['ns']}]")
          else:
              # Main agent event
              print("[main agent]")
          print(chunk["data"])
  ```

  ```python Anthropic theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
      stream_mode="updates",
      subgraphs=True,  # [!code highlight]
      version="v2",  # [!code highlight]
  ):
      if chunk["type"] == "updates":
          if chunk["ns"]:
              # Subagent event - namespace identifies the source
              print(f"[subagent: {chunk['ns']}]")
          else:
              # Main agent event
              print("[main agent]")
          print(chunk["data"])
  ```

  ```python OpenRouter theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
      stream_mode="updates",
      subgraphs=True,  # [!code highlight]
      version="v2",  # [!code highlight]
  ):
      if chunk["type"] == "updates":
          if chunk["ns"]:
              # Subagent event - namespace identifies the source
              print(f"[subagent: {chunk['ns']}]")
          else:
              # Main agent event
              print("[main agent]")
          print(chunk["data"])
  ```

  ```python Fireworks theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
      stream_mode="updates",
      subgraphs=True,  # [!code highlight]
      version="v2",  # [!code highlight]
  ):
      if chunk["type"] == "updates":
          if chunk["ns"]:
              # Subagent event - namespace identifies the source
              print(f"[subagent: {chunk['ns']}]")
          else:
              # Main agent event
              print("[main agent]")
          print(chunk["data"])
  ```

  ```python Baseten theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
      stream_mode="updates",
      subgraphs=True,  # [!code highlight]
      version="v2",  # [!code highlight]
  ):
      if chunk["type"] == "updates":
          if chunk["ns"]:
              # Subagent event - namespace identifies the source
              print(f"[subagent: {chunk['ns']}]")
          else:
              # Main agent event
              print("[main agent]")
          print(chunk["data"])
  ```

  ```python Ollama theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
      stream_mode="updates",
      subgraphs=True,  # [!code highlight]
      version="v2",  # [!code highlight]
  ):
      if chunk["type"] == "updates":
          if chunk["ns"]:
              # Subagent event - namespace identifies the source
              print(f"[subagent: {chunk['ns']}]")
          else:
              # Main agent event
              print("[main agent]")
          print(chunk["data"])
  ```
</CodeGroup>

## 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:

```python theme={null}
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Plan my vacation"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        # Check if this event came from a subagent
        is_subagent = any(
            segment.startswith("tools:") for segment in chunk["ns"]
        )

        if is_subagent:
            # Extract the tool call ID from the namespace
            tool_call_id = next(
                s.split(":")[1] for s in chunk["ns"] if s.startswith("tools:")
            )
            print(f"Subagent {tool_call_id}: {chunk['data']}")
        else:
            print(f"Main agent: {chunk['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.

<CodeGroup>
  ```python Google theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
      stream_mode="updates",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "updates":
          # Main agent updates (empty namespace)
          if not chunk["ns"]:
              for node_name, data in chunk["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 chunk["data"].items():
                  print(f"  [{chunk['ns'][0]}] step: {node_name}")
  ```

  ```python OpenAI theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
      stream_mode="updates",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "updates":
          # Main agent updates (empty namespace)
          if not chunk["ns"]:
              for node_name, data in chunk["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 chunk["data"].items():
                  print(f"  [{chunk['ns'][0]}] step: {node_name}")
  ```

  ```python Anthropic theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
      stream_mode="updates",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "updates":
          # Main agent updates (empty namespace)
          if not chunk["ns"]:
              for node_name, data in chunk["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 chunk["data"].items():
                  print(f"  [{chunk['ns'][0]}] step: {node_name}")
  ```

  ```python OpenRouter theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
      stream_mode="updates",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "updates":
          # Main agent updates (empty namespace)
          if not chunk["ns"]:
              for node_name, data in chunk["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 chunk["data"].items():
                  print(f"  [{chunk['ns'][0]}] step: {node_name}")
  ```

  ```python Fireworks theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
      stream_mode="updates",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "updates":
          # Main agent updates (empty namespace)
          if not chunk["ns"]:
              for node_name, data in chunk["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 chunk["data"].items():
                  print(f"  [{chunk['ns'][0]}] step: {node_name}")
  ```

  ```python Baseten theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
      stream_mode="updates",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "updates":
          # Main agent updates (empty namespace)
          if not chunk["ns"]:
              for node_name, data in chunk["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 chunk["data"].items():
                  print(f"  [{chunk['ns'][0]}] step: {node_name}")
  ```

  ```python Ollama theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Write a short summary about AI safety"}]},
      stream_mode="updates",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "updates":
          # Main agent updates (empty namespace)
          if not chunk["ns"]:
              for node_name, data in chunk["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 chunk["data"].items():
                  print(f"  [{chunk['ns'][0]}] step: {node_name}")
  ```
</CodeGroup>

```shell title="Output" theme={null}
[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.

```python theme={null}
current_source = ""

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

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

        if is_subagent:
            # Token from a subagent
            subagent_ns = next(s for s in chunk["ns"] 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.

```python theme={null}
from langchain.messages import AIMessageChunk, ToolMessage

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

        # Identify source: "main" or the subagent namespace segment
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
        source = next((s for s in chunk["ns"] 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`](https://reference.langchain.com/python/langgraph/config/get_stream_writer) inside your subagent tools to emit custom progress events:

<CodeGroup>
  ```python Google theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
      stream_mode="custom",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "custom":
          custom_event_count += 1
          is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
          if is_subagent:
              subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
              print(f"[{subagent_ns}]", chunk["data"])
          else:
              print("[main]", chunk["data"])
  ```

  ```python OpenAI theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
      stream_mode="custom",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "custom":
          custom_event_count += 1
          is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
          if is_subagent:
              subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
              print(f"[{subagent_ns}]", chunk["data"])
          else:
              print("[main]", chunk["data"])
  ```

  ```python Anthropic theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
      stream_mode="custom",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "custom":
          custom_event_count += 1
          is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
          if is_subagent:
              subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
              print(f"[{subagent_ns}]", chunk["data"])
          else:
              print("[main]", chunk["data"])
  ```

  ```python OpenRouter theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
      stream_mode="custom",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "custom":
          custom_event_count += 1
          is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
          if is_subagent:
              subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
              print(f"[{subagent_ns}]", chunk["data"])
          else:
              print("[main]", chunk["data"])
  ```

  ```python Fireworks theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
      stream_mode="custom",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "custom":
          custom_event_count += 1
          is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
          if is_subagent:
              subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
              print(f"[{subagent_ns}]", chunk["data"])
          else:
              print("[main]", chunk["data"])
  ```

  ```python Baseten theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
      stream_mode="custom",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "custom":
          custom_event_count += 1
          is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
          if is_subagent:
              subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
              print(f"[{subagent_ns}]", chunk["data"])
          else:
              print("[main]", chunk["data"])
  ```

  ```python Ollama theme={null}
  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 chunk in agent.stream(
      {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
      stream_mode="custom",
      subgraphs=True,
      version="v2",
  ):
      if chunk["type"] == "custom":
          custom_event_count += 1
          is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
          if is_subagent:
              subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
              print(f"[{subagent_ns}]", chunk["data"])
          else:
              print("[main]", chunk["data"])
  ```
</CodeGroup>

```shell title="Output" theme={null}
[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:

```python theme={null}
# 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 chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze the impact of remote work on team productivity"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
    version="v2",
):
    is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
    source = "subagent" if is_subagent else "main"

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

    elif chunk["type"] == "messages":
        token, metadata = chunk["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 chunk["type"] == "custom":
        if mid_line:
            print()
            mid_line = False
        print(f"[{source}] custom event:", chunk["data"])

print()
```

## Common patterns

### Track subagent lifecycle

Monitor when subagents start, run, and complete:

```python theme={null}
active_subagents = {}

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research the latest AI safety developments"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        for node_name, data in chunk["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 chunk["ns"] 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 chunk["ns"] and chunk["ns"][0].startswith("tools:"):
                pregel_id = chunk["ns"][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 chunk["ns"] 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`](/oss/python/deepagents/human-in-the-loop) 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()`:

```python theme={null}
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:  # [!code highlight]
            for interrupt_obj in data["__interrupt__"]:
                print(f"Approval needed: {interrupt_obj.value}")
```

See [Human-in-the-loop: Handle interrupts with streaming](/oss/python/deepagents/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)`.

```python theme={null}
# 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)](/oss/python/langgraph/streaming#stream-output-format-v2). Deep Agents examples on this page use the tuple format, which matches the LangGraph version Deep Agents currently depends on.

## Related

* [Human-in-the-loop](/oss/python/deepagents/human-in-the-loop)—Configure approval workflows for sensitive tool operations
* [Subagents](/oss/python/deepagents/subagents)—Configure and use subagents with Deep Agents
* [Frontend streaming](/oss/python/deepagents/frontend/overview)—Build React UIs with [`useStream`](https://reference.langchain.com/javascript/langchain-react/index/useStream) for Deep Agents
* [LangChain Event Streaming](/oss/python/langchain/event-streaming)—General streaming concepts with LangChain agents

***

<div className="source-links">
  <Callout icon="terminal-2">
    [Connect these docs](/use-these-docs) to your agent of choice via MCP for real-time answers.
  </Callout>

  <Callout icon="edit">
    [Edit this page on GitHub](https://github.com/langchain-ai/docs/edit/main/src/oss/deepagents/streaming.mdx) or [file an issue](https://github.com/langchain-ai/docs/issues/new/choose).
  </Callout>
</div>