""" Genesis V2 MCP Stdio Server ============================ Stdio wrapper for MCP integration with IDEs. Implements Standard Input/Output protocol for MCP. """ import asyncio import os import sys # Ensure project root is in path root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) if root_dir not in sys.path: sys.path.insert(0, root_dir) sys.path.insert(0, os.path.join(root_dir, "lib")) import json import logging from typing import Dict, Any # Configure logging to stderr (stdout is for MCP protocol) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', stream=sys.stderr ) logger = logging.getLogger(__name__) # Import our native server from genesis_v2.mcp.native_server import ( configure, kg_search, kg_get, kg_ingest, kg_context, kg_stats, browser_navigate, browser_click, browser_screenshot, get_identity, health_check, ) from genesis_v2.mcp.antigravity_tools import ( antigravity_plan, antigravity_execute, gemini_query, sync_push, sync_pull, sync_status, memory_query, memory_store, antigravity_browser, ) from genesis_v2.mcp.task_board_tools import ( task_board_create, task_board_list, task_board_update, task_board_claim, task_board_complete, ) from genesis_v2.mcp.file_lock_tools import ( file_lock_claim, file_lock_release, file_lock_status, ) from genesis_v2.mcp.handoff_tools import ( handoff_push, handoff_pull, ) from genesis_v2.mcp.execution_bridge_tools import ( exec_decompose, exec_story, exec_swarm, exec_status, ) from genesis_v2.core.ultrathink_bridge import bridge as ultrathink_bridge # ============================================================================= # MCP Protocol Constants # ============================================================================= MCP_VERSION = "2024-11-05" # MCP protocol version TOOLS = { "kg_search": { "description": "Search knowledge graph for entities matching query", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "limit": {"type": "integer", "description": "Max results", "default": 10} }, "required": ["query"] } }, "kg_get": { "description": "Get specific entity by ID", "inputSchema": { "type": "object", "properties": { "entity_id": {"type": "string", "description": "Entity ID"} }, "required": ["entity_id"] } }, "kg_ingest": { "description": "Add new entity to knowledge graph", "inputSchema": { "type": "object", "properties": { "entity_json": {"type": "string", "description": "JSON entity data"} }, "required": ["entity_json"] } }, "kg_context": { "description": "Get knowledge graph context for agent", "inputSchema": {"type": "object", "properties": {}} }, "kg_stats": { "description": "Get knowledge graph statistics", "inputSchema": {"type": "object", "properties": {}} }, "browser_navigate": { "description": "Navigate browser to URL", "inputSchema": { "type": "object", "properties": { "url": {"type": "string", "description": "Target URL"} }, "required": ["url"] } }, "browser_click": { "description": "Click element by CSS selector", "inputSchema": { "type": "object", "properties": { "selector": {"type": "string", "description": "CSS selector"} }, "required": ["selector"] } }, "browser_screenshot": { "description": "Capture browser screenshot", "inputSchema": {"type": "object", "properties": {}} }, "get_identity": { "description": "Get agent identity and persona", "inputSchema": {"type": "object", "properties": {}} }, "health_check": { "description": "Check server health status of all components", "inputSchema": {"type": "object", "properties": {}} }, "antigravity_plan": { "description": "Create a workflow plan via Gemini Antigravity system", "inputSchema": { "type": "object", "properties": { "goal": {"type": "string", "description": "The goal to plan for"}, "context_json": {"type": "string", "description": "Optional JSON context", "default": "{}"} }, "required": ["goal"] } }, "antigravity_execute": { "description": "Execute a previously created Antigravity workflow plan", "inputSchema": { "type": "object", "properties": { "plan_id": {"type": "string", "description": "ID of the plan to execute"} }, "required": ["plan_id"] } }, "gemini_query": { "description": "Direct Gemini execution for analysis, planning, or generation", "inputSchema": { "type": "object", "properties": { "prompt": {"type": "string", "description": "The prompt to send to Gemini"}, "task_type": {"type": "string", "description": "Task type (general, analysis, planning, code_review)", "default": "general"}, "temperature": {"type": "number", "description": "Sampling temperature 0.0-1.0", "default": 0.7} }, "required": ["prompt"] } }, "sync_push": { "description": "Push context to shared state for cross-agent Claude/Gemini sync", "inputSchema": { "type": "object", "properties": { "context_json": {"type": "string", "description": "JSON context data to push"} }, "required": ["context_json"] } }, "sync_pull": { "description": "Pull latest shared state from cross-agent sync", "inputSchema": {"type": "object", "properties": {}} }, "sync_status": { "description": "Get sync health and component status for Antigravity system", "inputSchema": {"type": "object", "properties": {}} }, "memory_query": { "description": "Query memory bridge with Gemini context for enriched responses", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "The query to search memory for"} }, "required": ["query"] } }, "memory_store": { "description": "Store an insight in the memory bridge for persistent context", "inputSchema": { "type": "object", "properties": { "insight_json": {"type": "string", "description": "JSON with 'content' and optional 'category' fields"} }, "required": ["insight_json"] } }, # Task Board tools (shared Claude/Antigravity board) "task_board_create": { "description": "Create a task on the shared Claude/Antigravity task board", "inputSchema": { "type": "object", "properties": { "task_json": {"type": "string", "description": "JSON with title, description, type, urgency (1-10), impact (1-10)"} }, "required": ["task_json"] } }, "task_board_list": { "description": "List tasks from the shared task board", "inputSchema": { "type": "object", "properties": { "status": {"type": "string", "description": "Filter: pending, in_progress, completed, failed (empty=all)", "default": ""}, "limit": {"type": "integer", "description": "Max results", "default": 20} } } }, "task_board_update": { "description": "Update a task on the shared board (status, priority, etc.)", "inputSchema": { "type": "object", "properties": { "task_id": {"type": "string", "description": "Task ID to update"}, "updates_json": {"type": "string", "description": "JSON with fields to update"} }, "required": ["task_id", "updates_json"] } }, "task_board_claim": { "description": "Claim ownership of a task (atomic operation)", "inputSchema": { "type": "object", "properties": { "task_id": {"type": "string", "description": "Task ID to claim"}, "agent": {"type": "string", "description": "Agent name (claude or antigravity)"} }, "required": ["task_id", "agent"] } }, "task_board_complete": { "description": "Mark a task as complete with result data", "inputSchema": { "type": "object", "properties": { "task_id": {"type": "string", "description": "Task ID to complete"}, "result_json": {"type": "string", "description": "JSON with result data"} }, "required": ["task_id", "result_json"] } }, # File Lock tools (prevent concurrent edits) "file_lock_claim": { "description": "Lock files to prevent concurrent editing by Claude/Antigravity", "inputSchema": { "type": "object", "properties": { "files_json": {"type": "string", "description": "JSON array of file paths to lock"}, "agent": {"type": "string", "description": "Agent claiming the lock"} }, "required": ["files_json", "agent"] } }, "file_lock_release": { "description": "Release file locks", "inputSchema": { "type": "object", "properties": { "files_json": {"type": "string", "description": "JSON array of file paths to unlock"}, "agent": {"type": "string", "description": "Agent releasing the lock"} }, "required": ["files_json", "agent"] } }, "file_lock_status": { "description": "Show all active file locks across agents", "inputSchema": {"type": "object", "properties": {}} }, # Handoff tools (structured context transfer) "handoff_push": { "description": "Push structured handoff when finishing a block of work", "inputSchema": { "type": "object", "properties": { "summary": {"type": "string", "description": "What was done"}, "files_changed_json": {"type": "string", "description": "JSON array of changed file paths", "default": "[]"}, "next_actions_json": {"type": "string", "description": "JSON array of recommended next actions", "default": "[]"} }, "required": ["summary"] } }, "handoff_pull": { "description": "Pull latest handoff(s) and recent events since last sync", "inputSchema": { "type": "object", "properties": { "agent": {"type": "string", "description": "Filter by agent (empty=all)", "default": ""} } } }, # Execution Bridge tools (Claude -> Gemini swarm) "exec_decompose": { "description": "Decompose a task into RWL stories via Gemini", "inputSchema": { "type": "object", "properties": { "task": {"type": "string", "description": "Task to decompose"}, "context": {"type": "string", "description": "Optional context", "default": ""} }, "required": ["task"] } }, "exec_story": { "description": "Execute a single RWL story via Gemini", "inputSchema": { "type": "object", "properties": { "story_json": {"type": "string", "description": "JSON story with id, title, description, acceptance_criteria, priority"} }, "required": ["story_json"] } }, "exec_swarm": { "description": "Execute multiple RWL stories in parallel via Gemini swarm", "inputSchema": { "type": "object", "properties": { "stories_json": {"type": "string", "description": "JSON array of story objects"}, "max_parallel": {"type": "integer", "description": "Max concurrent Gemini agents", "default": 3} }, "required": ["stories_json"] } }, "exec_status": { "description": "Get execution layer status: rate maximizer utilization, stats, best model", "inputSchema": {"type": "object", "properties": {}} }, "ultrathink_status": { "description": "Get status of the Ultrathink Bridge (session, active tools, dispatch queue)", "inputSchema": {"type": "object", "properties": {}} }, "ultrathink_dispatch": { "description": "Dispatch a task via the Ultrathink Bridge to the optimal tool", "inputSchema": { "type": "object", "properties": { "task": {"type": "string", "description": "The task description"}, "priority": {"type": "string", "description": "Priority (low, medium, high)", "default": "medium"} }, "required": ["task"] } }, "antigravity_browser": { "description": "Execute a browser task via the Antigravity Master Browser Engine (High Stealth)", "inputSchema": { "type": "object", "properties": { "task": {"type": "string", "description": "Description of the browser task to perform"} }, "required": ["task"] } } } TOOL_HANDLERS = { "kg_search": lambda args: kg_search(args.get("query", ""), args.get("limit", 10)), "kg_get": lambda args: kg_get(args.get("entity_id", "")), "kg_ingest": lambda args: kg_ingest(args.get("entity_json", "{}")), "kg_context": lambda args: kg_context(), "kg_stats": lambda args: kg_stats(), "browser_navigate": lambda args: browser_navigate(args.get("url", "")), "browser_click": lambda args: browser_click(args.get("selector", "")), "browser_screenshot": lambda args: browser_screenshot(), "get_identity": lambda args: get_identity(), "health_check": lambda args: health_check(), "antigravity_plan": lambda args: antigravity_plan(args.get("goal", ""), args.get("context_json", "{}")), "antigravity_execute": lambda args: antigravity_execute(args.get("plan_id", "")), "gemini_query": lambda args: gemini_query(args.get("prompt", ""), args.get("task_type", "general"), args.get("temperature", 0.7)), "sync_push": lambda args: sync_push(args.get("context_json", "{}")), "sync_pull": lambda args: sync_pull(), "sync_status": lambda args: sync_status(), "memory_query": lambda args: memory_query(args.get("query", "")), "memory_store": lambda args: memory_store(args.get("insight_json", "{}")), # Task Board "task_board_create": lambda args: task_board_create(args.get("task_json", "{}")), "task_board_list": lambda args: task_board_list(args.get("status", ""), args.get("limit", 20)), "task_board_update": lambda args: task_board_update(args.get("task_id", ""), args.get("updates_json", "{}")), "task_board_claim": lambda args: task_board_claim(args.get("task_id", ""), args.get("agent", "")), "task_board_complete": lambda args: task_board_complete(args.get("task_id", ""), args.get("result_json", "{}")), # File Locks "file_lock_claim": lambda args: file_lock_claim(args.get("files_json", "[]"), args.get("agent", "")), "file_lock_release": lambda args: file_lock_release(args.get("files_json", "[]"), args.get("agent", "")), "file_lock_status": lambda args: file_lock_status(), # Handoffs "handoff_push": lambda args: handoff_push(args.get("summary", ""), args.get("files_changed_json", "[]"), args.get("next_actions_json", "[]")), "handoff_pull": lambda args: handoff_pull(args.get("agent", "")), # Execution Bridge "exec_decompose": lambda args: exec_decompose(args.get("task", ""), args.get("context", "")), "exec_story": lambda args: exec_story(args.get("story_json", "{}")), "exec_swarm": lambda args: exec_swarm(args.get("stories_json", "[]"), args.get("max_parallel", 3)), "exec_status": lambda args: exec_status(), "ultrathink_status": lambda args: ultrathink_bridge.get_bridge_status(), "ultrathink_dispatch": lambda args: ultrathink_bridge.dispatch(args.get("task", ""), args.get("priority", "medium")), "antigravity_browser": lambda args: antigravity_browser(args.get("task", "")), } # ============================================================================= # MCP Protocol Handler # ============================================================================= def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: """Handle an MCP request and return response.""" method = request.get("method", "") request_id = request.get("id") params = request.get("params", {}) logger.info(f"MCP request: {method}") if method == "initialize": return { "jsonrpc": "2.0", "id": request_id, "result": { "protocolVersion": MCP_VERSION, "capabilities": { "tools": {} }, "serverInfo": { "name": "genesis-v2", "version": "2.0.0" } } } elif method == "tools/list": tools_list = [ {"name": name, **spec} for name, spec in TOOLS.items() ] return { "jsonrpc": "2.0", "id": request_id, "result": {"tools": tools_list} } elif method == "tools/call": tool_name = params.get("name", "") arguments = params.get("arguments", {}) if tool_name in TOOL_HANDLERS: try: result = TOOL_HANDLERS[tool_name](arguments) return { "jsonrpc": "2.0", "id": request_id, "result": { "content": [{"type": "text", "text": result}] } } except Exception as e: logger.error(f"Tool error: {e}") return { "jsonrpc": "2.0", "id": request_id, "error": {"code": -32000, "message": str(e)} } else: return { "jsonrpc": "2.0", "id": request_id, "error": {"code": -32601, "message": f"Unknown tool: {tool_name}"} } elif method == "notifications/initialized": # This is a notification, no response needed return None else: return { "jsonrpc": "2.0", "id": request_id, "error": {"code": -32601, "message": f"Method not found: {method}"} } def read_message() -> Dict[str, Any]: """Read a JSON-RPC message from stdin.""" # Read Content-Length header headers = {} while True: line = sys.stdin.readline() if line == "\r\n" or line == "\n" or line == "": break if ":" in line: key, value = line.strip().split(":", 1) headers[key.strip()] = value.strip() content_length = int(headers.get("Content-Length", 0)) if content_length > 0: content = sys.stdin.read(content_length) return json.loads(content) return {} def write_message(message: Dict[str, Any]): """Write a JSON-RPC message to stdout.""" content = json.dumps(message) sys.stdout.write(f"Content-Length: {len(content)}\r\n\r\n{content}") sys.stdout.flush() def run_stdio_server(): """Run the MCP server using stdio transport.""" logger.info("Genesis V2 MCP Server starting (stdio mode)") # Configure the server configure() logger.info("Server ready, waiting for requests...") try: while True: try: request = read_message() if not request: continue response = handle_request(request) if response: write_message(response) except json.JSONDecodeError as e: logger.error(f"JSON decode error: {e}") except Exception as e: logger.error(f"Error processing request: {e}") except KeyboardInterrupt: logger.info("Server shutting down") except Exception as e: logger.error(f"Server error: {e}") # ============================================================================= # Entry Point # ============================================================================= if __name__ == "__main__": run_stdio_server()