simple_router.py•4.33 kB
"""
Simple router implementation for MVP - no ML dependencies.
"""
import logging
from typing import Dict, List, Any, Optional
from katamari_mcp.capabilities.web_search import WebSearchCapability
from katamari_mcp.capabilities.web_scrape import WebScrapeCapability
logger = logging.getLogger(__name__)
class SimpleRouter:
"""Simple router that directly maps tools to capabilities without ML."""
def __init__(self):
self.capabilities = {
"web_search": WebSearchCapability(),
"web_scrape": WebScrapeCapability(),
}
# Tool definitions for MCP
self.tool_definitions = {
"web_search": {
"name": "web_search",
"description": "Search the web without API tokens using DuckDuckGo and Brave Search",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
},
"max_results": {
"type": "integer",
"description": "Maximum number of results (default: 5)",
"default": 5
}
},
"required": ["query"]
}
},
"web_scrape": {
"name": "web_scrape",
"description": "Extract and convert web content to markdown",
"input_schema": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "URL to scrape"
}
},
"required": ["url"]
}
}
}
async def list_capabilities(self) -> List[Dict[str, Any]]:
"""List all available capabilities."""
return list(self.tool_definitions.values())
async def route_call(self, tool_name: str, arguments: Dict[str, Any]) -> str:
"""Route tool call to appropriate capability."""
if tool_name not in self.capabilities:
raise ValueError(f"Unknown tool: {tool_name}")
capability = self.capabilities[tool_name]
try:
if tool_name == "web_search":
query = arguments.get("query")
max_results = arguments.get("max_results", 5)
if not query:
raise ValueError("Query parameter is required for web_search")
search_result = await capability.search(query, max_results)
results = search_result.get('results', []) if isinstance(search_result, dict) else search_result
return f"Search results for '{query}':\n\n{self._format_search_results(results)}"
elif tool_name == "web_scrape":
url = arguments.get("url")
if not url:
raise ValueError("URL parameter is required for web_scrape")
content = await capability.scrape(url)
return f"Content from {url}:\n\n{content}"
else:
raise ValueError(f"Tool {tool_name} not implemented")
except Exception as e:
logger.error(f"Error executing {tool_name}: {e}")
raise
def _format_search_results(self, results: List[Dict[str, Any]]) -> str:
"""Format search results for display."""
if not results:
return "No results found."
formatted = []
for i, result in enumerate(results, 1):
formatted.append(f"{i}. **{result.get('title', 'No title')}**")
formatted.append(f" {result.get('url', 'No URL')}")
formatted.append(f" {result.get('snippet', 'No description')}")
formatted.append("")
return "\n".join(formatted)
async def format_error(self, error: Exception) -> str:
"""Format error for user display."""
return f"Error: {str(error)}"