Real-time AI Agent Communication with SSE

Why SSE for AI Agents?#

Server-Sent Events (SSE) is a web standard that enables servers to push real-time updates to clients over a single HTTP connection. For MCP-compatible AI agents, SSE provides a critical capability: streaming responses. When an AI agent invokes a tool that takes time (a complex analysis, a database query, a multi-step workflow), SSE lets you stream progress updates and partial results instead of making the agent wait for a complete response. This dramatically improves the user experience because the AI can relay progress in real-time. SSE is simpler than WebSockets, works through firewalls and proxies, and is the standard transport for MCP real-time communication.

Implementing an SSE Endpoint#

The SSE endpoint should be available at /api/v1/mcp/sse (or your preferred path, declared in mcp.json). The implementation uses standard HTTP with the text/event-stream content type.

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

app = FastAPI()

async def event_generator(request: Request, tool_name: str, params: dict):
    """Generate SSE events for a tool invocation."""
    # Send initial acknowledgment
    yield f"data: {json.dumps({'type': 'started', 'tool': tool_name})}\n\n"

    # Execute tool and stream progress
    async for update in execute_tool_streaming(tool_name, params):
        if await request.is_disconnected():
            break
        yield f"data: {json.dumps({'type': 'progress', 'data': update})}\n\n"
        await asyncio.sleep(0)

    # Send completion
    yield f"data: {json.dumps({'type': 'complete'})}\n\n"

@app.post("/api/v1/mcp/sse")
async def mcp_sse(request: Request):
    body = await request.json()
    return StreamingResponse(
        event_generator(request, body["tool"], body.get("params", {})),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
    )

SSE Event Format#

MCP SSE events follow a standard format with type, data, and optional metadata fields. The three main event types are: started (acknowledges the request), progress (streams partial results), and complete (signals the final result). Keep events small and frequent for the best agent experience.

Testing Your SSE Endpoint#

Test your SSE endpoint using curl or any HTTP client that supports streaming responses. Verify that events are properly formatted, the connection stays open for long-running operations, and the endpoint handles client disconnection gracefully.

1. 1Start your server with the SSE endpoint enabled.
2. 2Test with curl: curl -N -X POST http://localhost:8000/api/v1/mcp/sse -H 'Content-Type: application/json' -d '{"tool": "scan_domain", "params": {"domain": "example.com"}}'
3. 3Verify events stream in real-time (not batched at the end).
4. 4Test client disconnection by pressing Ctrl+C mid-stream.
5. 5Declare the SSE endpoint in your mcp.json manifest.