ACP-MCP-Server

Instructions

Implementation Reference

• acp_mcp_server/interactive_manager.py:271-282 (handler)

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

• acp_mcp_server/interactive_manager.py:195-222 (helper)

  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)

• acp_mcp_server/interactive_manager.py:17-24 (schema)

  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