list_pending_interactions
Identify and list all interactive agents awaiting input within the ACP-MCP-Server, enabling efficient management of pending interactions in ACP-MCP workflows.
Instructions
List all pending interactive agents waiting for input
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Input Schema (JSON Schema)
{
"properties": {},
"type": "object"
}
Implementation Reference
- MCP tool handler for 'list_pending_interactions'. Decorated with @mcp.tool() for registration. Delegates to InteractiveManager.get_pending_interactions() and returns formatted JSON.@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}"
- Core helper method implementing the logic to list pending interactions: filters active ones, handles timeouts, cleans up expired, returns structured dict.async def get_pending_interactions(self) -> Dict[str, Any]: """Get all pending interactions""" current_time = asyncio.get_event_loop().time() active_interactions = {} expired_interactions = [] for run_id, pending in self.pending_interactions.items(): if current_time - pending.timestamp > pending.timeout_seconds: expired_interactions.append(run_id) else: active_interactions[run_id] = { "agent_name": pending.agent_name, "message": pending.await_message, "waiting_seconds": int(current_time - pending.timestamp), "timeout_in_seconds": int(pending.timeout_seconds - (current_time - pending.timestamp)) } # Clean up expired interactions for run_id in expired_interactions: del self.pending_interactions[run_id] return { "active_count": len(active_interactions), "expired_count": len(expired_interactions), "interactions": active_interactions }
- acp_mcp_server/server.py:89-89 (registration)Invocation of register_interactive_tools during server initialization, which registers the list_pending_interactions tool with FastMCP.register_interactive_tools(self.mcp, self.interactive_manager)
- Pydantic schema for PendingInteraction used by InteractiveManager to store and manage pending interaction data.class PendingInteraction(BaseModel): run_id: str agent_name: str session_id: Optional[str] await_message: str timestamp: float timeout_seconds: int = 300 # 5 minutes default