Agent Skills: Streaming Patterns

Streaming Patterns

Server-sent events, WebSocket connections, LLM token streaming, real-time broadcasting, and streaming error handling in FastAPI.

Server-Sent Events (SSE)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def event_generator():
    for i in range(5):
        yield f"data: Message {i}\n\n"
        await asyncio.sleep(1)

@app.get("/stream")
async def stream():
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Frontend SSE Client

const eventSource = new EventSource('/stream')
eventSource.onmessage = (e) => console.log(e.data)
eventSource.onerror = () => eventSource.close()

WebSocket

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    while True:
        data = await ws.receive_text()
        await ws.send_text(f"Echo: {data}")

Frontend WebSocket Client

const ws = new WebSocket('ws://localhost:8000/ws')
ws.onmessage = (e) => console.log(e.data)
ws.send('Hello')

LLM Token Streaming

async def stream_llm(prompt: str):
    async for chunk in llm.astream(prompt):
        yield f"data: {chunk.content}\n\n"

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    return StreamingResponse(
        stream_llm(request.prompt),
        media_type="text/event-stream"
    )

Redis Pub/Sub Broadcasting

import redis.asyncio as redis

redis_client = redis.Redis()

async def broadcast(channel: str, message: str):
    await redis_client.publish(channel, message)

async def subscribe(channel: str):
    pubsub = redis_client.pubsub()
    await pubsub.subscribe(channel)
    async for message in pubsub.listen():
        yield f"data: {message['data'].decode()}\n\n"

Backpressure & Throughput

from asyncio import Queue

class BackpressureQueue:
    def __init__(self, maxsize: int = 100):
        self.queue = Queue(maxsize=maxsize)

    async def put(self, item):
        if self.queue.full():
            await self.queue.get()  # Drop oldest
        await self.queue.put(item)

    async def get(self):
        return await self.queue.get()

Streaming Error Handling

async def safe_stream():
    try:
        async for chunk in generator():
            yield f"data: {chunk}\n\n"
    except Exception as e:
        yield f"data: [ERROR] {str(e)}\n\n"
    finally:
        yield "data: [DONE]\n\n"

Monitoring

• Track tokens/second for LLM streams
• Monitor active WebSocket connections
• Log SSE disconnects for debugging
• Set connection timeouts (default: 30s)

Best Practices

• Always handle disconnects gracefully
• Set appropriate timeouts
• Use backpressure queues for high-throughput
• Send [DONE] signal when stream ends
• Log stream errors for observability