lmiranda
72b3436a24
- design-gate.md and data-gate.md declare gate_contract: v1 - domain-consultation.md Gate Command Reference includes Contract column - validate_workflow_integration now checks contract version compatibility - Tests added for match, mismatch, and missing contract scenarios Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
310 lines
13 KiB
Python
310 lines
13 KiB
Python
"""
|
|
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
|
|
|
|
from .parse_tools import ParseTools
|
|
from .validation_tools import ValidationTools
|
|
from .report_tools import ReportTools
|
|
|
|
# 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")
|
|
self.parse_tools = ParseTools()
|
|
self.validation_tools = ValidationTools()
|
|
self.report_tools = ReportTools()
|
|
|
|
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"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="validate_workflow_integration",
|
|
description="Validate that a domain plugin exposes the required advisory interfaces (gate command, review command, advisory agent) expected by projman's domain-consultation skill. Also checks gate contract version compatibility.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"plugin_path": {
|
|
"type": "string",
|
|
"description": "Path to the domain plugin directory"
|
|
},
|
|
"domain_label": {
|
|
"type": "string",
|
|
"description": "The Domain/* label it claims to handle, e.g. Domain/Viz"
|
|
},
|
|
"expected_contract": {
|
|
"type": "string",
|
|
"description": "Expected contract version (e.g., 'v1'). If provided, validates the gate command's contract matches."
|
|
}
|
|
},
|
|
"required": ["plugin_path", "domain_label"]
|
|
}
|
|
),
|
|
# 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 == "validate_workflow_integration":
|
|
result = await self._validate_workflow_integration(**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)
|
|
)]
|
|
|
|
# Parse tool implementations (Issue #186)
|
|
|
|
async def _parse_plugin_interface(self, plugin_path: str) -> dict:
|
|
"""Parse plugin interface from README.md"""
|
|
return await self.parse_tools.parse_plugin_interface(plugin_path)
|
|
|
|
async def _parse_claude_md_agents(self, claude_md_path: str) -> dict:
|
|
"""Parse agents from CLAUDE.md"""
|
|
return await self.parse_tools.parse_claude_md_agents(claude_md_path)
|
|
|
|
# Validation tool implementations (Issue #187)
|
|
|
|
async def _validate_compatibility(self, plugin_a: str, plugin_b: str) -> dict:
|
|
"""Validate compatibility between plugins"""
|
|
return await self.validation_tools.validate_compatibility(plugin_a, plugin_b)
|
|
|
|
async def _validate_agent_refs(self, agent_name: str, claude_md_path: str, plugin_paths: list = None) -> dict:
|
|
"""Validate agent tool references"""
|
|
return await self.validation_tools.validate_agent_refs(agent_name, claude_md_path, plugin_paths)
|
|
|
|
async def _validate_data_flow(self, agent_name: str, claude_md_path: str) -> dict:
|
|
"""Validate agent data flow"""
|
|
return await self.validation_tools.validate_data_flow(agent_name, claude_md_path)
|
|
|
|
async def _validate_workflow_integration(
|
|
self,
|
|
plugin_path: str,
|
|
domain_label: str,
|
|
expected_contract: str = None
|
|
) -> dict:
|
|
"""Validate domain plugin exposes required advisory interfaces"""
|
|
return await self.validation_tools.validate_workflow_integration(
|
|
plugin_path, domain_label, expected_contract
|
|
)
|
|
|
|
# Report tool implementations (Issue #188)
|
|
|
|
async def _generate_compatibility_report(self, marketplace_path: str, format: str = "markdown") -> dict:
|
|
"""Generate comprehensive compatibility report"""
|
|
return await self.report_tools.generate_compatibility_report(marketplace_path, format)
|
|
|
|
async def _list_issues(self, marketplace_path: str, severity: str = "all", issue_type: str = "all") -> dict:
|
|
"""List validation issues with filtering"""
|
|
return await self.report_tools.list_issues(marketplace_path, severity, 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())
|