run_acp_agent
Run an ACP agent with specified input text and mode to enable communication between ACP-based AI agents and MCP-compatible tools via the ACP-MCP-Server.
Instructions
Execute an ACP agent with specified mode
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| agent_name | Yes | ||
| input_text | Yes | ||
| mode | No | sync | |
| session_id | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The core handler function for the 'run_acp_agent' tool. Decorated with @mcp.tool(), it orchestrates ACP agent execution in sync or async modes using the RunOrchestrator class.
async def run_acp_agent( agent_name: str, input_text: str, mode: str = "sync", session_id: str = None ) -> str: """Execute an ACP agent with specified mode""" try: if mode == "sync": run = await orchestrator.execute_agent_sync(agent_name, input_text, session_id) if run.status == RunStatus.COMPLETED: # Convert output back to readable format if run.output: # Handle ACP output format 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" return output_text.strip() if output_text else "Agent completed with no text output" else: return "Agent completed with no output" else: return f"Error: {run.error}" elif mode == "async": run_id = await orchestrator.execute_agent_async(agent_name, input_text, session_id) return f"Started async run with ID: {run_id}" else: return f"Unsupported mode: {mode}. Use 'sync' or 'async'" except Exception as e: return f"Error: {e}" - acp_mcp_server/server.py:87-87 (registration)Registration of the orchestrator tools, including 'run_acp_agent', by calling register_orchestrator_tools on the FastMCP instance.
register_orchestrator_tools(self.mcp, self.orchestrator)