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
| Name | Required | Description | Default |
|---|---|---|---|
| agent_name | Yes | ||
| input_text | Yes | ||
| mode | No | sync | |
| session_id | No |
Input Schema (JSON Schema)
{
"properties": {
"agent_name": {
"title": "Agent Name",
"type": "string"
},
"input_text": {
"title": "Input Text",
"type": "string"
},
"mode": {
"default": "sync",
"title": "Mode",
"type": "string"
},
"session_id": {
"default": null,
"title": "Session Id",
"type": "string"
}
},
"required": [
"agent_name",
"input_text"
],
"type": "object"
}
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)