A2A MCP Server

Instructions

Implementation Reference

• a2a_mcp_server.py:546-682 (handler)

  This is the main handler function for the 'get_task_result' MCP tool. It is decorated with @mcp.tool(), making it the registered tool implementation. The function looks up the agent URL for the given task_id from a global task_agent_mapping, creates an A2AClient, calls client.get_task() to fetch the result from the A2A agent, and parses the response to extract status, message, artifacts, and history into a dictionary response.

  @mcp.tool()
  async def get_task_result(
      task_id: str,
      history_length: Optional[int] = None,
      ctx: Context = None,
  ) -> Dict[str, Any]:
      """
      Retrieve the result of a task from an A2A agent.
      
      Args:
          task_id: ID of the task to retrieve
          history_length: Optional number of history items to include (null for all)
          
      Returns:
          Task result including status, message, and artifacts if available
      """
      if task_id not in task_agent_mapping:
          return {
              "status": "error",
              "message": f"Task ID not found: {task_id}",
          }
      
      agent_url = task_agent_mapping[task_id]
      
      # Create a client for the agent
      client = A2AClient(url=agent_url)
      
      try:
          # Create the request payload
          payload = {
              "id": task_id,
              "historyLength": history_length
          }
          
          if ctx:
              await ctx.info(f"Retrieving task result for task_id: {task_id}")
          
          # Send the get task request
          result = await client.get_task(payload)
          
          # Debug: Print the raw response for analysis
          if ctx:
              await ctx.info(f"Raw task result: {result}")
              
          # Create a response dictionary with as much info as we can extract
          response = {
              "status": "success",
              "task_id": task_id,
          }
          
          # Try to extract task data
          try:
              if hasattr(result, "result"):
                  task = result.result
                  
                  # Add basic task info
                  if hasattr(task, "sessionId"):
                      response["session_id"] = task.sessionId
                  else:
                      response["session_id"] = None
                  
                  # Add task status
                  if hasattr(task, "status"):
                      status = task.status
                      if hasattr(status, "state"):
                          response["state"] = status.state
                      
                      # Extract message from status
                      if hasattr(status, "message") and status.message:
                          response_text = ""
                          for part in status.message.parts:
                              if part.type == "text":
                                  response_text += part.text
                          if response_text:
                              response["message"] = response_text
                  
                  # Extract artifacts
                  if hasattr(task, "artifacts") and task.artifacts:
                      artifacts_data = []
                      for artifact in task.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
                  
                  # Extract message history if available
                  if hasattr(task, "history") and task.history:
                      history_data = []
                      for message in task.history:
                          message_data = {
                              "role": message.role,
                              "parts": [],
                          }
                          
                          for part in message.parts:
                              if part.type == "text":
                                  message_data["parts"].append({
                                      "type": "text",
                                      "text": part.text,
                                  })
                              elif hasattr(part, "data"):
                                  message_data["parts"].append({
                                      "type": "data",
                                      "data": part.data,
                                  })
                          
                          history_data.append(message_data)
                      
                      response["history"] = history_data
              else:
                  response["error"] = "No result in response"
                  
          except Exception as e:
              response["parsing_error"] = f"Error parsing task result: {str(e)}"
              
          return response
      except Exception as e:
          return {
              "status": "error",
              "message": f"Error retrieving task result: {str(e)}",
          }