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

TableJSON Schema

Name	Required	Default
`input_text`	Yes
`mode`	No	sync
`session_id`	No
`strategy`	No	default

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)
```

acp_mcp_server/agent_router.py:120-185 (helper)

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"
        }

ACP-MCP-Server