03_sse_server_complete.py

# MCP with Server-Sent Events (SSE) - Basic Structure
# This demonstrates the endpoint structure for SSE-based MCP servers

"""
MCP SSE TRANSPORT EXPLAINED
===========================

Server-Sent Events (SSE) is one of three transport mechanisms in MCP:
1. STDIO - Process communication (local servers)
2. SSE - HTTP-based communication (remote servers) - **LEGACY**
3. Streamable HTTP - New unified transport (replacing SSE) - **RECOMMENDED**

SSE Transport Architecture:
--------------------------
┌─────────────┐    GET /sse     ┌─────────────┐
│   Client    │ ──────────────► │   Server    │
│  (Claude,   │                 │             │
│   Cursor)   │ ◄────────────── │ (Your MCP   │
└─────────────┘ POST /messages  │  Server)    │
                                └─────────────┘

SSE provides:
• Persistent connection from client to server
• Server can push updates/responses in real-time
• Bi-directional communication (SSE + HTTP POST)
• Session management with unique session IDs
• Automatic reconnection on connection drops

Key Differences from STDIO:
• Network-based (can be remote)
• Stateless HTTP foundation
• More complex session management
• Better for distributed systems
• Requires CORS handling
"""

# PSEUDOCODE - Basic SSE MCP Server Structure
# This shows the essential endpoints and flow

# === SERVER SETUP ===
app = create_web_server()
app.enable_cors()  # Required for browser clients
active_sessions = {}  # Track client sessions


# === ENDPOINT 1: SSE Connection ===
@app.get("/sse")
def establish_sse_connection(request):
    """
    Establishes persistent Server-Sent Events connection

    Flow:
    1. Generate unique session_id
    2. Send 'endpoint' event with message URL
    3. Keep connection open for real-time updates
    """
    session_id = generate_uuid()
    active_sessions[session_id] = create_session_state()

    def event_stream():
        # Tell client where to send messages
        yield "event: endpoint\n"
        yield f"data: /messages?session_id={session_id}\n\n"

        # Send periodic pings and updates
        while client_connected():
            yield "event: ping\n"
            yield f'data: {{"timestamp": "{now()}", "session": "{session_id}"}}\n\n'
            sleep(5)

    return streaming_response(event_stream(), "text/event-stream")


# === ENDPOINT 2: Message Handler ===
@app.post("/messages")
def handle_mcp_messages(request):
    """
    Processes all MCP JSON-RPC messages

    Handles:
    - initialize: MCP handshake
    - tools/list: Available tools
    - tools/call: Execute tool
    - resources/list: Available resources
    - resources/read: Read resource content
    """
    session_id = get_query_param("session_id")
    if not valid_session(session_id):
        return error_response("Invalid session")

    message = parse_json_rpc(request.body)
    method = message.method

    if method == "initialize":
        return {
            "jsonrpc": "2.0",
            "id": message.id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}, "resources": {}},
                "serverInfo": {"name": "SSE Demo Server"},
            },
        }

    elif method == "tools/list":
        return {
            "jsonrpc": "2.0",
            "id": message.id,
            "result": {
                "tools": [
                    {"name": "get_time", "description": "Get server time"},
                    {"name": "monitor", "description": "Start monitoring"},
                ]
            },
        }

    elif method == "tools/call":
        tool_name = message.params.name
        result = execute_tool(tool_name, message.params.arguments)
        return {
            "jsonrpc": "2.0",
            "id": message.id,
            "result": {"content": [{"type": "text", "text": result}]},
        }

    # ... handle other methods ...


# === ENDPOINT 3: Server Info ===
@app.get("/")
def server_info():
    return {
        "name": "MCP SSE Demo Server",
        "transport": "Server-Sent Events",
        "endpoints": {"sse": "/sse", "messages": "/messages"},
        "active_sessions": len(active_sessions),
    }


# === CLIENT EXAMPLE ===
def connect_to_sse_mcp_server():
    """
    Pseudocode: How to connect to SSE MCP server
    """
    # 1. Open SSE stream
    sse_connection = http_get("http://server:8000/sse")

    # 2. Parse SSE events for endpoint info
    for event in sse_connection.events():
        if event.type == "endpoint":
            messages_url = f"http://server:8000{event.data}"
            break

    # 3. Send MCP initialization
    response = http_post(
        messages_url,
        {
            "jsonrpc": "2.0",
            "method": "initialize",
            "params": {"protocolVersion": "2024-11-05"},
        },
    )

    # 4. Use MCP protocol normally
    tools = http_post(messages_url, {"jsonrpc": "2.0", "method": "tools/list"})


# === KEY POINTS ===
"""
SSE MCP Server Requirements:
1. Two endpoints: /sse (GET) and /messages (POST)
2. Session management with unique IDs
3. JSON-RPC 2.0 message handling
4. CORS support for browser clients
5. Persistent SSE connection for real-time updates

To run: python sse_server.py
To connect: Use MCP Inspector or custom client

⚠️  **IMPORTANT**: SSE transport is now considered legacy. For new projects, use Streamable HTTP (see 04_https_streamable.py) which provides a simpler, more efficient transport mechanism.

This file demonstrates SSE concepts for educational purposes only.
"""

if __name__ == "__main__":
    print(__doc__)
    print("\n📋 This is pseudocode showing SSE MCP server structure")
    print("🔧 For a working server, see: 04_https_streamable.py")
    print("\n" + "=" * 60)