5. Streaming Tool Calls

Overview

Imagine your agent gets a request: "Tell me the weather in Mexico City, the flights to Cancun, and recommend me a hotel." The agent needs to call 3 tools — get_weather, search_flights, find_hotel. Each tool takes 2-4 seconds. Without streaming, the user stares at a blank screen for 8-12 seconds. No spinner, no message, nothing. As far as the user is concerned, the app is frozen.

Streaming solves this. Instead of waiting for everything to finish before showing anything, you receive tokens and tool call chunks as they're generated. You can show "Thinking...", then "Looking up the weather in Mexico City...", then "Searching flights to Cancun...", and finally stream the answer token by token. The user knows exactly what's happening at every moment.

This isn't a nice-to-have. In production, streaming is the difference between an app users abandon (because it looks broken) and one that feels fast and smart (because it communicates progress). Every modern chat app — ChatGPT, Claude, Gemini — uses streaming. Your agents should do the same.

The Problem: The "Frozen" App

Why perceived speed matters

There's a concept in UX called perceived performance — the speed the user perceives, not the actual speed. A 10-second operation with feedback ("Searching...", "Processing...", "Almost there...") feels faster than a 5-second one with no feedback.

With agents, this is amplified. An agent that runs 3 tools in sequence can easily take 10 seconds:

Without streaming:
┌──────────────────────────────────────────────┐
│ User sends a message                         │
│ ...                                          │
│ (10 seconds of nothing)                      │
│ ...                                          │
│ The full answer appears all at once          │
└──────────────────────────────────────────────┘

With streaming:
┌──────────────────────────────────────────────┐
│ User sends a message                         │
│ → "Thinking..."                     (0.3s)   │
│ → "Looking up weather in CDMX..."   (0.5s)   │
│ → "Result: 24°C, sunny"             (2.5s)   │
│ → "Searching flights to Cancun..."  (2.6s)   │
│ → "Found 3 options..."              (5.0s)   │
│ → "Searching hotels..."             (5.1s)   │
│ → Final answer, token by token      (8.0s)   │
└──────────────────────────────────────────────┘

Same total time. Completely different experience.

The real cost of no streaming

Without streaming: abandonment (users close the app thinking it crashed), duplicate requests (they resend the message → double the cost), distrust ("Is it doing anything?"), and unnecessary support tickets. Implementing streaming takes 30 minutes. Not doing it costs you users.

Token Streaming

The basics: model.stream()

Before you bring tool calls into it, understand how plain text streaming works. Instead of model.invoke(), which waits for the complete response, model.stream() returns an iterator of chunks:

from langchain.chat_models import init_chat_model

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

for chunk in model.stream("Explain what an API is in 2 sentences"):
    print(chunk.content, end="", flush=True)

Each chunk is an AIMessageChunk with a fragment of the text. The chunks arrive while the model generates — you don't wait for the whole answer to finish. The end="" avoids line breaks, and flush=True forces immediate output (no buffering).

Anatomy of an AIMessageChunk

Each chunk is an AIMessageChunk — it has the same properties as an AIMessage but represents a fragment. The key property: tool_call_chunks — empty when it's generating text, populated when it's generating tool calls.

Accumulating chunks into a complete message

Chunks can be added together with + to rebuild the complete message:

from langchain.chat_models import init_chat_model

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

chunks = list(model.stream("Say hello in 3 languages"))

full_message = chunks[0]
for chunk in chunks[1:]:
    full_message = full_message + chunk

print(full_message.content)
# "Hello! (English), ¡Hola! (Spanish), Bonjour! (French)"

This matters: when the model decides to call tools, the arguments arrive fragmented. You need to accumulate them before executing.

Streaming Tool Calls: ToolCallChunks

What changes when there are tools

When a model with tools decides to invoke a tool instead of generating text, the chunks contain tool_call_chunks instead of content. Each chunk carries a fragment of the tool name and/or of its JSON arguments:

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"24°C and sunny in {city}"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])

messages = [("user", "What's the weather in Guadalajara?")]

for chunk in model_with_tools.stream(messages):
    if chunk.tool_call_chunks:
        print(f"TOOL CHUNK: {chunk.tool_call_chunks}")
    elif chunk.content:
        print(f"TEXT: '{chunk.content}'")

The output looks something like this:

TOOL CHUNK: [{'name': 'get_weather', 'args': '', 'id': 'call_abc123', 'index': 0}]
TOOL CHUNK: [{'name': None, 'args': '{"ci', 'id': None, 'index': 0}]
TOOL CHUNK: [{'name': None, 'args': 'ty":', 'id': None, 'index': 0}]
TOOL CHUNK: [{'name': None, 'args': ' "Gua', 'id': None, 'index': 0}]
TOOL CHUNK: [{'name': None, 'args': 'dalaj', 'id': None, 'index': 0}]
TOOL CHUNK: [{'name': None, 'args': 'ara"}', 'id': None, 'index': 0}]

Understanding ToolCallChunks

Each tool_call_chunk is a dictionary with:

The first chunk carries the name and the id. The following ones carry args fragments — the JSON gets built piece by piece. The index tells you which tool call it is (relevant when there are parallel calls with multiple tools).

Detecting which type of chunk you have

In streaming, a chunk is text or a tool call, never both at the same time. This pattern is the foundation of all streaming with tools:

for chunk in model_with_tools.stream(messages):
    if chunk.tool_call_chunks:
        for tc_chunk in chunk.tool_call_chunks:
            if tc_chunk["name"]:
                print(f"→ Calling: {tc_chunk['name']}")
            if tc_chunk["args"]:
                print(f"  args fragment: {tc_chunk['args']}")
    elif chunk.content:
        print(chunk.content, end="", flush=True)

Accumulating ToolCallChunks

You can't execute a tool with '{"ci' as its argument. You need the complete JSON '{"city": "Guadalajara"}'. The solution: accumulate every chunk with + and then read tool_calls (not tool_call_chunks) from the accumulated message.

Pattern: Accumulate, then execute

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"24°C and sunny in {city}"

@tool
def search_flights(origin: str, destination: str) -> str:
    """Search flights between two cities."""
    return f"3 flights found from {origin} to {destination}"

tools = [get_weather, search_flights]
tools_by_name = {t.name: t for t in tools}

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

messages = [("user", "Weather in Mexico City and flights from Mexico City to Cancun")]

# Phase 1: Stream and accumulate, showing progress
full_message = None
seen_tools = set()

for chunk in model_with_tools.stream(messages):
    if full_message is None:
        full_message = chunk
    else:
        full_message = full_message + chunk

    for tc_chunk in chunk.tool_call_chunks:
        if tc_chunk["name"] and tc_chunk["name"] not in seen_tools:
            seen_tools.add(tc_chunk["name"])
            print(f"→ Preparing call to: {tc_chunk['name']}...")

    if chunk.content:
        print(chunk.content, end="", flush=True)

# Phase 2: Execute tools with the complete message
if full_message.tool_calls:
    print(f"\n\nExecuting {len(full_message.tool_calls)} tool(s)...")

    messages.append(full_message)

    for tc in full_message.tool_calls:
        print(f"  → {tc['name']}({tc['args']})")
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
        print(f"  ✓ Result: {result}")

The difference between tool_call_chunks and tool_calls

Use tool_call_chunks for UX (showing progress). Use tool_calls from the accumulated message to execute.

Streaming in an Agent Loop

The complete loop with streaming

A real agent runs a loop: (1) the LLM decides, (2) if there are tool calls → execute → append results → back to 1, (3) if there are no tool calls → return the final text. With streaming, every step shows progress:

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"24°C and sunny in {city}"

@tool
def get_population(city: str) -> str:
    """Get the population of a city."""
    return f"{city} has 9.2 million inhabitants"

tools = [get_weather, get_population]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)


def stream_agent_loop(user_message: str, max_iterations: int = 5):
    messages = [HumanMessage(content=user_message)]

    for iteration in range(1, max_iterations + 1):
        print(f"\n--- Iteration {iteration} ---")

        full_message = None
        seen_tools = set()

        for chunk in model_with_tools.stream(messages):
            full_message = chunk if full_message is None else full_message + chunk

            for tc_chunk in chunk.tool_call_chunks:
                if tc_chunk["name"] and tc_chunk["name"] not in seen_tools:
                    seen_tools.add(tc_chunk["name"])
                    print(f"→ Calling {tc_chunk['name']}...", flush=True)

            if chunk.content:
                print(chunk.content, end="", flush=True)

        messages.append(full_message)

        if not full_message.tool_calls:
            return full_message.content

        for tc in full_message.tool_calls:
            result = tools_by_name[tc["name"]].invoke(tc["args"])
            print(f"  ✓ {tc['name']}: {result}")
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))


response = stream_agent_loop("What's the weather in Guadalajara and how many people live there?")

The user never sees an empty screen. In every iteration they know which tools get called, which results come back, and when the final answer starts.

Error handling in the loop

In production, tools can fail. Wrap each invocation in try/except and send the error back as a ToolMessage — the model can decide to retry, use another tool, or answer with what it has:

for tc in full_message.tool_calls:
    try:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        print(f"  ✓ {tc['name']}: OK")
    except Exception as e:
        result = f"Error executing {tc['name']}: {str(e)}"
        print(f"  ✗ {tc['name']}: {e}")

    messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

This keeps the loop running even if a tool fails.

Streaming with LangGraph

From manual loops to LangGraph streaming

Manual loops work for understanding the mechanics. But in production you use LangGraph — and LangGraph has its own, far more powerful streaming system.

agent.stream() — Streaming by graph events

create_react_agent returns a compiled graph. Its stream() method emits one event per node it executes. With stream_mode="updates", you see each node's output (agent, tools) as it finishes — ideal for dashboards and progress logs.

stream_mode options

stream_mode="messages" — Token by token

For ChatGPT-style UX where the answer appears letter by letter, "messages" is what you need:

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"24°C and sunny in {city}"

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(model, [get_weather])

for chunk, metadata in agent.stream(
    {"messages": [("user", "What's the weather in Guadalajara?")]},
    stream_mode="messages",
):
    if chunk.content and metadata["langgraph_node"] == "agent":
        print(chunk.content, end="", flush=True)

    if hasattr(chunk, "tool_call_chunks") and chunk.tool_call_chunks:
        for tc_chunk in chunk.tool_call_chunks:
            if tc_chunk["name"]:
                print(f"\n[Tool: {tc_chunk['name']}]", flush=True)

The metadata["langgraph_node"] tells you which node generated that chunk — useful for filtering and showing only what the user needs to see.

astream_events — Granular async streaming

For async applications (FastAPI, websockets), astream_events gives you maximum control:

import asyncio
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent


@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"24°C and sunny in {city}"


model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(model, [get_weather])


async def stream_with_events():
    async for event in agent.astream_events(
        {"messages": [("user", "Weather in Guadalajara")]},
        version="v2",
    ):
        kind = event["event"]

        if kind == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content:
                print(chunk.content, end="", flush=True)
            if chunk.tool_call_chunks:
                for tc in chunk.tool_call_chunks:
                    if tc["name"]:
                        print(f"\n→ Calling {tc['name']}...", flush=True)

        elif kind == "on_tool_start":
            tool_name = event["name"]
            print(f"\n⚙ Executing {tool_name}...", flush=True)

        elif kind == "on_tool_end":
            tool_name = event["name"]
            result = event["data"]["output"]
            content = result.content if hasattr(result, "content") else str(result)
            print(f"  ✓ {tool_name}: {content[:80]}", flush=True)


asyncio.run(stream_with_events())

astream_events emits events for every internal step — ideal for production UIs where you show specific indicators at each phase.

When to Use Streaming

General rule: if a human is waiting in real time → streaming. If it's automated processing → invoke.

Streaming as Server-Sent Events (SSE)

In production, your backend streams to the frontend via SSE with FastAPI's StreamingResponse. Each chunk is sent as data: {"type": "token", "content": "..."}\n\n. The frontend consumes it with EventSource or fetch + ReadableStream. You'll see a complete example of this in exercise 5.

Connection to the Project

This module's project (capsule 08) is an extraction + routing system. Streaming plays a central role: when the system receives a long document and runs extraction → classification → routing sequentially, the user needs to see progress at every phase.

What you learned here — detecting tool_call_chunks, accumulating messages, showing "Calling X..." — you'll apply directly in the project. And the LangGraph streaming patterns (stream_mode="messages" and astream_events) you'll use heavily starting in module 4 when you work with state machines and the evolving project.

Troubleshooting

Problem 1: "The tool_call_chunks are empty but the model does call tools"

Symptom: You use model.stream() and chunk.tool_call_chunks is always [], but model.invoke() with the same prompt does return tool_calls.

Cause: Not every provider returns tool call chunks the same way. Some models or configurations don't support streaming tool calls.

Solution: Check that you're using a model that supports streaming tool calls (GPT-4.1, Claude 3.5/4, Gemini 2). If you use a proxy or wrapper, make sure it isn't buffering the complete response. As a fallback, accumulate chunks and check full_message.tool_calls:

full = None
for chunk in model_with_tools.stream(messages):
    full = chunk if full is None else full + chunk
# full.tool_calls will have the complete tool calls even if tool_call_chunks was empty

Problem 2: "Error parsing args — incomplete JSON"

Symptom: You try to json.loads(tc_chunk["args"]) during streaming and get a JSONDecodeError.

Cause: The args in tool_call_chunks are JSON fragments, not complete JSON. '{"ci' is not valid JSON.

Solution: Never parse tool_call_chunks directly. Accumulate every chunk with + and use full_message.tool_calls, which already has the parsed JSON:

full_message = None
for chunk in model_with_tools.stream(messages):
    full_message = chunk if full_message is None else full_message + chunk

for tc in full_message.tool_calls:
    print(tc["args"])  # Already-parsed dict, ready to use

Problem 3: "Streaming feels slow — there's a long initial delay"

Symptom: The first tokens take 2-3 seconds to arrive. After that they flow fast.

Cause: The "time to first token" (TTFT) depends on the model and the prompt's complexity. It's normal for the first token to take longer than the ones after it.

Solution: Show an indicator immediately, before you start the stream. Don't wait for the first token to give feedback:

print("Thinking...", flush=True)
for chunk in model_with_tools.stream(messages):
    ...

Also consider using faster models for simple tasks (gpt-4.1-mini vs gpt-4.1) — the TTFT can be 2-3x lower.

Problem 4: "In astream_events, events arrive in an unexpected order"

Symptom: You get on_tool_end before on_chat_model_stream, or events that look out of order.

Cause: astream_events emits events from every internal component. If there's parallel tool execution, the events interleave.

Solution: Filter by the event name and by langgraph_node in the metadata. For debugging, print every event with its type:

async for event in agent.astream_events(inputs, version="v2"):
    print(f"[{event['event']}] {event.get('name', '?')}")

That shows you the real flow and helps you decide which events to filter.

Problem 5: "Streaming with parallel tool calls — the chunks get mixed up"

Symptom: The model calls 2 tools in parallel and the tool_call_chunks from both tools arrive interleaved.

Cause: That's normal behavior. When there are parallel calls, the chunks interleave, identified by index.

Solution: Use each tool_call_chunk's index field to separate and group chunks by tool call. Keep a {index: name} dictionary to track which tool corresponds to each index.

Exercises

Exercise 1: Basic streaming with a progress indicator (Easy)

Create a model with 2 tools (get_weather and get_time) and stream "What time is it in Tokyo and what's the weather like?". Show "Thinking..." at the start, "→ Calling [tool]..." when you detect tool calls, and the final answer token by token. After accumulating the tool calls, execute them, append the ToolMessages, and do a second stream for the final answer.

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"18°C and cloudy in {city}"

@tool
def get_time(timezone: str) -> str:
    """Get the current time in a timezone."""
    return f"It's 3:42 PM in {timezone}"

tools = [get_weather, get_time]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)
messages = [("user", "What time is it in Tokyo and what's the weather like?")]

print("Thinking...", flush=True)
full_message = None
seen_tools = set()

for chunk in model_with_tools.stream(messages):
    full_message = chunk if full_message is None else full_message + chunk
    for tc_chunk in chunk.tool_call_chunks:
        if tc_chunk["name"] and tc_chunk["name"] not in seen_tools:
            seen_tools.add(tc_chunk["name"])
            print(f"→ Calling {tc_chunk['name']}...", flush=True)

if full_message.tool_calls:
    messages.append(full_message)
    for tc in full_message.tool_calls:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    print("--- Final answer ---")
    for chunk in model_with_tools.stream(messages):
        if chunk.content:
            print(chunk.content, end="", flush=True)

Exercise 2: Accumulate chunks and count fragments (Easy)

Stream a tool call and count how many chunks arrive in total, how many contain tool_call_chunks, and how many contain content. Print the summary at the end along with the complete tool_calls from the accumulated message.

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def analyze_text(text: str, language: str) -> str:
    """Analyze a text in the given language."""
    return f"Analysis of '{text[:20]}...' in {language}: positive"

model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([analyze_text])
messages = [("user", "Analyze in English: 'The new economic policy generates optimism'")]

total_chunks = 0
tool_chunks = 0
content_chunks = 0
full_message = None

for chunk in model_with_tools.stream(messages):
    total_chunks += 1
    full_message = chunk if full_message is None else full_message + chunk
    if chunk.tool_call_chunks:
        tool_chunks += 1
    if chunk.content:
        content_chunks += 1

print(f"Total: {total_chunks} | Tool: {tool_chunks} | Content: {content_chunks}")
print(f"Complete tool calls: {full_message.tool_calls}")

Exercise 3: Agent loop with streaming and a max iteration count (Medium)

Implement an agent loop that: (1) uses streaming in every iteration, (2) shows progress for each tool, (3) executes tools and appends the results, (4) stops when there are no tool calls, (5) maxes out at 3 iterations. Use 3 tools: search_web, get_weather, calculate. Prompt: "What's the weather in Lima? Search for news about Peru and calculate 2**10".

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': 3 articles found"

@tool
def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"22°C, partly cloudy in {city}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

tools = [search_web, get_weather, calculate]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)
messages = [HumanMessage(content="What's the weather in Lima? Search for news about Peru and calculate 2**10")]

for iteration in range(1, 4):
    print(f"\n=== Iteration {iteration} ===")
    full_message = None
    seen = set()

    for chunk in model_with_tools.stream(messages):
        full_message = chunk if full_message is None else full_message + chunk
        for tc_chunk in chunk.tool_call_chunks:
            if tc_chunk["name"] and tc_chunk["name"] not in seen:
                seen.add(tc_chunk["name"])
                print(f"  → {tc_chunk['name']}...", flush=True)
        if chunk.content:
            print(chunk.content, end="", flush=True)

    messages.append(full_message)
    if not full_message.tool_calls:
        print("\nAgent finished.")
        break
    for tc in full_message.tool_calls:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        print(f"  ✓ {tc['name']} → {result}")
        messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

Exercise 4: LangGraph streaming with stream_mode="messages" (Medium)

Use create_react_agent with 2 tools and stream_mode="messages". Filter to show only: (a) tool names when you detect a tool_call_chunk with a name, and (b) tokens of the final answer (when metadata["langgraph_node"] == "agent" and there's content).

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"26°C and sunny in {city}"

@tool
def translate(text: str, target_language: str) -> str:
    """Translate a text into the given language."""
    return f"[{target_language}] {text}"

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(model, [get_weather, translate])

is_streaming = False
for chunk, metadata in agent.stream(
    {"messages": [("user", "Weather in Bogota and translate 'it's hot' into Spanish")]},
    stream_mode="messages",
):
    if hasattr(chunk, "tool_call_chunks") and chunk.tool_call_chunks:
        for tc in chunk.tool_call_chunks:
            if tc["name"]:
                print(f"→ Tool: {tc['name']}", flush=True)
    elif chunk.content and metadata.get("langgraph_node") == "agent":
        if not is_streaming:
            is_streaming = True
            print("\nAnswer: ", end="")
        print(chunk.content, end="", flush=True)

Exercise 5: Async streaming with astream_events for SSE (Hard)

Implement async def generate_sse_events(user_message) that uses agent.astream_events() and generates SSE: {"type": "thinking"} at the start, {"type": "tool_call", "name": "..."} when a tool starts, {"type": "tool_result", "name": "...", "content": "..."} when it finishes, {"type": "token", "content": "..."} per token, and {"type": "done"} at the end. Each yield: f"data: {json.dumps(...)}\n\n".

import asyncio, json
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"24°C and sunny in {city}"

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(model, [get_weather])

async def generate_sse_events(user_message: str):
    yield f"data: {json.dumps({'type': 'thinking'})}\n\n"
    async for event in agent.astream_events(
        {"messages": [("user", user_message)]}, version="v2",
    ):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.tool_call_chunks:
                for tc in chunk.tool_call_chunks:
                    if tc["name"]:
                        yield f"data: {json.dumps({'type': 'tool_call', 'name': tc['name']})}\n\n"
            elif chunk.content:
                yield f"data: {json.dumps({'type': 'token', 'content': chunk.content})}\n\n"
        elif kind == "on_tool_end":
            output = event["data"]["output"]
            content = output.content if hasattr(output, "content") else str(output)
            yield f"data: {json.dumps({'type': 'tool_result', 'name': event['name'], 'content': content})}\n\n"
    yield f"data: {json.dumps({'type': 'done'})}\n\n"

async def main():
    async for sse in generate_sse_events("Weather in Mexico City"):
        data = json.loads(sse.replace("data: ", "").strip())
        if data["type"] == "token":
            print(data["content"], end="", flush=True)
        else:
            print(f"\n[{data['type']}] {data.get('name', '')}", flush=True)

asyncio.run(main())

Summary

In this capsule you learned:

• Streaming is critical for production UX — without it, any operation longer than 2 seconds makes the user think the app froze. It's not a nice-to-have, it's a requirement
• model.stream() returns AIMessageChunk with text fragments or tool_call_chunks — never both at the same time
• tool_call_chunks contain JSON fragments: the name and id arrive in the first chunk, the args arrive fragmented across later chunks
• Accumulating with + turns fragmented chunks into a complete message with tool_calls ready to execute — never try to parse tool_call_chunks directly
• The agent loop with streaming follows the same pattern as always (LLM → tools → LLM), but every iteration shows progress in real time
• LangGraph streaming offers three modes: "updates" (per node), "values" (full state), "messages" (token by token) — pick based on your use case
• astream_events is the most granular: you see tool start/end, individual tokens, and everything happening internally — ideal for FastAPI and production UIs
• General rule: if a human is waiting → streaming. If it's batch processing → invoke

Next capsule: Tool Composition and Chaining — tools that call other tools, tool pipelines, and how to compose complex capabilities without circular dependencies.

Additional Resources

1. LangChain — Streaming — Streaming concepts in LangChain with examples
2. LangChain — Stream Tool Calls — Specific guide to streaming with tool calls
3. LangGraph — Streaming — Every streaming mode in LangGraph
4. LangGraph — astream_events — Async events from inside tools
5. Server-Sent Events (MDN) — SSE reference for streaming to the frontend
6. FastAPI — StreamingResponse — Streaming response documentation in FastAPI