Real-Time Streaming Agents

Real-Time Streaming Agents

import anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio, json

client = anthropic.Anthropic()
app    = FastAPI()

@app.post("/agent/stream")
async def stream_agent(goal: str):
    async def generate():
        messages = [{"role":"user","content":goal}]

        for turn in range(10):
            # Stream LLM response token by token
            with client.messages.stream(
                model="claude-opus-4-5", max_tokens=1024,
                tools=tools, messages=messages
            ) as stream:
                full_response = []
                async for event in stream:
                    if hasattr(event,"delta") and hasattr(event.delta,"text"):
                        token = event.delta.text
                        full_response.append(token)
                        yield f"data: {json.dumps({'type':'token','text':token})}

"

            final = stream.get_final_message()
            if final.stop_reason == "end_turn": break

            # Execute tools
            for block in final.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    yield f"data: {json.dumps({'type':'tool','name':block.name,'result':result[:200]})}

"

    return StreamingResponse(generate(), media_type="text/event-stream")