Phase 2: Tool Design & Function Calling · 50 min · Python · asyncio · FastAPI

Streaming Tool Outputs

Streaming is the difference between a demo and a product. Users see progress — they don't watch spinners.

What you will learn

Implement Server-Sent Events (SSE) to stream tool progress to a frontend
Use async generators to yield tool results as they complete
Design a progress event schema for multi-tool agent runs

The Problem

A user asks your research agent to summarize 10 documents. The agent calls a summarize_document tool 10 times. Without streaming, the user stares at a spinner for 45 seconds. With streaming, they see results appear one by one — "Document 1: done. Document 2: done..." — and the experience feels responsive even though the total time is the same.

UX research consistently shows users tolerate longer waits when they see progress. A 30-second wait with visible progress feels shorter and more trustworthy than a 30-second wait with nothing but a spinner. For agents that run 5-20 tool calls, streaming individual results is the difference between a usable product and one users abandon.

Async generators for tool streaming

asyncio.as_completed() is the right primitive for streaming tool results as they complete — not in submission order, but in completion order (whichever tool finishes first gets streamed first).

import asyncio

async def stream_tool_results(tool_calls):
    tasks = {asyncio.ensure_future(run_tool(tc)): tc for tc in tool_calls}
    
    for completed_future in asyncio.as_completed(tasks.keys()):
        result = await completed_future
        yield result  # Stream this result immediately

The key insight: asyncio.gather() waits for all tools before returning anything. asyncio.as_completed() yields each future as it completes. For a streaming UX, as_completed is almost always the right choice.

You have 5 tools running in parallel with asyncio.gather(). You want to stream each result to the UI as it completes. Which approach is correct?

asyncio.as_completed() is the correct pattern for "stream results in completion order." asyncio.gather() batches all results — nothing is returned until every tool completes. asyncio.as_completed() wraps each coroutine in a future and yields them one by one as they finish. asyncio.wait() with FIRST_COMPLETED in a loop achieves the same result but requires more code.

Server-sent events (SSE)

SSE is the right protocol for streaming agent progress from a server to a browser:

Unidirectional: server pushes events, client only listens — no bidirectional complexity
Built on HTTP/2: works over standard infrastructure, no special WebSocket upgrades
Auto-reconnect: browsers reconnect automatically if the connection drops
Native browser support: EventSource API is built into all modern browsers

WebSockets are bidirectional — great for chat applications where the client also sends messages. For agent progress (server → client only), SSE is simpler and sufficient.

SSE wire format:

data: {"type": "tool_start", "tool": "summarize_document", "id": "tc_3"}

data: {"type": "tool_complete", "tool": "summarize_document", "id": "tc_3", "duration_ms": 1200}

data: {"type": "agent_complete", "final_answer": "Here are your summaries..."}

Each event is data: <json>\n\n. The double newline terminates the event.

Progress event schema

Define a standard event format that your frontend can render:

// Agent started
{"type": "agent_start", "tool_count": 5, "timestamp": 1705123456}

// Individual tool events
{"type": "tool_start", "tool": "summarize_document", "id": "tc_1", "args": {"doc_id": "doc_001"}}
{"type": "tool_complete", "tool": "summarize_document", "id": "tc_1", "duration_ms": 1150, "result": {...}}
{"type": "tool_error", "tool": "summarize_document", "id": "tc_2", "error": "rate_limit", "duration_ms": 200}

// Final answer
{"type": "agent_complete", "final_answer": "...", "total_ms": 4500}

With this schema, a React frontend can maintain a live progress list that shows each tool's status as events arrive — without polling.

For streaming agent progress from a FastAPI backend to a React frontend, which protocol is most appropriate?

SSE is the right choice for agent progress streaming: the server pushes events and the client only listens. No bidirectional communication is needed. SSE works over standard HTTP/2, requires no special infrastructure, has built-in reconnection in all browsers, and is supported natively via EventSource. WebSockets add complexity without benefit for this use case.

Build

Build it

The code below implements the complete streaming stack: an AsyncStreamingAgent that yields progress events, an SSE serializer, and a FastAPI endpoint that serves the stream.

"""
lesson_02_05_streaming.py

Streaming agent progress with asyncio.as_completed() and SSE.

Run:
  pip install openai fastapi uvicorn
  export OPENAI_API_KEY=...
  python lesson_02_05_streaming.py
  # Then: curl -N http://localhost:8000/process
"""

import asyncio
import json
import os
import random
import time
from dataclasses import asdict, dataclass
from enum import Enum
from typing import AsyncGenerator, Optional

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

class EventType(str, Enum):
    AGENT_START = "agent_start"
    TOOL_START = "tool_start"
    TOOL_COMPLETE = "tool_complete"
    TOOL_ERROR = "tool_error"
    AGENT_COMPLETE = "agent_complete"

@dataclass
class ProgressEvent:
    type: EventType
    tool: Optional[str] = None
    id: Optional[str] = None
    args: Optional[dict] = None
    result: Optional[dict] = None
    error: Optional[str] = None
    duration_ms: Optional[float] = None
    tool_count: Optional[int] = None
    final_answer: Optional[str] = None
    total_ms: Optional[float] = None
    timestamp: float = 0.0

    def __post_init__(self):
        self.timestamp = time.time()

    def to_sse(self) -> str:
        """Format as SSE wire format: data: <json>\\n\\n"""
        data = {k: v for k, v in asdict(self).items() if v is not None}
        data["type"] = self.type.value
        return f"data: {json.dumps(data)}\n\n"

# Simulated async document processor
async def process_document(doc_id: str, call_id: str) -> tuple[str, dict]:
    delay = random.uniform(0.5, 2.5)
    await asyncio.sleep(delay)
    return call_id, {
        "doc_id": doc_id,
        "summary": f"Summary of document {doc_id} (processed in {delay:.1f}s)",
        "word_count": random.randint(50, 500),
    }

async def stream_agent_progress(doc_ids: list[str]) -> AsyncGenerator[str, None]:
    """
    Async generator that yields SSE-formatted progress events
    as each document is processed.
    """
    t_start = time.time()
    
    # Event 1: agent started
    yield ProgressEvent(
        type=EventType.AGENT_START,
        tool_count=len(doc_ids),
    ).to_sse()

    # Submit all tasks and emit tool_start events
    tasks = {}
    for i, doc_id in enumerate(doc_ids):
        call_id = f"tc_{i}"
        task = asyncio.ensure_future(process_document(doc_id, call_id))
        tasks[task] = (doc_id, call_id)
        
        yield ProgressEvent(
            type=EventType.TOOL_START,
            tool="process_document",
            id=call_id,
            args={"doc_id": doc_id},
        ).to_sse()

    # Stream results as they complete
    all_results = []
    t_tool_start = time.time()
    for completed in asyncio.as_completed(tasks.keys()):
        try:
            call_id, result = await completed
            doc_id = tasks[list(tasks.keys())[int(call_id.split("_")[1])]][0]
            duration = (time.time() - t_tool_start) * 1000
            all_results.append(result)
            
            yield ProgressEvent(
                type=EventType.TOOL_COMPLETE,
                tool="process_document",
                id=call_id,
                result=result,
                duration_ms=round(duration, 1),
            ).to_sse()
        except Exception as e:
            yield ProgressEvent(
                type=EventType.TOOL_ERROR,
                tool="process_document",
                error=str(e),
            ).to_sse()

    # Final completion event
    total_ms = (time.time() - t_start) * 1000
    yield ProgressEvent(
        type=EventType.AGENT_COMPLETE,
        final_answer=f"Processed {len(all_results)} documents successfully.",
        total_ms=round(total_ms, 1),
    ).to_sse()

# FastAPI app
app = FastAPI()

@app.get("/process")
async def process_documents():
    """Stream document processing progress via SSE."""
    doc_ids = [f"doc_{i:03d}" for i in range(5)]
    
    return StreamingResponse(
        stream_agent_progress(doc_ids),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Disable nginx buffering
        },
    )

# Terminal demo (no server needed)
async def demo_terminal():
    print("STREAMING AGENT DEMO (terminal)")
    print("=" * 50)
    doc_ids = [f"doc_{i:03d}" for i in range(5)]
    
    async for event_str in stream_agent_progress(doc_ids):
        event_data = json.loads(event_str.replace("data: ", "").strip())
        event_type = event_data.get("type")
        
        if event_type == "agent_start":
            print(f"\n[START] Processing {event_data['tool_count']} documents...")
        elif event_type == "tool_start":
            print(f"  → Starting {event_data['id']} ({event_data['args']['doc_id']})")
        elif event_type == "tool_complete":
            print(f"  ✓ Done {event_data['id']}: {event_data['result']['summary'][:50]}...")
        elif event_type == "tool_error":
            print(f"  ✗ Error {event_data.get('id', '?')}: {event_data.get('error')}")
        elif event_type == "agent_complete":
            print(f"\n[DONE] {event_data['final_answer']} ({event_data['total_ms']:.0f}ms total)")

if __name__ == "__main__":
    import sys
    if "--server" in sys.argv:
        import uvicorn
        print("Starting FastAPI server on http://localhost:8000")
        print("Test with: curl -N http://localhost:8000/process")
        uvicorn.run(app, host="0.0.0.0", port=8000)
    else:
        asyncio.run(demo_terminal())

The snippet below simulates completion-order streaming without asyncio, using precomputed "finish times" to sort results the way as_completed() would deliver them.

import random

def simulate_as_completed(tasks):
    """tasks: dict of {name: simulated_duration}. Returns names in completion order."""
    finish_times = {name: duration for name, duration in tasks.items()}
    return sorted(finish_times, key=finish_times.get)

random.seed(7)
tasks = {
    "summarize_doc_1": round(random.uniform(0.5, 2.5), 2),
    "summarize_doc_2": round(random.uniform(0.5, 2.5), 2),
    "summarize_doc_3": round(random.uniform(0.5, 2.5), 2),
    "summarize_doc_4": round(random.uniform(0.5, 2.5), 2),
}

print("Submission order:", list(tasks.keys()))
completion_order = simulate_as_completed(tasks)

for name in completion_order:
    print(f"tool_complete: {name} (took {tasks[name]}s)")

print("agent_complete: all", len(tasks), "tools finished")

Unlock the full lesson

You've read the first 2 sections. The rest of this lesson covers Check Yourself — plus a hands-on lab, quiz, and project artifact.

Create a free account — the opening phases of 29 of 35 courses are free, no credit card.