Katamari MCP Server

""" Development CLI Tools for Capability Management Command-line interface for managing capabilities, validation, and testing. """ import argparse import asyncio import json import sys from pathlib import Path from typing import List, Optional from .validator import PackageValidator, ValidationResult from .testing import CapabilityTester, TestType class DevCLI: """Development CLI for Katamari MCP capabilities.""" def __init__(self): self.validator = PackageValidator() self.tester = CapabilityTester() def create_parser(self) -> argparse.ArgumentParser: """Create the argument parser for the CLI.""" parser = argparse.ArgumentParser( prog="katamari-dev", description="Katamari MCP Development Tools" ) subparsers = parser.add_subparsers(dest="command", help="Available commands") # Validate command validate_parser = subparsers.add_parser("validate", help="Validate a capability package") validate_parser.add_argument("path", help="Path to the capability package") validate_parser.add_argument("--output", "-o", help="Output file for validation report") validate_parser.add_argument("--json", action="store_true", help="Output JSON format") validate_parser.add_argument("--python-version", default="3.9", help="Target Python version") # Test command test_parser = subparsers.add_parser("test", help="Test capabilities") test_parser.add_argument("path", nargs="?", default=".", help="Path to test (default: current directory)") test_parser.add_argument("--suite", "-s", help="Specific test suite to run") test_parser.add_argument("--type", "-t", choices=[t.value for t in TestType], help="Test type filter") test_parser.add_argument("--isolated", action="store_true", help="Run tests in isolated environments") test_parser.add_argument("--output", "-o", help="Output file for test report") test_parser.add_argument("--json", action="store_true", help="Output JSON format") # List command list_parser = subparsers.add_parser("list", help="List available capabilities") list_parser.add_argument("path", nargs="?", default=".", help="Path to search (default: current directory)") list_parser.add_argument("--type", choices=["capability", "test"], help="Filter by type") # Create command create_parser = subparsers.add_parser("create", help="Create a new capability") create_parser.add_argument("name", help="Name of the capability") create_parser.add_argument("--path", "-p", default=".", help="Path to create capability (default: current directory)") create_parser.add_argument("--template", choices=["basic", "advanced", "web"], default="basic", help="Capability template") # Check command check_parser = subparsers.add_parser("check", help="Quick health check of the environment") check_parser.add_argument("--detailed", action="store_true", help="Show detailed information") return parser async def cmd_validate(self, args) -> int: """Execute the validate command.""" package_path = Path(args.path) if not package_path.exists(): print(f"❌ Error: Path '{package_path}' does not exist") return 1 if not package_path.is_dir(): print(f"❌ Error: Path '{package_path}' is not a directory") return 1 print(f"🔍 Validating package: {package_path}") # Update validator with target Python version self.validator.compatibility_validator.target_python_version = args.python_version # Perform validation result = self.validator.validate_package(package_path) # Output results if args.json: output_data = { "is_valid": result.is_valid, "package_hash": result.package_hash, "metadata": result.metadata, "issues": [issue.__dict__ for issue in result.issues] } if args.output: with open(args.output, 'w') as f: json.dump(output_data, f, indent=2) print(f"📄 JSON report saved to: {args.output}") else: print(json.dumps(output_data, indent=2)) else: report = self.validator.generate_report(result) if args.output: with open(args.output, 'w') as f: f.write(report) print(f"📄 Report saved to: {args.output}") else: print(report) return 0 if result.is_valid else 1 async def cmd_test(self, args) -> int: """Execute the test command.""" test_path = Path(args.path) if not test_path.exists(): print(f"❌ Error: Path '{test_path}' does not exist") return 1 print(f"🧪 Testing capabilities in: {test_path}") # Configure tester self.tester.isolated = args.isolated # Discover and register capabilities if test_path.is_dir(): # Look for capability directories for item in test_path.iterdir(): if item.is_dir() and (item / "__init__.py").exists(): suite = self.tester.register_capability(item) # Discover tests discovered_tests = self.tester.discover_tests(item) for test in discovered_tests: suite.add_test(test) # Run tests if args.suite: if args.suite not in self.tester.test_suites: print(f"❌ Error: Test suite '{args.suite}' not found") return 1 results = {args.suite: await self.tester.run_test_suite(args.suite)} else: results = await self.tester.run_all_tests() # Generate report if args.json: output_data = { "results": { suite: [result.to_dict() for result in suite_results] for suite, suite_results in results.items() } } if args.output: with open(args.output, 'w') as f: json.dump(output_data, f, indent=2) print(f"📄 JSON report saved to: {args.output}") else: print(json.dumps(output_data, indent=2)) else: report = self.tester.generate_report(results) if args.output: with open(args.output, 'w') as f: f.write(report) print(f"📄 Report saved to: {args.output}") else: print(report) # Cleanup self.tester.cleanup() # Return non-zero if any tests failed total_failed = sum( 1 for suite_results in results.values() for result in suite_results if result.status in ["failed", "error"] ) return 1 if total_failed > 0 else 0 def cmd_list(self, args) -> int: """Execute the list command.""" search_path = Path(args.path) if not search_path.exists(): print(f"❌ Error: Path '{search_path}' does not exist") return 1 print(f"📋 Listing items in: {search_path}") if args.type == "capability" or not args.type: print("\n🔧 Capabilities:") for item in search_path.rglob("*/"): if (item / "__init__.py").exists(): rel_path = item.relative_to(search_path) print(f" • {rel_path}") if args.type == "test" or not args.type: print("\n🧪 Test Files:") for test_file in search_path.rglob("test_*.py"): rel_path = test_file.relative_to(search_path) print(f" • {rel_path}") for test_file in search_path.rglob("*_test.py"): rel_path = test_file.relative_to(search_path) print(f" • {rel_path}") return 0 def cmd_create(self, args) -> int: """Execute the create command.""" create_path = Path(args.path) capability_path = create_path / args.name if capability_path.exists(): print(f"❌ Error: Capability '{args.name}' already exists") return 1 print(f"🏗️ Creating capability: {args.name}") # Create directory structure capability_path.mkdir(parents=True, exist_ok=True) # Create __init__.py init_content = self._get_capability_template(args.template, args.name) with open(capability_path / "__init__.py", 'w') as f: f.write(init_content) # Create test file test_content = self._get_test_template(args.template, args.name) with open(capability_path / f"test_{args.name}.py", 'w') as f: f.write(test_content) # Create README readme_content = self._get_readme_template(args.template, args.name) with open(capability_path / "README.md", 'w') as f: f.write(readme_content) print(f"✅ Capability '{args.name}' created successfully at: {capability_path}") print(f"📝 Next steps:") print(f" 1. Edit {capability_path / '__init__.py'} to implement your capability") print(f" 2. Run tests: katamari-dev test {capability_path}") print(f" 3. Validate: katamari-dev validate {capability_path}") return 0 def cmd_check(self, args) -> int: """Execute the check command.""" print("🔍 Katamari MCP Environment Check") print("=" * 40) # Check Python version import sys print(f"Python Version: {sys.version}") # Check dependencies try: import pydantic print(f"✅ Pydantic: {pydantic.__version__}") except ImportError: print("❌ Pydantic: Not installed") try: import mcp print(f"✅ MCP: {mcp.__version__}") except ImportError: print("❌ MCP: Not installed") # Check project structure current_path = Path.cwd() print(f"\n📁 Current Directory: {current_path}") required_files = ["pyproject.toml", "README.md"] for file in required_files: if (current_path / file).exists(): print(f"✅ {file}: Found") else: print(f"❌ {file}: Not found") # Check capabilities directory capabilities_dir = current_path / "katamari_mcp" / "capabilities" if capabilities_dir.exists(): capabilities = list(capabilities_dir.iterdir()) print(f"✅ Capabilities: {len(capabilities)} found") if args.detailed: for cap in capabilities: if cap.is_dir(): print(f" • {cap.name}") else: print("❌ Capabilities directory not found") # Check ACP components acp_dir = current_path / "katamari_mcp" / "acp" if acp_dir.exists(): acp_files = list(acp_dir.glob("*.py")) print(f"✅ ACP Components: {len(acp_files)} found") if args.detailed: for file in acp_files: print(f" • {file.name}") else: print("❌ ACP directory not found") return 0 def _get_capability_template(self, template_type: str, name: str) -> str: """Get capability template content.""" if template_type == "basic": return f'''""" {name} Capability Basic capability template for Katamari MCP. """ from typing import Any, Dict, List from mcp.types import Tool class {name.title()}Capability: """Basic {name} capability implementation.""" def __init__(self): self.name = "{name}" self.version = "1.0.0" async def get_tools(self) -> List[Tool]: """Get available tools from this capability.""" return [ Tool( name="{name}_basic", description="Basic {name} functionality", inputSchema={{ "type": "object", "properties": {{ "input": {{ "type": "string", "description": "Input for {name} operation" }} }}, "required": ["input"] }} ) ] async def handle_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle a tool call.""" if tool_name == "{name}_basic": return await self._handle_basic(arguments.get("input")) else: raise ValueError(f"Unknown tool: {{tool_name}}") async def _handle_basic(self, input_data: str) -> Dict[str, Any]: """Handle basic {name} operation.""" # Basic processing implementation processed_data = input_data.strip().upper() if input_data else "" return {{ "success": True, "result": f"Processed: {{processed_data}}", "capability": self.name }} # Export the capability capability = {name.title()}Capability() ''' elif template_type == "advanced": return f'''""" {name} Capability (Advanced) Advanced capability template with error handling and logging. """ import logging from typing import Any, Dict, List, Optional from mcp.types import Tool logger = logging.getLogger(__name__) class {name.title()}Capability: """Advanced {name} capability with comprehensive features.""" def __init__(self): self.name = "{name}" self.version = "1.0.0" self.logger = logging.getLogger(f"capability.{{self.name}}") async def get_tools(self) -> List[Tool]: """Get available tools from this capability.""" return [ Tool( name="{name}_process", description="Process data with {name} capability", inputSchema={{ "type": "object", "properties": {{ "data": {{ "type": "string", "description": "Data to process" }}, "options": {{ "type": "object", "description": "Processing options" }} }}, "required": ["data"] }} ), Tool( name="{name}_validate", description="Validate input for {name} capability", inputSchema={{ "type": "object", "properties": {{ "input": {{ "type": "string", "description": "Input to validate" }} }}, "required": ["input"] }} ) ] async def handle_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle a tool call with error handling.""" try: self.logger.info(f"Handling tool call: {{tool_name}}") if tool_name == "{name}_process": return await self._handle_process(arguments) elif tool_name == "{name}_validate": return await self._handle_validate(arguments) else: raise ValueError(f"Unknown tool: {{tool_name}}") except Exception as e: self.logger.error(f"Error in {{tool_name}}: {{e}}") return {{ "success": False, "error": str(e), "tool": tool_name }} async def _handle_process(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle process operation.""" data = arguments.get("data") options = arguments.get("options", {{}}) # Advanced processing implementation self.logger.info(f"Processing data with options: {{options}}") processed_result = data if options.get("uppercase", False): processed_result = str(processed_result).upper() if options.get("reverse", False): processed_result = str(processed_result)[::-1] return {{ "success": True, "result": f"Processed: {{processed_result}}", "options_used": options, "capability": self.name }} async def _handle_validate(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle validation operation.""" input_data = arguments.get("input") # Basic validation implementation is_valid = bool(input_data and len(input_data.strip()) > 0) return {{ "success": True, "valid": is_valid, "input": input_data, "capability": self.name }} # Export the capability capability = {name.title()}Capability() ''' else: # web template return f'''""" {name} Capability (Web) Web-focused capability template with HTTP handling. """ import aiohttp from typing import Any, Dict, List, Optional from mcp.types import Tool class {name.title()}Capability: """Web-focused {name} capability.""" def __init__(self): self.name = "{name}" self.version = "1.0.0" self.session: Optional[aiohttp.ClientSession] = None async def __aenter__(self): """Async context manager entry.""" self.session = aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager exit.""" if self.session: await self.session.close() async def get_tools(self) -> List[Tool]: """Get available tools from this capability.""" return [ Tool( name="{name}_fetch", description="Fetch data from web API", inputSchema={{ "type": "object", "properties": {{ "url": {{ "type": "string", "description": "URL to fetch" }}, "method": {{ "type": "string", "enum": ["GET", "POST"], "default": "GET" }} }}, "required": ["url"] }} ) ] async def handle_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle a tool call.""" if tool_name == "{name}_fetch": return await self._handle_fetch(arguments) else: raise ValueError(f"Unknown tool: {{tool_name}}") async def _handle_fetch(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle web fetch operation.""" url = arguments.get("url") method = arguments.get("method", "GET") if not self.session: self.session = aiohttp.ClientSession() try: async with self.session.request(method, url) as response: content = await response.text() return {{ "success": True, "status": response.status, "content": content[:1000], # Limit content size "url": url, "capability": self.name }} except Exception as e: return {{ "success": False, "error": str(e), "url": url, "capability": self.name }} # Export the capability capability = {name.title()}Capability() ''' def _get_test_template(self, template_type: str, name: str) -> str: """Get test template content.""" return f'''""" Tests for {name} Capability Test suite for the {name} capability. """ import pytest import asyncio from katamari_mcp.capabilities.{name} import capability class Test{name.title()}: """Test class for {name} capability.""" @pytest.fixture async def cap(self): """Fixture providing the capability instance.""" return capability @pytest.mark.asyncio async def test_get_tools(self, cap): """Test getting available tools.""" tools = await cap.get_tools() assert len(tools) > 0 assert all(hasattr(tool, 'name') for tool in tools) @pytest.mark.asyncio async def test_basic_functionality(self, cap): """Test basic functionality.""" tools = await cap.get_tools() if tools: tool_name = tools[0].name # Test with valid input result = await cap.handle_call(tool_name, {{"input": "test"}}) assert result is not None @pytest.mark.asyncio async def test_error_handling(self, cap): """Test error handling.""" # Test with invalid tool name with pytest.raises(ValueError): await cap.handle_call("invalid_tool", {{}}) @pytest.mark.asyncio async def test_input_validation(self, cap): """Test input validation.""" tools = await cap.get_tools() if tools: tool_name = tools[0].name # Test with missing required input result = await cap.handle_call(tool_name, {{}}) # Should handle gracefully (either raise error or return error result) assert result is not None ''' def _get_readme_template(self, template_type: str, name: str) -> str: """Get README template content.""" return f'''# {name.title()} Capability {template_type.title()} capability for Katamari MCP. ## Description This capability provides {name} functionality for the Katamari MCP system. ## Features - Basic {name} operations - Error handling and validation - Comprehensive test suite ## Usage ### Tools Available 1. `{name}_basic` - Basic {name} functionality ### Example Usage ```python # Import the capability from katamari_mcp.capabilities.{name} import capability # Get available tools tools = await capability.get_tools() # Use a tool result = await capability.handle_call("{name}_basic", {{ "input": "example data" }}) ``` ## Development ### Running Tests ```bash katamari-dev test {name} ``` ### Validation ```bash katamari-dev validate {name} ``` ## Requirements - Python 3.9+ - Katamari MCP framework ## License MIT License ''' async def run(self, args: Optional[List[str]] = None) -> int: """Run the CLI with the given arguments.""" parser = self.create_parser() parsed_args = parser.parse_args(args) if not parsed_args.command: parser.print_help() return 1 try: if parsed_args.command == "validate": return await self.cmd_validate(parsed_args) elif parsed_args.command == "test": return await self.cmd_test(parsed_args) elif parsed_args.command == "list": return self.cmd_list(parsed_args) elif parsed_args.command == "create": return self.cmd_create(parsed_args) elif parsed_args.command == "check": return self.cmd_check(parsed_args) else: print(f"❌ Unknown command: {parsed_args.command}") return 1 except KeyboardInterrupt: print("\n⚠️ Operation cancelled by user") return 130 except Exception as e: print(f"❌ Error: {e}") return 1 async def main(): """Main entry point for the CLI.""" cli = DevCLI() return await cli.run() if __name__ == "__main__": import sys sys.exit(asyncio.run(main()))