""" MCP Server entry point for Contract Validator. Provides cross-plugin compatibility validation and Claude.md agent verification tools to Claude Code via JSON-RPC 2.0 over stdio. """ import asyncio import logging import json from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent # Suppress noisy MCP validation warnings on stderr logging.basicConfig(level=logging.INFO) logging.getLogger("root").setLevel(logging.ERROR) logging.getLogger("mcp").setLevel(logging.ERROR) logger = logging.getLogger(__name__) class ContractValidatorMCPServer: """MCP Server for cross-plugin compatibility validation""" def __init__(self): self.server = Server("contract-validator-mcp") async def initialize(self): """Initialize server.""" logger.info("Contract Validator MCP Server initialized") def setup_tools(self): """Register all available tools with the MCP server""" @self.server.list_tools() async def list_tools() -> list[Tool]: """Return list of available tools""" tools = [ # Parse tools (to be implemented in #186) Tool( name="parse_plugin_interface", description="Parse plugin README.md to extract interface declarations (inputs, outputs, tools)", inputSchema={ "type": "object", "properties": { "plugin_path": { "type": "string", "description": "Path to plugin directory or README.md" } }, "required": ["plugin_path"] } ), Tool( name="parse_claude_md_agents", description="Parse Claude.md to extract agent definitions and their tool sequences", inputSchema={ "type": "object", "properties": { "claude_md_path": { "type": "string", "description": "Path to CLAUDE.md file" } }, "required": ["claude_md_path"] } ), # Validation tools (to be implemented in #187) Tool( name="validate_compatibility", description="Validate compatibility between two plugin interfaces", inputSchema={ "type": "object", "properties": { "plugin_a": { "type": "string", "description": "Path to first plugin" }, "plugin_b": { "type": "string", "description": "Path to second plugin" } }, "required": ["plugin_a", "plugin_b"] } ), Tool( name="validate_agent_refs", description="Validate that all tool references in an agent definition exist", inputSchema={ "type": "object", "properties": { "agent_name": { "type": "string", "description": "Name of agent to validate" }, "claude_md_path": { "type": "string", "description": "Path to CLAUDE.md containing agent" }, "plugin_paths": { "type": "array", "items": {"type": "string"}, "description": "Paths to available plugins" } }, "required": ["agent_name", "claude_md_path"] } ), Tool( name="validate_data_flow", description="Validate data flow through an agent's tool sequence", inputSchema={ "type": "object", "properties": { "agent_name": { "type": "string", "description": "Name of agent to validate" }, "claude_md_path": { "type": "string", "description": "Path to CLAUDE.md containing agent" } }, "required": ["agent_name", "claude_md_path"] } ), # Report tools (to be implemented in #188) Tool( name="generate_compatibility_report", description="Generate a comprehensive compatibility report for all plugins", inputSchema={ "type": "object", "properties": { "marketplace_path": { "type": "string", "description": "Path to marketplace root directory" }, "format": { "type": "string", "enum": ["markdown", "json"], "default": "markdown", "description": "Output format" } }, "required": ["marketplace_path"] } ), Tool( name="list_issues", description="List validation issues with optional filtering", inputSchema={ "type": "object", "properties": { "marketplace_path": { "type": "string", "description": "Path to marketplace root directory" }, "severity": { "type": "string", "enum": ["error", "warning", "info", "all"], "default": "all", "description": "Filter by severity" }, "issue_type": { "type": "string", "enum": ["missing_tool", "interface_mismatch", "optional_dependency", "undeclared_output", "all"], "default": "all", "description": "Filter by issue type" } }, "required": ["marketplace_path"] } ) ] return tools @self.server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Handle tool invocation.""" try: # All tools return placeholder responses for now # Actual implementation will be added in issues #186, #187, #188 if name == "parse_plugin_interface": result = await self._parse_plugin_interface(**arguments) elif name == "parse_claude_md_agents": result = await self._parse_claude_md_agents(**arguments) elif name == "validate_compatibility": result = await self._validate_compatibility(**arguments) elif name == "validate_agent_refs": result = await self._validate_agent_refs(**arguments) elif name == "validate_data_flow": result = await self._validate_data_flow(**arguments) elif name == "generate_compatibility_report": result = await self._generate_compatibility_report(**arguments) elif name == "list_issues": result = await self._list_issues(**arguments) else: raise ValueError(f"Unknown tool: {name}") return [TextContent( type="text", text=json.dumps(result, indent=2, default=str) )] except Exception as e: logger.error(f"Tool {name} failed: {e}") return [TextContent( type="text", text=json.dumps({"error": str(e)}, indent=2) )] # Placeholder implementations - to be completed in subsequent issues async def _parse_plugin_interface(self, plugin_path: str) -> dict: """Parse plugin interface from README.md (placeholder)""" return { "status": "not_implemented", "message": "Implementation pending - Issue #186", "plugin_path": plugin_path } async def _parse_claude_md_agents(self, claude_md_path: str) -> dict: """Parse agents from CLAUDE.md (placeholder)""" return { "status": "not_implemented", "message": "Implementation pending - Issue #186", "claude_md_path": claude_md_path } async def _validate_compatibility(self, plugin_a: str, plugin_b: str) -> dict: """Validate compatibility between plugins (placeholder)""" return { "status": "not_implemented", "message": "Implementation pending - Issue #187", "plugin_a": plugin_a, "plugin_b": plugin_b } async def _validate_agent_refs(self, agent_name: str, claude_md_path: str, plugin_paths: list = None) -> dict: """Validate agent tool references (placeholder)""" return { "status": "not_implemented", "message": "Implementation pending - Issue #187", "agent_name": agent_name, "claude_md_path": claude_md_path } async def _validate_data_flow(self, agent_name: str, claude_md_path: str) -> dict: """Validate agent data flow (placeholder)""" return { "status": "not_implemented", "message": "Implementation pending - Issue #187", "agent_name": agent_name, "claude_md_path": claude_md_path } async def _generate_compatibility_report(self, marketplace_path: str, format: str = "markdown") -> dict: """Generate compatibility report (placeholder)""" return { "status": "not_implemented", "message": "Implementation pending - Issue #188", "marketplace_path": marketplace_path, "format": format } async def _list_issues(self, marketplace_path: str, severity: str = "all", issue_type: str = "all") -> dict: """List validation issues (placeholder)""" return { "status": "not_implemented", "message": "Implementation pending - Issue #188", "marketplace_path": marketplace_path, "severity": severity, "issue_type": issue_type } async def run(self): """Run the MCP server""" await self.initialize() self.setup_tools() 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""" server = ContractValidatorMCPServer() await server.run() if __name__ == "__main__": asyncio.run(main())