MCP Router

#!/usr/bin/env python3 """ MCP Server for Cursor IDE Integration Exposes the router as an MCP server so Cursor can use it for intelligent model routing. """ import asyncio import json import os import sys from pathlib import Path from typing import Any, Optional, Sequence # Add parent directory to path sys.path.insert(0, str(Path(__file__).parent.parent)) try: from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent HAS_MCP = True except ImportError: # Fallback if MCP SDK structure is different try: from mcp import Server, types from mcp.server.stdio import stdio_server HAS_MCP = True except ImportError: HAS_MCP = False print("Warning: MCP SDK not installed. Install with: pip install mcp") from src.router import MCPRouter, QueryContext, TaskType, Complexity # Note: We don't need MCPRouterClient since Cursor will handle API calls class CursorMCPRouter: """MCP Server wrapper for the router.""" def __init__(self): """Initialize the MCP router server.""" if not HAS_MCP: raise ImportError( "MCP SDK required. Install with: pip install mcp" ) # Router doesn't need API keys - it just recommends models self.router = MCPRouter() # Initialize server try: self.server = Server("mcp-router") except: # Fallback initialization self.server = Server() # Register handlers self._register_handlers() def _register_handlers(self): """Register MCP handlers.""" @self.server.list_tools() async def list_tools() -> list[Tool]: """List available tools.""" return [ Tool( name="route_query", description="Route a query to the best model based on query characteristics and routing strategy", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The query/prompt to route" }, "strategy": { "type": "string", "enum": ["balanced", "cost", "speed", "quality"], "description": "Routing strategy", "default": "balanced" }, "system_prompt": { "type": "string", "description": "Optional system prompt for context" } }, "required": ["query"] } ), Tool( name="get_model_recommendation", description="Get model recommendation for Cursor to use (Cursor handles API calls with its own keys)", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The query/prompt to route" }, "strategy": { "type": "string", "enum": ["balanced", "cost", "speed", "quality"], "description": "Routing strategy", "default": "balanced" } }, "required": ["query"] } ), Tool( name="list_models", description="List all available models in the router", inputSchema={ "type": "object", "properties": {} } ), Tool( name="get_routing_stats", description="Get statistics about routing decisions", inputSchema={ "type": "object", "properties": {} } ), Tool( name="analyze_query", description="Analyze a query to determine its characteristics", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The query to analyze" } }, "required": ["query"] } ) ] @self.server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Handle tool calls.""" try: if name == "route_query": query = arguments.get("query", "") strategy = arguments.get("strategy", "balanced") # Analyze query first to get context context = self.router.analyzer.analyze(query) decision = self.router.route(query, context=context, strategy=strategy) result = { "selected_model": { "name": decision.selected_model.name, "model_id": decision.selected_model.model_id, "provider": decision.selected_model.provider }, "confidence": decision.confidence, "reasoning": decision.reasoning, "estimated_cost": decision.estimated_cost, "estimated_latency_ms": decision.estimated_latency_ms, "alternatives": [ { "name": alt.name, "model_id": alt.model_id, "provider": alt.provider } for alt in decision.alternatives[:3] ] } return [TextContent( type="text", text=json.dumps(result, indent=2) )] elif name == "get_model_recommendation": query = arguments.get("query", "") strategy = arguments.get("strategy", "balanced") # Analyze query first to get context context = self.router.analyzer.analyze(query) # Route query to get best model decision = self.router.route(query, context=context, strategy=strategy) # Return model recommendation for Cursor to use # Cursor will handle the actual API call with its own keys result = { "recommended_model": { "model_id": decision.selected_model.model_id, "provider": decision.selected_model.provider, "name": decision.selected_model.name }, "confidence": decision.confidence, "reasoning": decision.reasoning, "estimated_cost": decision.estimated_cost, "estimated_latency_ms": decision.estimated_latency_ms, "alternatives": [ { "model_id": alt.model_id, "provider": alt.provider, "name": alt.name } for alt in decision.alternatives[:3] ], "query_analysis": { "task_type": context.task_type.value if context.task_type else None, "complexity": context.complexity.value if context.complexity else None, "estimated_tokens": context.estimated_tokens } } return [TextContent( type="text", text=json.dumps(result, indent=2) )] elif name == "list_models": models = [] for model in self.router.models.values(): models.append({ "name": model.name, "model_id": model.model_id, "provider": model.provider, "context_window": model.context_window, "cost_per_1k_input": model.cost_per_1k_tokens_input, "cost_per_1k_output": model.cost_per_1k_tokens_output, "avg_latency_ms": model.avg_latency_ms, "reasoning_quality": model.reasoning_quality, "code_quality": model.code_quality }) return [TextContent( type="text", text=json.dumps({"models": models}, indent=2) )] elif name == "get_routing_stats": stats = self.router.get_routing_stats() return [TextContent( type="text", text=json.dumps(stats, indent=2) )] elif name == "analyze_query": query = arguments.get("query", "") context = self.router.analyzer.analyze(query) return [TextContent( type="text", text=json.dumps({ "query": query, "task_type": context.task_type.value if context.task_type else None, "complexity": context.complexity.value if context.complexity else None, "estimated_tokens": context.estimated_tokens, "requires_streaming": context.requires_streaming, "requires_multimodal": context.requires_multimodal, "requires_embeddings": context.requires_embeddings }, indent=2) )] else: return [TextContent( type="text", text=json.dumps({"error": f"Unknown tool: {name}"}) )] except Exception as e: return [TextContent( type="text", text=json.dumps({"error": str(e)}) )] async def run(self): """Run the MCP server.""" async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, self.server.create_initialization_options() ) async def main(): """Main entry point.""" if not HAS_MCP: print("Error: MCP SDK not installed.", file=sys.stderr) print("Install with: pip install mcp", file=sys.stderr) sys.exit(1) try: router_server = CursorMCPRouter() await router_server.run() except Exception as e: print(f"Error starting MCP server: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": asyncio.run(main())