A2A MCP Server

Instructions

Implementation Reference

• a2a_mcp_server.py:546-682 (handler)

  The primary handler for the 'get_task_result' MCP tool. Decorated with @mcp.tool() for registration. Retrieves task results from A2A agents by looking up the agent URL via task_agent_mapping, calling A2AClient.get_task(), and parsing the response into a structured dictionary containing status, session_id, state, message, artifacts, and history.
  @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)}", }