A2A MCP Server

Instructions

Implementation Reference

• a2a_mcp_server.py:416-544 (handler)

  The handler function for the 'send_message' MCP tool. Decorated with @mcp.tool() which registers it automatically in FastMCP. Sends a message to a registered A2A agent via A2AClient.send_task, generates task_id, stores mapping, extracts and returns response details including state, message, and artifacts.

  @mcp.tool()
  async def send_message(
      agent_url: str,
      message: str,
      session_id: Optional[str] = None,
      ctx: Context = None,
  ) -> Dict[str, Any]:
      """
      Send a message to an A2A agent.
      
      Args:
          agent_url: URL of the A2A agent
          message: Message to send
          session_id: Optional session ID for multi-turn conversations
          
      Returns:
          Agent's response with task_id for future reference
      """
      if agent_url not in registered_agents:
          return {
              "status": "error",
              "message": f"Agent not registered: {agent_url}",
          }
      
      # Create a client for the agent
      client = A2AClient(url=agent_url)
      
      try:
          # Generate a task ID
          task_id = str(uuid.uuid4())
          
          # Store the mapping of task_id to agent_url for later reference
          task_agent_mapping[task_id] = agent_url
          
          # Create the message
          a2a_message = Message(
              role="user",
              parts=[TextPart(text=message)],
          )
          
          if ctx:
              await ctx.info(f"Sending message to agent: {message}")
          
          # Create payload as a single dictionary
          payload = {
              "id": task_id,
              "message": a2a_message,
          }
          if session_id:
              payload["sessionId"] = session_id
          
          # Send the task with the payload
          result = await client.send_task(payload)
          
          # Save task mapping to disk
          save_to_json(task_agent_mapping, TASK_AGENT_MAPPING_FILE)
          
          # Debug: Print the raw response for analysis
          if ctx:
              await ctx.info(f"Raw response: {result}")
              
          # Create a response dictionary with as much info as we can extract
          response = {
              "status": "success",
              "task_id": task_id,
          }
          
          # Add any available fields from the result
          if hasattr(result, "sessionId"):
              response["session_id"] = result.sessionId
          else:
              response["session_id"] = None
              
          # Try to get the state
          try:
              if hasattr(result, "status") and hasattr(result.status, "state"):
                  response["state"] = result.status.state
              else:
                  response["state"] = "unknown"
          except Exception as e:
              response["state"] = f"error_getting_state: {str(e)}"
              
          # Try to extract response message
          try:
              if hasattr(result, "status") and hasattr(result.status, "message") and result.status.message:
                  response_text = ""
                  for part in result.status.message.parts:
                      if part.type == "text":
                          response_text += part.text
                  if response_text:
                      response["message"] = response_text
          except Exception as e:
              response["message_error"] = f"Error extracting message: {str(e)}"
          
          # Try to get artifacts
          try:
              if hasattr(result, "artifacts") and result.artifacts:
                  artifacts_data = []
                  for artifact in result.artifacts:
                      artifact_data = {
                          "name": artifact.name if hasattr(artifact, "name") else "unnamed_artifact",
                          "contents": [],
                      }
                      
                      for part in artifact.parts:
                          if part.type == "text":
                              artifact_data["contents"].append({
                                  "type": "text",
                                  "text": part.text,
                              })
                          elif part.type == "data":
                              artifact_data["contents"].append({
                                  "type": "data",
                                  "data": part.data,
                              })
                      
                      artifacts_data.append(artifact_data)
                  
                  response["artifacts"] = artifacts_data
          except Exception as e:
              response["artifacts_error"] = f"Error extracting artifacts: {str(e)}"
              
          return response
      except Exception as e:
          return {
              "status": "error",
              "message": f"Error sending message: {str(e)}",
          }