smart_route_request
Route user requests to the optimal ACP agent using the ACP-MCP-Server, ensuring efficient processing and integration with MCP-compatible tools for enhanced AI agent communication.
Instructions
Intelligently route a request to the best ACP agent
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| input_text | Yes | ||
| mode | No | sync | |
| session_id | No | ||
| strategy | No | default |
Input Schema (JSON Schema)
{
"properties": {
"input_text": {
"title": "Input Text",
"type": "string"
},
"mode": {
"default": "sync",
"title": "Mode",
"type": "string"
},
"session_id": {
"default": null,
"title": "Session Id",
"type": "string"
},
"strategy": {
"default": "default",
"title": "Strategy",
"type": "string"
}
},
"required": [
"input_text"
],
"type": "object"
}
Implementation Reference
- acp_mcp_server/agent_router.py:190-211 (handler)The main handler implementation for the 'smart_route_request' tool. This async function, decorated with @mcp.tool(), handles the tool execution by calling the router's execute_routed_request method and returning a JSON-formatted result.@mcp.tool() async def smart_route_request( input_text: str, strategy: str = "default", mode: str = "sync", session_id: str = None ) -> str: """Intelligently route a request to the best ACP agent""" try: result = await router.execute_routed_request( input_text=input_text, strategy_name=strategy, mode=mode, session_id=session_id ) return json.dumps(result, indent=2) except Exception as e: return f"Error: {e}"
- acp_mcp_server/server.py:88-88 (registration)The registration point where register_router_tools is called on the FastMCP instance, which defines and registers the smart_route_request tool using the @mcp.tool() decorator.register_router_tools(self.mcp, self.router)
- Supporting helper method in AgentRouter class that performs the actual routing and execution logic invoked by the smart_route_request tool handler.async def execute_routed_request( self, input_text: str, strategy_name: str = "default", mode: str = "sync", session_id: Optional[str] = None ) -> Dict[str, Any]: """Route and execute a request""" try: # Route the request target_agent, routing_reason = await self.route_request(input_text, strategy_name) # Execute the agent if mode == "sync": run = await self.orchestrator.execute_agent_sync( target_agent, input_text, session_id ) result = { "routed_to": target_agent, "routing_reason": routing_reason, "execution_mode": mode, "status": run.status, "run_id": run.run_id } if run.output: # Handle ACP output format - run.output is already a list of messages output_text = "" for message in run.output: if isinstance(message, dict) and "parts" in message: for part in message["parts"]: if isinstance(part, dict) and "content" in part: output_text += part["content"] + "\n" result["output"] = output_text.strip() if output_text else "No text content" if run.error: result["error"] = run.error return result else: # Async mode run_id = await self.orchestrator.execute_agent_async( target_agent, input_text, session_id ) return { "routed_to": target_agent, "routing_reason": routing_reason, "execution_mode": mode, "run_id": run_id, "status": "async_started" } except Exception as e: return { "error": str(e), "routed_to": None, "routing_reason": "Routing failed" }