start_interactive_agent
Initiate an interactive ACP agent session, enabling user input integration and communication via the ACP-MCP Server for AI agent workflows with customizable timeouts.
Instructions
Start an interactive ACP agent that may require user input
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| agent_name | Yes | ||
| initial_input | Yes | ||
| session_id | No | ||
| timeout_minutes | No |
Implementation Reference
- FastMCP @tool decorator and handler function for 'start_interactive_agent' that invokes the InteractiveManager method and serializes the result to JSON.@mcp.tool() async def start_interactive_agent( agent_name: str, initial_input: str, session_id: str = None, timeout_minutes: int = 5 ) -> str: """Start an interactive ACP agent that may require user input""" try: result = await manager.start_interactive_agent( agent_name=agent_name, initial_input=initial_input, session_id=session_id, timeout_seconds=timeout_minutes * 60 ) return json.dumps(result, indent=2) except Exception as e: return f"Error: {e}"
- Core logic for starting an interactive agent in the InteractiveManager class, handling execution, pending interactions, and output processing.async def start_interactive_agent( self, agent_name: str, initial_input: str, session_id: Optional[str] = None, timeout_seconds: int = 300 ) -> Dict[str, Any]: """Start an interactive agent that may require user input""" try: # Start the agent execution run = await self.orchestrator.execute_agent_sync( agent_name=agent_name, input_text=initial_input, session_id=session_id ) # Check if agent is waiting for input if hasattr(run, 'await_request') and run.await_request: # Agent is waiting for input pending = PendingInteraction( run_id=run.run_id, agent_name=agent_name, session_id=session_id, await_message=run.await_request.get('message', 'Agent is waiting for input'), timestamp=asyncio.get_event_loop().time(), timeout_seconds=timeout_seconds ) self.pending_interactions[run.run_id] = pending return { "status": "awaiting_input", "run_id": run.run_id, "message": pending.await_message, "timeout_seconds": timeout_seconds } else: # Agent completed normally output = "" 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" output = output_text.strip() if output_text else "No text content" return { "status": "completed", "run_id": run.run_id, "output": output, "error": run.error } except Exception as e: return { "status": "error", "error": str(e) }
- acp_mcp_server/server.py:89-89 (registration)Invocation of register_interactive_tools which defines and registers the MCP tool 'start_interactive_agent' along with related interactive tools.register_interactive_tools(self.mcp, self.interactive_manager)
- acp_mcp_server/interactive_manager.py:232-297 (registration)Registration function that defines and registers the interactive tools including 'start_interactive_agent' using @mcp.tool() decorators.def register_interactive_tools(mcp: FastMCP, manager: InteractiveManager): @mcp.tool() async def start_interactive_agent( agent_name: str, initial_input: str, session_id: str = None, timeout_minutes: int = 5 ) -> str: """Start an interactive ACP agent that may require user input""" try: result = await manager.start_interactive_agent( agent_name=agent_name, initial_input=initial_input, session_id=session_id, timeout_seconds=timeout_minutes * 60 ) return json.dumps(result, indent=2) except Exception as e: return f"Error: {e}" @mcp.tool() async def provide_user_input( run_id: str, user_input: str ) -> str: """Provide user input to resume a waiting interactive agent""" try: result = await manager.resume_interactive_agent(run_id, user_input) return json.dumps(result, indent=2) except Exception as e: return f"Error: {e}" @mcp.tool() async def list_pending_interactions() -> str: """List all pending interactive agents waiting for input""" try: interactions = await manager.get_pending_interactions() return json.dumps(interactions, indent=2) except Exception as e: return f"Error: {e}" @mcp.tool() async def cancel_interaction(run_id: str) -> str: """Cancel a pending interactive agent""" try: success = await manager.cancel_interaction(run_id) if success: return f"Successfully cancelled interaction: {run_id}" else: return f"No pending interaction found with ID: {run_id}" except Exception as e: return f"Error: {e}"