A2A MCP Server

Instructions

Call an agent with a prompt.

Args:
    agent_name: Name of the agent to call
    prompt: Prompt to send to the agent
    
Returns:
    Dict with response from the agent

Implementation Reference

• a2a_mcp_server/server.py:190-310 (handler)

  The core handler function for the 'call_agent' tool. It verifies the agent in the registry, fetches the agent card if necessary, connects to the A2A server using A2aMinClient, sends the prompt as a message, and returns the task response including artifacts.

  @mcp.tool()
  async def call_agent(agent_name: str, prompt: str) -> Dict[str, Any]:
      """
      Call an agent with a prompt.
      
      Args:
          agent_name: Name of the agent to call
          prompt: Prompt to send to the agent
          
      Returns:
          Dict with response from the agent
      """
      try:
          logger.debug(f"call_agent called with agent_name={agent_name}, prompt={prompt[:50]}...")
          logger.debug(f"Current registry: {state.registry}")
          
          # Verify the agent exists
          if agent_name not in state.registry:
              logger.warning(f"Agent '{agent_name}' not found in registry")
              return {
                  "status": "error",
                  "message": f"Agent '{agent_name}' not found in registry"
              }
          
          # Get the URL for the agent
          url = state.registry[agent_name]
          logger.debug(f"Using URL: {url}")
          
          try:
              # Get or create agent card if needed
              if agent_name not in state.cache:
                  logger.debug(f"Agent card not in cache, fetching for {agent_name}")
                  card = await fetch_agent_card(url)
                  if card:
                      state.cache[agent_name] = card
                  else:
                      logger.error(f"Failed to fetch agent card for {agent_name}")
                      return {
                          "status": "error",
                          "message": f"Failed to fetch agent card for {agent_name}"
                      }
              
              # Create a client using A2aMinClient
              try:
                  # Connect to the A2A server
                  logger.debug(f"Connecting to A2A server at {url}")
                  client = A2aMinClient.connect(url)
                  
                  # Prepare message and send
                  logger.debug(f"Sending message to agent {agent_name}: {prompt[:50]}...")
                  task_id = str(uuid.uuid4())
                  session_id = str(uuid.uuid4())
                  
                  # Send the message and get response
                  task = await client.send_message(
                      message=prompt,
                      task_id=task_id,
                      session_id=session_id
                  )
                  
                  # Process and return the response
                  logger.debug(f"Received task response: {task}")
                  
                  response_data = {
                      "status": "success",
                      "task_id": task.id,
                      "state": task.status.state if hasattr(task, 'status') else "unknown",
                      "artifacts": []
                  }
                  
                  # Extract artifacts
                  artifacts = task.artifacts if hasattr(task, 'artifacts') else []
                  if artifacts:
                      logger.debug(f"Received {len(artifacts)} artifacts")
                      for artifact in artifacts:
                          artifact_data = {
                              "name": artifact.name,
                              "content": []
                          }
                          
                          for part in artifact.parts:
                              part_type = part.type
                              if part_type == "text":
                                  artifact_data["content"].append({
                                      "type": "text", 
                                      "text": part.text
                                  })
                              elif part_type == "file":
                                  file_data = part.file
                                  artifact_data["content"].append({
                                      "type": "file", 
                                      "name": file_data.name,
                                      "mime_type": file_data.mimeType
                                  })
                              elif part_type == "data":
                                  artifact_data["content"].append({
                                      "type": "data", 
                                      "data": part.data
                                  })
                          
                          response_data["artifacts"].append(artifact_data)
                  
                  logger.info(f"Successfully called agent {agent_name}")
                  return response_data
                  
              except Exception as e:
                  logger.exception(f"Error using A2aMinClient: {e}")
                  return {
                      "status": "error",
                      "message": f"Error calling agent with A2aMinClient: {str(e)}"
                  }
          
          except Exception as e:
              logger.exception(f"Error calling agent {agent_name}: {e}")
              return {
                  "status": "error",
                  "message": f"Error calling agent: {str(e)}"
              }
      except Exception as e:
          logger.exception(f"Error in call_agent: {str(e)}")
          return {"status": "error", "message": f"Error: {str(e)}"}

• a2a_mcp_server/server.py:192-201 (schema)

  Input schema defined in the tool's docstring: agent_name (str), prompt (str). Output is Dict[str, Any] with status, task_id, state, artifacts.

  """
  Call an agent with a prompt.
  
  Args:
      agent_name: Name of the agent to call
      prompt: Prompt to send to the agent
      
  Returns:
      Dict with response from the agent
  """

• a2a_mcp_server/server.py:190-190 (registration)

  The @mcp.tool() decorator registers the call_agent function as an MCP tool.

  @mcp.tool()

• a2a_mcp_server/server.py:46-64 (helper)

  Helper function used by call_agent to fetch and cache the agent's card if not already cached.

  async def fetch_agent_card(url: str) -> Optional[AgentCard]:
      """Fetch agent card from an A2A server using A2ACardResolver."""
      try:
          logger.debug(f"Fetching agent card from {url}")
          
          # Use the A2ACardResolver from a2a_min to get the agent card
          card_resolver = A2ACardResolver(url)
          try:
              card = card_resolver.get_agent_card()
              logger.debug(f"Received agent card: {card}")
              return card
          except Exception as e:
              logger.error(f"Failed to fetch agent card: {str(e)}")
              return None
                  
      except Exception as e:
          logger.exception(f"Error fetching agent card from {url}: {e}")
          return None