"""Main MCP Server implementation for RAVERSE""" import sys import asyncio import json import logging from pathlib import Path from typing import Any, Dict, Optional, List, Callable # Official MCP SDK imports from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource import mcp.types as types from .config import get_config, MCPServerConfig from .logging_config import setup_logging, get_logger from .database import DatabaseManager from .cache import CacheManager from .tools_binary_analysis import BinaryAnalysisTools from .tools_knowledge_base import KnowledgeBaseTools from .tools_web_analysis import WebAnalysisTools from .tools_infrastructure import InfrastructureTools from .tools_analysis_advanced import AdvancedAnalysisTools from .tools_management import ManagementTools from .tools_utilities import UtilityTools from .tools_system import SystemTools from .tools_nlp_validation import NLPValidationTools from .errors import RAVERSEMCPError from .setup_guide import print_setup_guide, is_first_time_setup from .setup_wizard import run_setup_wizard logger = get_logger(__name__) # MCP Protocol Constants MCP_VERSION = "2024-11-05" class MCPServer: """RAVERSE MCP Server implementation""" def __init__(self, config: Optional[MCPServerConfig] = None): """ Lightweight initialization. Sets up configuration and attributes to None. Does NOT establish connections or load heavy models. """ self.config = config or get_config() # Initialize all tool managers to None to prevent AttributeError self.db_manager: Optional[DatabaseManager] = None self.cache_manager: Optional[CacheManager] = None self.binary_tools: Optional[BinaryAnalysisTools] = None self.kb_tools: Optional[KnowledgeBaseTools] = None self.web_tools: Optional[WebAnalysisTools] = None self.infra_tools: Optional[InfrastructureTools] = None self.advanced_tools: Optional[AdvancedAnalysisTools] = None self.management_tools: Optional[ManagementTools] = None self.utility_tools: Optional[UtilityTools] = None self.system_tools: Optional[SystemTools] = None self.nlp_tools: Optional[NLPValidationTools] = None self._is_initialized = False def _initialize(self) -> None: """Initialize server components lazily (Heavy Lifting)""" if self._is_initialized: return try: # Check if .env file exists, if not run setup wizard env_file = Path(__file__).parent.parent / ".env" if not env_file.exists(): logger.info("No .env file found. Starting setup wizard...") run_setup_wizard() # Reload config after setup wizard completes self.config = get_config() logger.info( "Initializing RAVERSE MCP Server components", version=self.config.server_version, log_level=self.config.log_level, ) # Initialize database (if enabled) if self.config.enable_infrastructure: try: self.db_manager = DatabaseManager(self.config) self.cache_manager = CacheManager(self.config) except Exception as db_error: if is_first_time_setup(): print_setup_guide(str(db_error)) logger.error(f"Database init failed: {db_error}") # Initialize tool modules based on config if self.config.enable_binary_analysis: self.binary_tools = BinaryAnalysisTools() if self.config.enable_knowledge_base and self.db_manager and self.cache_manager: self.kb_tools = KnowledgeBaseTools(self.db_manager, self.cache_manager) if self.config.enable_web_analysis: self.web_tools = WebAnalysisTools() if self.config.enable_infrastructure and self.db_manager and self.cache_manager: self.infra_tools = InfrastructureTools(self.db_manager, self.cache_manager) # Advanced analysis self.advanced_tools = AdvancedAnalysisTools() # Management self.management_tools = ManagementTools() # Utilities self.utility_tools = UtilityTools() # System self.system_tools = SystemTools() # NLP self.nlp_tools = NLPValidationTools() self._is_initialized = True logger.info("RAVERSE MCP Server components initialized successfully") except Exception as e: logger.error(f"Server initialization failed: {str(e)}") raise async def handle_tool_call( self, tool_name: str, arguments: Dict[str, Any], ) -> Dict[str, Any]: """Handle a tool call""" # Ensure initialization happened self._initialize() try: logger.info(f"Tool call received: {tool_name}") # --- Binary Analysis Tools --- if self.binary_tools: if tool_name == "disassemble_binary": return self.binary_tools.disassemble_binary( arguments.get("binary_path"), arguments.get("architecture"), ).dict() elif tool_name == "generate_code_embedding": return self.binary_tools.generate_code_embedding( arguments.get("code_content"), arguments.get("model", "all-MiniLM-L6-v2"), ).dict() elif tool_name == "apply_patch": return self.binary_tools.apply_patch( arguments.get("binary_path"), arguments.get("patches", []), arguments.get("backup", True), ).dict() elif tool_name == "verify_patch": return self.binary_tools.verify_patch( arguments.get("original_binary"), arguments.get("patched_binary"), ).dict() # --- Knowledge Base Tools --- if self.kb_tools: if tool_name == "ingest_content": return self.kb_tools.ingest_content( arguments.get("content"), arguments.get("metadata"), ).dict() elif tool_name == "search_knowledge_base": return self.kb_tools.search_knowledge_base( arguments.get("query"), arguments.get("limit", 5), arguments.get("threshold", 0.7), ).dict() elif tool_name == "retrieve_entry": return self.kb_tools.retrieve_entry( arguments.get("entry_id"), ).dict() elif tool_name == "delete_entry": return self.kb_tools.delete_entry( arguments.get("entry_id"), ).dict() # --- Web Analysis Tools --- if self.web_tools: if tool_name == "reconnaissance": return self.web_tools.reconnaissance( arguments.get("target_url"), ).dict() elif tool_name == "analyze_javascript": return self.web_tools.analyze_javascript( arguments.get("js_code"), arguments.get("deobfuscate", True), ).dict() elif tool_name == "reverse_engineer_api": return self.web_tools.reverse_engineer_api( arguments.get("traffic_data", {}), arguments.get("js_analysis"), ).dict() elif tool_name == "analyze_wasm": return self.web_tools.analyze_wasm( arguments.get("wasm_data", b""), ).dict() elif tool_name == "security_analysis": return self.web_tools.security_analysis( arguments.get("analysis_data", {}), arguments.get("check_headers", True), arguments.get("check_cves", True), ).dict() # --- Infrastructure Tools --- if self.infra_tools: if tool_name == "database_query": return self.infra_tools.database_query( arguments.get("query"), arguments.get("params"), ).dict() elif tool_name == "cache_operation": return self.infra_tools.cache_operation( arguments.get("operation"), arguments.get("key"), arguments.get("value"), arguments.get("ttl"), ).dict() elif tool_name == "publish_message": return self.infra_tools.publish_message( arguments.get("channel"), arguments.get("message", {}), ).dict() elif tool_name == "fetch_content": return self.infra_tools.fetch_content( arguments.get("url"), arguments.get("timeout", 30), arguments.get("retries", 3), ).dict() elif tool_name == "record_metric": return self.infra_tools.record_metric( arguments.get("metric_name"), arguments.get("value"), arguments.get("labels"), ).dict() # --- Advanced Analysis Tools --- if self.advanced_tools: if tool_name == "logic_identification": return self.advanced_tools.logic_identification( arguments.get("disassembly_data", {}), arguments.get("analyze_control_flow", True), arguments.get("analyze_data_flow", True), ).dict() elif tool_name == "traffic_interception": return self.advanced_tools.traffic_interception( arguments.get("target_url"), arguments.get("ssl_intercept", True), arguments.get("capture_duration", 60), ).dict() elif tool_name == "generate_report": return self.advanced_tools.generate_report( arguments.get("analysis_results", {}), arguments.get("format", "json"), arguments.get("include_summary", True), ).dict() elif tool_name == "rag_orchestration": return self.advanced_tools.rag_orchestration( arguments.get("query"), arguments.get("context_limit", 5), arguments.get("threshold", 0.7), ).dict() elif tool_name == "deep_research": return self.advanced_tools.deep_research( arguments.get("topic"), arguments.get("max_sources", 10), arguments.get("synthesize", True), ).dict() # --- Management Tools --- if self.management_tools: if tool_name == "version_management": return self.management_tools.version_management( arguments.get("component_name"), arguments.get("version"), arguments.get("check_vulnerabilities", True), ).dict() elif tool_name == "quality_gate": return self.management_tools.quality_gate( arguments.get("analysis_results", {}), arguments.get("metrics", {}), arguments.get("threshold", 0.8), ).dict() elif tool_name == "governance_check": return self.management_tools.governance_check( arguments.get("action"), arguments.get("context", {}), arguments.get("require_approval", False), ).dict() elif tool_name == "generate_document": return self.management_tools.generate_document( arguments.get("document_type"), arguments.get("data", {}), arguments.get("format", "markdown"), ).dict() elif tool_name == "session_management": return self.management_tools.session_management( arguments.get("operation"), arguments.get("session_id"), ).dict() elif tool_name == "task_scheduler": return self.management_tools.task_scheduler( arguments.get("task_type"), arguments.get("schedule"), arguments.get("parameters"), ).dict() elif tool_name == "result_aggregation": return self.management_tools.result_aggregation( arguments.get("results"), arguments.get("aggregation_type"), ).dict() # --- Utility Tools --- if self.utility_tools: if tool_name == "url_frontier_operation" or tool_name == "url_frontier": # Fix: Default 'operation' to 'add' if missing return self.utility_tools.url_frontier_operation( arguments.get("operation", "add"), arguments.get("url"), arguments.get("priority", 5), ).dict() elif tool_name == "api_pattern_matcher": return self.utility_tools.api_pattern_matcher( arguments.get("traffic_data", {}), arguments.get("pattern_type", "rest"), ).dict() elif tool_name == "response_classifier": return self.utility_tools.response_classifier( arguments.get("response_data", {}), arguments.get("infer_schema", True), ).dict() elif tool_name == "websocket_analyzer": return self.utility_tools.websocket_analyzer( arguments.get("websocket_data", {}), arguments.get("analyze_handshake", True), ).dict() elif tool_name == "crawl_scheduler": return self.utility_tools.crawl_scheduler( arguments.get("operation"), arguments.get("job_data"), arguments.get("priority", 5), ).dict() # --- System Tools --- if self.system_tools: if tool_name == "metrics_collector": return self.system_tools.metrics_collector( arguments.get("metric_type"), arguments.get("metric_name"), arguments.get("value"), arguments.get("labels"), ).dict() elif tool_name == "multi_level_cache": return self.system_tools.multi_level_cache( arguments.get("operation"), arguments.get("key"), arguments.get("value"), arguments.get("ttl", 3600), ).dict() elif tool_name == "configuration_service": return self.system_tools.configuration_service( arguments.get("operation"), arguments.get("key"), arguments.get("value"), ).dict() elif tool_name == "llm_interface": return self.system_tools.llm_interface( arguments.get("prompt"), arguments.get("model", "gpt-4"), arguments.get("max_tokens", 2048), arguments.get("temperature", 0.7), ).dict() # --- NLP and Validation Tools --- if self.nlp_tools: if tool_name == "natural_language_interface": return self.nlp_tools.natural_language_interface( arguments.get("command"), arguments.get("context"), ).dict() elif tool_name == "poc_validation": return self.nlp_tools.poc_validation( arguments.get("vulnerability_finding", {}), arguments.get("generate_poc", True), arguments.get("execute_poc", False), ).dict() return { "success": False, "error": f"Unknown tool: {tool_name} or tool category disabled.", "error_code": "UNKNOWN_TOOL", } except RAVERSEMCPError as e: logger.error(f"Tool execution error: {str(e)}") return e.to_dict() except Exception as e: logger.error(f"Unexpected error in tool call: {str(e)}") return { "success": False, "error": f"Unexpected error: {str(e)}", "error_code": "INTERNAL_ERROR", } def get_tools_list(self) -> List[Dict[str, Any]]: """Get list of all available tools based on configuration""" tools = [] # IMPORTANT: Check config flags before adding tools to the list. # This prevents clients from seeing tools that are disabled/unsupported. # Binary Analysis Tools if self.config.enable_binary_analysis: tools.extend([ { "name": "disassemble_binary", "description": "Disassemble binary files into assembly code", "inputSchema": { "type": "object", "properties": { "binary_path": {"type": "string", "description": "Path to binary file"}, "architecture": {"type": "string", "description": "Target architecture (x86, x64, arm, etc.)"} }, "required": ["binary_path"] } }, { "name": "generate_code_embedding", "description": "Generate semantic embeddings for code", "inputSchema": { "type": "object", "properties": { "code_content": {"type": "string", "description": "Code to embed"}, "model": {"type": "string", "description": "Embedding model to use"} }, "required": ["code_content"] } }, { "name": "apply_patch", "description": "Apply patches to binary files", "inputSchema": { "type": "object", "properties": { "binary_path": {"type": "string", "description": "Path to binary"}, "patches": {"type": "array", "description": "List of patches to apply"}, "backup": {"type": "boolean", "description": "Create backup before patching"} }, "required": ["binary_path"] } }, { "name": "verify_patch", "description": "Verify patch application", "inputSchema": { "type": "object", "properties": { "original_binary": {"type": "string", "description": "Original binary path"}, "patched_binary": {"type": "string", "description": "Patched binary path"} }, "required": ["original_binary", "patched_binary"] } } ]) # Knowledge Base Tools (Requires Infrastructure) if self.config.enable_knowledge_base and self.config.enable_infrastructure: tools.extend([ { "name": "ingest_content", "description": "Ingest content into knowledge base", "inputSchema": { "type": "object", "properties": { "content": {"type": "string", "description": "Content to ingest"}, "metadata": {"type": "object", "description": "Metadata for content"} }, "required": ["content"] } }, { "name": "search_knowledge_base", "description": "Search knowledge base", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "limit": {"type": "integer", "description": "Result limit"}, "threshold": {"type": "number", "description": "Similarity threshold"} }, "required": ["query"] } }, { "name": "retrieve_entry", "description": "Retrieve knowledge base entry", "inputSchema": { "type": "object", "properties": { "entry_id": {"type": "string", "description": "Entry ID"} }, "required": ["entry_id"] } }, { "name": "delete_entry", "description": "Delete knowledge base entry", "inputSchema": { "type": "object", "properties": { "entry_id": {"type": "string", "description": "Entry ID"} }, "required": ["entry_id"] } } ]) # Web Analysis Tools if self.config.enable_web_analysis: tools.extend([ { "name": "reconnaissance", "description": "Perform web reconnaissance", "inputSchema": { "type": "object", "properties": { "target_url": {"type": "string", "description": "Target URL"} }, "required": ["target_url"] } }, { "name": "analyze_javascript", "description": "Analyze JavaScript code", "inputSchema": { "type": "object", "properties": { "js_code": {"type": "string", "description": "JavaScript code"}, "deobfuscate": {"type": "boolean", "description": "Deobfuscate code"} }, "required": ["js_code"] } }, { "name": "reverse_engineer_api", "description": "Reverse engineer API endpoints", "inputSchema": { "type": "object", "properties": { "traffic_data": {"type": "object", "description": "Network traffic data"}, "js_analysis": {"type": "object", "description": "JavaScript analysis results"} } } }, { "name": "analyze_wasm", "description": "Analyze WebAssembly modules", "inputSchema": { "type": "object", "properties": { "wasm_data": {"type": "string", "description": "WASM binary data"} } } }, { "name": "security_analysis", "description": "Perform security analysis", "inputSchema": { "type": "object", "properties": { "analysis_data": {"type": "object", "description": "Data to analyze"}, "check_headers": {"type": "boolean", "description": "Check security headers"}, "check_cves": {"type": "boolean", "description": "Check for CVEs"} } } } ]) # Infrastructure Tools if self.config.enable_infrastructure: tools.extend([ { "name": "database_query", "description": "Execute database query", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "SQL query"}, "params": {"type": "array", "description": "Query parameters"} }, "required": ["query"] } }, { "name": "cache_operation", "description": "Perform cache operation", "inputSchema": { "type": "object", "properties": { "operation": {"type": "string", "description": "Operation type"}, "key": {"type": "string", "description": "Cache key"}, "value": {"type": "string", "description": "Cache value"}, "ttl": {"type": "integer", "description": "Time to live"} }, "required": ["operation", "key"] } }, { "name": "publish_message", "description": "Publish message to channel", "inputSchema": { "type": "object", "properties": { "channel": {"type": "string", "description": "Channel name"}, "message": {"type": "object", "description": "Message data"} }, "required": ["channel"] } }, { "name": "fetch_content", "description": "Fetch content from URL", "inputSchema": { "type": "object", "properties": { "url": {"type": "string", "description": "URL to fetch"}, "timeout": {"type": "integer", "description": "Request timeout"}, "retries": {"type": "integer", "description": "Number of retries"} }, "required": ["url"] } }, { "name": "record_metric", "description": "Record metric", "inputSchema": { "type": "object", "properties": { "metric_name": {"type": "string", "description": "Metric name"}, "value": {"type": "number", "description": "Metric value"}, "labels": {"type": "object", "description": "Metric labels"} }, "required": ["metric_name", "value"] } } ]) # Advanced Analysis Tools (Always enabled if server is running, usually) tools.extend([ { "name": "logic_identification", "description": "Identify logic patterns", "inputSchema": { "type": "object", "properties": { "disassembly_data": {"type": "object", "description": "Disassembly data"}, "analyze_control_flow": {"type": "boolean", "description": "Analyze control flow"}, "analyze_data_flow": {"type": "boolean", "description": "Analyze data flow"} } } }, { "name": "traffic_interception", "description": "Intercept network traffic", "inputSchema": { "type": "object", "properties": { "target_url": {"type": "string", "description": "Target URL"}, "ssl_intercept": {"type": "boolean", "description": "Intercept SSL"}, "capture_duration": {"type": "integer", "description": "Capture duration"} }, "required": ["target_url"] } }, { "name": "generate_report", "description": "Generate analysis report", "inputSchema": { "type": "object", "properties": { "analysis_results": {"type": "object", "description": "Analysis results"}, "format": {"type": "string", "description": "Report format"}, "include_summary": {"type": "boolean", "description": "Include summary"} } } }, { "name": "rag_orchestration", "description": "RAG Orchestration", "inputSchema": { "type": "object", "properties": { "query": {"type": "string"}, "context_limit": {"type": "integer"}, "threshold": {"type": "number"} } } }, { "name": "deep_research", "description": "Deep Research Agent", "inputSchema": { "type": "object", "properties": { "topic": {"type": "string"}, "max_sources": {"type": "integer"}, "synthesize": {"type": "boolean"} } } } ]) # Management Tools tools.extend([ { "name": "session_management", "description": "Manage analysis sessions", "inputSchema": { "type": "object", "properties": { "operation": {"type": "string", "description": "Operation type"}, "session_id": {"type": "string", "description": "Session ID"} }, "required": ["operation"] } }, { "name": "task_scheduler", "description": "Schedule analysis tasks", "inputSchema": { "type": "object", "properties": { "task_type": {"type": "string", "description": "Task type"}, "schedule": {"type": "string", "description": "Schedule"}, "parameters": {"type": "object", "description": "Task parameters"} }, "required": ["task_type"] } }, { "name": "result_aggregation", "description": "Aggregate analysis results", "inputSchema": { "type": "object", "properties": { "results": {"type": "array", "description": "Results to aggregate"}, "aggregation_type": {"type": "string", "description": "Aggregation type"} }, "required": ["results"] } }, { "name": "version_management", "description": "Version management", "inputSchema": { "type": "object", "properties": { "component_name": {"type": "string"}, "version": {"type": "string"} } } }, { "name": "quality_gate", "description": "Quality Gate", "inputSchema": { "type": "object", "properties": { "analysis_results": {"type": "object"} } } }, { "name": "governance_check", "description": "Governance Check", "inputSchema": { "type": "object", "properties": { "action": {"type": "string"} } } }, { "name": "generate_document", "description": "Generate Document", "inputSchema": { "type": "object", "properties": { "document_type": {"type": "string"}, "data": {"type": "object"} } } } ]) # Utility Tools tools.extend([ { "name": "url_frontier_operation", "description": "Manage URL frontier", "inputSchema": { "type": "object", "properties": { "operation": {"type": "string", "description": "Operation (e.g., add, delete)"}, "url": {"type": "string", "description": "URL to add"}, "priority": {"type": "integer", "description": "URL priority"} }, "required": ["url"] } }, { "name": "url_frontier", "description": "Alias for url_frontier_operation", "inputSchema": { "type": "object", "properties": { "operation": {"type": "string", "description": "Operation"}, "url": {"type": "string", "description": "URL"}, "priority": {"type": "integer"} }, "required": ["url"] } }, { "name": "api_pattern_matcher", "description": "Match API patterns", "inputSchema": { "type": "object", "properties": { "traffic_data": {"type": "object", "description": "Traffic data"}, "pattern_type": {"type": "string", "description": "Pattern type"} } } }, { "name": "response_classifier", "description": "Classify responses", "inputSchema": { "type": "object", "properties": { "response_data": {"type": "object", "description": "Response data"}, "infer_schema": {"type": "boolean", "description": "Infer schema"} } } }, { "name": "websocket_analyzer", "description": "Analyze WebSocket connections", "inputSchema": { "type": "object", "properties": { "websocket_data": {"type": "object", "description": "WebSocket data"}, "analyze_handshake": {"type": "boolean", "description": "Analyze handshake"} } } }, { "name": "crawl_scheduler", "description": "Schedule crawl operations", "inputSchema": { "type": "object", "properties": { "operation": {"type": "string", "description": "Operation type"}, "job_data": {"type": "object", "description": "Job data"}, "priority": {"type": "integer", "description": "Job priority"} }, "required": ["operation"] } } ]) # System Tools tools.extend([ { "name": "metrics_collector", "description": "Collect metrics", "inputSchema": { "type": "object", "properties": { "metric_type": {"type": "string", "description": "Metric type"}, "metric_name": {"type": "string", "description": "Metric name"}, "value": {"type": "number", "description": "Metric value"}, "labels": {"type": "object", "description": "Metric labels"} }, "required": ["metric_type", "metric_name", "value"] } }, { "name": "multi_level_cache", "description": "Multi-level cache operations", "inputSchema": { "type": "object", "properties": { "operation": {"type": "string", "description": "Operation type"}, "key": {"type": "string", "description": "Cache key"}, "value": {"type": "string", "description": "Cache value"}, "ttl": {"type": "integer", "description": "Time to live"} }, "required": ["operation", "key"] } }, { "name": "configuration_service", "description": "Configuration service", "inputSchema": { "type": "object", "properties": { "operation": {"type": "string", "description": "Operation type"}, "key": {"type": "string", "description": "Config key"}, "value": {"type": "string", "description": "Config value"} }, "required": ["operation", "key"] } }, { "name": "llm_interface", "description": "LLM interface", "inputSchema": { "type": "object", "properties": { "prompt": {"type": "string", "description": "Prompt"}, "model": {"type": "string", "description": "Model name"}, "max_tokens": {"type": "integer", "description": "Max tokens"}, "temperature": {"type": "number", "description": "Temperature"} }, "required": ["prompt"] } } ]) # NLP Tools tools.extend([ { "name": "natural_language_interface", "description": "Natural language interface", "inputSchema": { "type": "object", "properties": { "command": {"type": "string", "description": "Command"}, "context": {"type": "object", "description": "Context"} }, "required": ["command"] } }, { "name": "poc_validation", "description": "Validate proof of concept", "inputSchema": { "type": "object", "properties": { "vulnerability_finding": {"type": "object", "description": "Vulnerability finding"}, "generate_poc": {"type": "boolean", "description": "Generate PoC"}, "execute_poc": {"type": "boolean", "description": "Execute PoC"} } } } ]) return tools def shutdown(self) -> None: """Shutdown server and cleanup resources""" # Prevent double shutdown if hasattr(self, '_shutdown_called') and self._shutdown_called: return self._shutdown_called = True logger.info("Shutting down RAVERSE MCP Server") if self.db_manager: try: self.db_manager.close() except Exception as e: logger.warning(f"Error closing database: {str(e)}") if self.cache_manager: try: self.cache_manager.close() except Exception as e: logger.warning(f"Error closing cache: {str(e)}") logger.info("RAVERSE MCP Server shutdown complete") async def main_async(): """Async Main entry point""" config = get_config() setup_logging(config.log_level) logger.info(f"Starting RAVERSE MCP Server v{config.server_version} (Official SDK)") # Create server instance normally using __init__ # This ensures attributes are initialized to None app_server = MCPServer(config) # Initialize MCP SDK Server mcp_server = Server("raverse-mcp-server") @mcp_server.list_tools() async def list_tools() -> List[types.Tool]: tools_data = app_server.get_tools_list() return [types.Tool(**t) for t in tools_data] @mcp_server.call_tool() async def call_tool(name: str, arguments: dict) -> List[types.TextContent | types.ImageContent | types.EmbeddedResource]: result = await app_server.handle_tool_call(name, arguments) return [types.TextContent(type="text", text=json.dumps(result))] # Run with stdio async with stdio_server() as (read_stream, write_stream): try: await mcp_server.run( read_stream, write_stream, mcp_server.create_initialization_options() ) except Exception as e: logger.error(f"MCP Server Error: {e}") finally: if hasattr(app_server, 'shutdown'): app_server.shutdown() def main(): """Entry point for setuptools""" try: asyncio.run(main_async()) except KeyboardInterrupt: pass except Exception as e: sys.stderr.write(f"Fatal Error: {e}\n") sys.exit(1) if __name__ == "__main__": main()