Spaces:
Running
Running
| """ | |
| Query Router - Stage 1 | |
| Simple router that always queries all MCP servers for alert generation. | |
| The intelligence is in the compiler, not the router. | |
| """ | |
| from typing import Dict, Any | |
| class QueryRouter: | |
| """ | |
| Router for alert generation system. | |
| For alert generation, ALWAYS queries all MCP servers since we need | |
| comprehensive data to identify potential issues. The compiler handles | |
| the intelligence of extracting only alerting/concerning information. | |
| """ | |
| def __init__(self): | |
| """Initialize router - no LLM needed for simple all-server routing""" | |
| pass | |
| def route_alert_query(self, location: Dict[str, float]) -> Dict[str, bool]: | |
| """ | |
| Route for alert generation - always query ALL servers. | |
| Args: | |
| location: Dict with 'latitude' and 'longitude' keys | |
| Returns: | |
| Dict with all servers set to True | |
| """ | |
| return { | |
| "weather": True, | |
| "soil": True, | |
| "water": True, | |
| "elevation": True, | |
| "pests": True | |
| } | |
| def route_query(self, query: str, location: Dict[str, float]) -> Dict[str, bool]: | |
| """ | |
| Route a general query - for now, also queries all servers. | |
| In the future, could use LLM to determine which servers are relevant | |
| to the specific query. But for alert generation, we always want all data. | |
| Args: | |
| query: User's query | |
| location: Dict with 'latitude' and 'longitude' keys | |
| Returns: | |
| Dict indicating which servers to query | |
| """ | |
| # For now, query all servers for any query | |
| # The compiler will extract relevant information | |
| return self.route_alert_query(location) |