ACP-MCP-Server

Instructions

Implementation Reference

• acp_mcp_server/interactive_manager.py:234-255 (handler)

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

• acp_mcp_server/interactive_manager.py:31-94 (helper)

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