hypercorn_server.py

# hypercorn_server.py - MCP Server with Hypercorn (HTTP/2)
"""
Hypercorn-based MCP server for ASGI performance benchmarking.
Optimized for HTTP/2 multiplexed agent communication patterns.
"""

import asyncio
from hypercorn.config import Config
from hypercorn.asyncio import serve
from mcp.server.fastmcp import FastMCP
import mcp.types as types
import time
import json

# Create MCP server optimized for HTTP/2 agent communication
mcp_app = FastMCP(
    name="hypercorn-mcp-benchmark",
    description="MCP server running on Hypercorn for HTTP/2 performance testing",
    stateless_http=True,      # Essential for benchmarking
    debug=False,              # Production mode
    log_level="WARNING"       # Minimal logging for performance
)

@mcp_app.tool()
def agent_task(task_id: str, data: str, complexity: str = "simple") -> dict:
    """Simulate typical agent task processing with varying complexity."""
    start_time = time.time()
    
    # Simulate different processing complexities
    if complexity == "simple":
        result = f"Processed {len(data)} bytes"
    elif complexity == "medium":
        # Simulate some CPU work
        result = f"Analyzed {len(data)} bytes with {len(data.split())} tokens"
    else:  # complex
        # Simulate more intensive processing
        result = {
            "analysis": f"Deep analysis of {len(data)} bytes",
            "tokens": len(data.split()),
            "patterns": ["pattern1", "pattern2"] if len(data) > 50 else ["pattern1"],
            "confidence": 0.95
        }
    
    processing_time = time.time() - start_time
    
    return {
        "task_id": task_id,
        "result": result,
        "processing_time_ms": round(processing_time * 1000, 2),
        "timestamp": time.time(),
        "status": "complete",
        "server": "hypercorn"
    }

@mcp_app.tool()
def batch_process(tasks: list) -> dict:
    """Process multiple tasks in a batch - common agent pattern."""
    start_time = time.time()
    results = []
    
    for i, task in enumerate(tasks):
        results.append({
            "task_index": i,
            "data_length": len(str(task)),
            "processed": True
        })
    
    processing_time = time.time() - start_time
    
    return {
        "batch_id": f"batch_{int(time.time())}",
        "total_tasks": len(tasks),
        "results": results,
        "processing_time_ms": round(processing_time * 1000, 2),
        "server": "hypercorn"
    }

@mcp_app.tool()
def parallel_agent_tasks(task_configs: list) -> dict:
    """Process parallel agent tasks - showcases HTTP/2 multiplexing."""
    start_time = time.time()
    results = []
    
    for i, config in enumerate(task_configs):
        # Simulate processing different task types
        task_type = config.get("type", "unknown")
        data = config.get("data", "")
        
        results.append({
            "task_index": i,
            "task_type": task_type,
            "data_processed": len(str(data)),
            "status": "completed"
        })
    
    processing_time = time.time() - start_time
    
    return {
        "batch_id": f"parallel_{int(time.time())}",
        "total_tasks": len(task_configs),
        "results": results,
        "processing_time_ms": round(processing_time * 1000, 2),
        "server": "hypercorn"
    }

@mcp_app.resource("agent://{agent_id}/status")
async def agent_status(agent_id: str) -> dict:
    """Agent status check - enhanced for HTTP/2."""
    return {
        "agent_id": agent_id,
        "status": "active",
        "last_seen": time.time(),
        "server": "hypercorn",
        "protocol": "HTTP/2",
        "capabilities": [
            "tool_calling",
            "resource_access", 
            "batch_processing",
            "parallel_multiplexing",
            "header_compression"
        ],
        "http2_features": {
            "multiplexing": True,
            "server_push": True,
            "header_compression": True
        }
    }

@mcp_app.resource("benchmark://{test_type}/data")
async def benchmark_data(test_type: str) -> dict:
    """Provide test data optimized for HTTP/2 testing."""
    test_data = {
        "small": "x" * 100,           # 100 bytes
        "medium": "x" * 1000,         # 1KB  
        "large": "x" * 10000,         # 10KB
        "xlarge": "x" * 100000,       # 100KB - HTTP/2 compression test
    }
    
    return {
        "test_type": test_type,
        "data": test_data.get(test_type, test_data["small"]),
        "size_bytes": len(test_data.get(test_type, test_data["small"])),
        "server": "hypercorn",
        "compression_enabled": True
    }

@mcp_app.resource("stream://{stream_id}/data")
async def stream_data(stream_id: str) -> dict:
    """Simulate HTTP/2 stream-specific data."""
    return {
        "stream_id": stream_id,
        "data": f"Stream {stream_id} data payload",
        "timestamp": time.time(),
        "server": "hypercorn",
        "protocol": "HTTP/2"
    }

@mcp_app.prompt()
async def agent_communication(
    source_agent: str,
    target_agent: str,
    task_type: str = "data_analysis"
) -> list[types.PromptMessage]:
    """Generate structured prompts for A2A communication with HTTP/2 optimization."""
    return [
        types.PromptMessage(
            role="system",
            content=types.TextContent(
                text=f"You are {source_agent} communicating with {target_agent} over HTTP/2",
                type="text"
            )
        ),
        types.PromptMessage(
            role="user", 
            content=types.TextContent(
                text=f"Perform {task_type} using HTTP/2 multiplexing for optimal performance",
                type="text"
            )
        )
    ]

# Create the streamable HTTP app at module level (same pattern as Uvicorn)
streamable_http_app = mcp_app.streamable_http_app()

if __name__ == "__main__":
    print("🚀 Starting MCP Server with Hypercorn (HTTP/2)")
    print("=" * 50)
    print("Server: Hypercorn")
    print("Protocol: HTTP/2 + WebSockets")
    print("Optimization: Multiplexing, header compression")
    print("Target: Parallel agent communication")
    print("=" * 50)
    
    # Hypercorn configuration optimized for HTTP/2 benchmarking
    config = Config()
    config.bind = ["0.0.0.0:8001"]    # Different port to avoid conflict
    config.workers = 1                # Single worker for consistent benchmarking
    
    # HTTP/2 specific settings
    config.http2 = True
    config.alpn_protocols = ["h2", "http/1.1"]
    
    # Logging (minimal for performance)
    config.loglevel = "warning"
    config.access_log_format = None  # Disable access logs
    
    # Run with the same pattern as Uvicorn
    asyncio.run(serve(streamable_http_app, config))