04_https_streamable.py

# MCP with Streamable HTTP - Complete Guide
# This demonstrates how MCP works with Streamable HTTP for real-time communication

"""
MCP STREAMABLE HTTP TRANSPORT EXPLAINED
=======================================

Streamable HTTP is the new unified transport mechanism in MCP, replacing SSE.
It provides a modern, flexible transport layer that supports both batch responses and streaming.

Streamable HTTP Transport Architecture:
--------------------------------------
┌─────────────┐                    ┌─────────────┐
│   Client    │ ──POST /mcp──────► │   Server    │
│  (Claude,   │                    │             │
│   Cursor)   │ ◄──JSON Response── │ (Your MCP   │
└─────────────┘                    │  Server)    │
                                   └─────────────┘

• Single endpoint: /mcp handles all communication
• Bi-directional: POST requests, JSON responses
• Can support streaming responses for real-time data

Streamable HTTP provides:
• Unified endpoint for all MCP communication (tools, resources, prompts, logging)
• Supports both request/response and streaming paradigms
• Simpler session management (handled by the protocol)
• Recommended for web deployments and remote servers
• Requires CORS handling

Key Differences from SSE:
• Single endpoint for all communication
• No separate /sse and /messages endpoints
• More streamlined and efficient
"""

import asyncio
import json
import uuid
from datetime import datetime
from typing import Dict, List, Optional

# FastAPI implementation for Streamable HTTP MCP Server
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

# MCP Server implementation
from mcp.server.fastmcp import FastMCP

# Global state for demo purposes (use proper storage in production)
active_sessions: Dict[str, dict] = {}
monitoring_tasks: Dict[str, asyncio.Task] = {}

# Create FastMCP server instance
mcp = FastMCP("Streamable-HTTP-Demo-Server")


@mcp.tool()
def get_server_time() -> str:
    """Get current server time - demonstrates simple tool"""
    return f"Server time: {datetime.now().isoformat()}"


@mcp.tool()
def start_monitoring(session_id: str, interval: int = 5) -> str:
    """Start real-time system monitoring - demonstrates streaming updates"""
    if session_id in monitoring_tasks:
        return "Monitoring already active for this session"

    # This would start a background task that sends periodic updates
    # via streaming to the client
    active_sessions[session_id] = {
        "monitoring": True,
        "interval": interval,
        "started_at": datetime.now().isoformat(),
    }

    return f"Started monitoring with {interval}s intervals"


@mcp.tool()
def stop_monitoring(session_id: str) -> str:
    """Stop real-time monitoring"""
    if session_id in active_sessions:
        active_sessions[session_id]["monitoring"] = False
        return "Monitoring stopped"
    return "No active monitoring found"


@mcp.resource("session://{session_id}")
def get_session_info(session_id: str) -> str:
    """Get information about the current session"""
    session = active_sessions.get(session_id, {})
    return json.dumps(
        {
            "session_id": session_id,
            "active": session_id in active_sessions,
            "monitoring": session.get("monitoring", False),
            "created_at": session.get("started_at", "unknown"),
        },
        indent=2,
    )


# FastAPI app for HTTP/Streamable transport
app = FastAPI(title="MCP Streamable HTTP Demo Server")

# Enable CORS for browser clients
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # In production, specify allowed origins
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
async def root():
    """Server info endpoint"""
    return {
        "name": "MCP Streamable HTTP Demo Server",
        "transport": "Streamable HTTP",
        "active_sessions": len(active_sessions),
        "endpoints": {"mcp": "/mcp"},
    }


# MCP endpoint - handle all MCP JSON-RPC messages
@app.post("/mcp")
async def mcp_endpoint(request: Request):
    """MCP endpoint - handles all MCP JSON-RPC messages"""
    try:
        data = await request.json()
        method = data.get("method")
        request_id = data.get("id")
        params = data.get("params", {})

        if method == "initialize":
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "protocolVersion": "2024-11-05",
                    "capabilities": {"tools": {}, "resources": {}, "prompts": {}},
                    "serverInfo": {
                        "name": "Streamable HTTP Demo Server",
                        "version": "1.0.0",
                    },
                },
            }

        elif method == "tools/list":
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "tools": [
                        {
                            "name": "get_server_time",
                            "description": "Get current server time",
                            "inputSchema": {"type": "object", "properties": {}},
                        },
                        {
                            "name": "start_monitoring",
                            "description": "Start real-time system monitoring",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "session_id": {"type": "string"},
                                    "interval": {"type": "integer", "default": 5},
                                },
                                "required": ["session_id"],
                            },
                        },
                        {
                            "name": "stop_monitoring",
                            "description": "Stop real-time monitoring",
                            "inputSchema": {
                                "type": "object",
                                "properties": {"session_id": {"type": "string"}},
                                "required": ["session_id"],
                            },
                        },
                    ]
                },
            }

        elif method == "tools/call":
            tool_name = params.get("name")
            arguments = params.get("arguments", {})

            try:
                if tool_name == "get_server_time":
                    result = get_server_time()
                elif tool_name == "start_monitoring":
                    # Generate a session ID if not provided
                    session_id = arguments.get("session_id", str(uuid.uuid4()))
                    interval = arguments.get("interval", 5)
                    result = start_monitoring(session_id, interval)
                elif tool_name == "stop_monitoring":
                    session_id = arguments.get("session_id", "default")
                    result = stop_monitoring(session_id)
                else:
                    raise ValueError(f"Unknown tool: {tool_name}")

                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "result": {"content": [{"type": "text", "text": result}]},
                }
            except Exception as e:
                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "error": {"code": -32000, "message": f"Tool execution error: {e}"},
                }

        elif method == "resources/list":
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "resources": [
                        {
                            "uri": "session://demo",
                            "name": "Session Information",
                            "description": "Current session details and status",
                            "mimeType": "application/json",
                        }
                    ]
                },
            }

        elif method == "resources/read":
            uri = params.get("uri")
            try:
                if uri and uri.startswith("session://"):
                    session_id = uri.split("://")[1] or "demo"
                    content = get_session_info(session_id)
                else:
                    raise ValueError(f"Unknown resource: {uri}")

                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "result": {
                        "contents": [
                            {
                                "uri": uri,
                                "mimeType": "application/json",
                                "text": content,
                            }
                        ]
                    },
                }
            except Exception as e:
                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "error": {"code": -32000, "message": f"Resource read error: {e}"},
                }

        else:
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {"code": -32601, "message": f"Method not found: {method}"},
            }

    except Exception as e:
        return {
            "jsonrpc": "2.0",
            "id": None,
            "error": {"code": -32700, "message": f"Parse error: {e}"},
        }


@app.get("/status")
async def status():
    """Get server status and active sessions"""
    return {
        "server": "MCP Streamable HTTP Demo",
        "status": "running",
        "active_sessions": len(active_sessions),
        "sessions": {
            session_id: {
                "created_at": session["created_at"],
                "monitoring": session.get("monitoring", False),
            }
            for session_id, session in active_sessions.items()
        },
    }


# Example client code for connecting to Streamable HTTP MCP server
async def demo_streamable_http_client():
    """
    Example of how to connect to an MCP Streamable HTTP server
    This demonstrates the client-side interaction
    """
    import aiohttp
    import json

    server_url = "http://localhost:8000"

    print("🔌 Connecting to MCP Streamable HTTP Server...")

    async with aiohttp.ClientSession() as session:
        # Step 1: Initialize MCP connection
        init_request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",  # Use the latest protocol version for Streamable HTTP
                "capabilities": {"roots": {"listChanged": True}},
                "clientInfo": {"name": "Demo Client", "version": "1.0.0"},
            },
        }

        async with session.post(f"{server_url}/mcp", json=init_request) as response:
            result = await response.json()
            print(f"🤝 Initialize response: {result}")

            # Step 2: List tools
            list_tools_request = {
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/list",
                "params": {},
            }
            async with session.post(
                f"{server_url}/mcp", json=list_tools_request
            ) as response:
                tools_result = await response.json()
                print(f"🛠️ Tools list response: {tools_result}")

            # Step 3: Call a tool (get_server_time)
            call_tool_request = {
                "jsonrpc": "2.0",
                "id": 3,
                "method": "tools/call",
                "params": {"name": "get_server_time", "arguments": {}},
            }
            async with session.post(
                f"{server_url}/mcp", json=call_tool_request
            ) as response:
                tool_call_result = await response.json()
                print(f"⏰ Get server time response: {tool_call_result}")

            # Step 4: Start monitoring (demonstrates streaming)
            # Note: For actual streaming, the client would need to handle a continuous response.
            # This example shows the initial call.
            start_monitoring_request = {
                "jsonrpc": "2.0",
                "id": 4,
                "method": "tools/call",
                "params": {
                    "name": "start_monitoring",
                    "arguments": {
                        "session_id": "demo-session-123",  # In a real scenario, this would be managed by the client
                        "interval": 2,
                    },
                },
            }
            async with session.post(
                f"{server_url}/mcp", json=start_monitoring_request
            ) as response:
                monitoring_result = await response.json()
                print(f"📊 Start monitoring response: {monitoring_result}")

            # Step 5: Read a resource (session info)
            read_resource_request = {
                "jsonrpc": "2.0",
                "id": 5,
                "method": "resources/read",
                "params": {"uri": "session://demo-session-123"},
            }
            async with session.post(
                f"{server_url}/mcp", json=read_resource_request
            ) as response:
                resource_read_result = await response.json()
                print(f"ℹ️ Resource read response: {resource_read_result}")


if __name__ == "__main__":
    # To run this server: uvicorn 04_https_streamable:app --reload
    # To run the client demo: python -c "import asyncio; from 04_https_streamable import demo_streamable_http_client; asyncio.run(demo_streamable_http_client())"
    print("To run the server, use: uvicorn 04_https_streamable:app --reload")
    print(
        'To run the client demo, use: python -c "import asyncio; from 04_https_streamable import demo_streamable_http_client; asyncio.run(demo_streamable_http_client())"'
    )