Streaming Protocol¶

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