Streaming Protocol: Real-Time Agent Responses

Overview

Streaming allows agents to send partial responses as they generate them.

Critical for responsive user interfaces.

---

Streaming Benefits

Regular API:
  Request → Wait 5 seconds → Full response

Streaming:
  Request → Get first token in 200ms
           → Next token every 50ms
           → User sees response building in real-time

---

Claude Streaming

Streaming with Claude

from anthropic import Anthropic

class StreamingAgent:
    def __init__(self):
        self.client = Anthropic()

    def stream_response(self, prompt: str):
        """Stream response to user"""

        with self.client.messages.stream(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[
                {"role": "user", "content": prompt}
            ]
        ) as stream:
            for text in stream.text_stream:
                # Send chunk to user
                print(text, end="", flush=True)

                # Can also use for real-time processing
                yield text

---

OpenAI Streaming

Streaming with GPT-4

from openai import OpenAI

class OpenAIStream:
    def __init__(self):
        self.client = OpenAI()

    def stream_response(self, messages: list):
        """Stream from OpenAI"""

        stream = self.client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            stream=True
        )

        for chunk in stream:
            if chunk.choices[0].delta.content:
                text = chunk.choices[0].delta.content
                print(text, end="", flush=True)
                yield text

---

Web UI Integration

Streaming to Browser

from fastapi import FastAPI, Response
from fastapi.responses import StreamingResponse

class StreamingWebAgent:
    app = FastAPI()

    @app.post("/stream")
    async def stream_endpoint(request: dict):
        """Endpoint that streams agent response"""

        async def event_generator():
            for chunk in self.agent.stream_response(request["prompt"]):
                yield f"data: {chunk}\n\n"

        return StreamingResponse(event_generator())

---

3 Warnings ⚠️

Warning 1: Streaming Overhead

# ❌ WRONG
# Stream every single character
# Network overhead > benefit

# ✅ RIGHT
# Stream in reasonable chunks
# Or buffer for 500ms before sending

Warning 2: Tool Calls During Stream

# ❌ WRONG
# Try to use tools while streaming
with stream:
    for text in stream.text_stream:
        if needs_tool:
            call_tool()  # Breaks streaming!

# ✅ RIGHT
# Collect full response first
response = agent.call(prompt)
if tool_use:
    call_tool()
    stream_results(tool_result)

Warning 3: Error Handling

# ❌ WRONG
# Error mid-stream, no way to tell user

for chunk in stream:
    send_to_user(chunk)
    # Error occurs here, user sees garbage

# ✅ RIGHT
# Send error marker
try:
    for chunk in stream:
        send_to_user(chunk)
except Exception as e:
    send_to_user("[ERROR: " + str(e) + "]")

---

Last Updated: August 9, 2026