ACP-MCP-Server

Instructions

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