A2A MCP Server

Instructions

Implementation Reference

• a2a_mcp_server.py:310-348 (handler)

  The core handler function for the 'register_agent' MCP tool. It fetches the agent card, creates AgentInfo, stores it in the global registered_agents dictionary, persists to JSON file, and returns success/error status. Registered via @mcp.tool() decorator.
  @mcp.tool() async def register_agent(url: str, ctx: Context) -> Dict[str, Any]: """ Register an A2A agent with the bridge server. Args: url: URL of the A2A agent Returns: Dictionary with registration status """ try: # Fetch the agent card directly agent_card = await fetch_agent_card(url) # Store the agent information agent_info = AgentInfo( url=url, name=agent_card.name, description=agent_card.description or "No description provided", ) registered_agents[url] = agent_info # Save to disk immediately agents_data = {url: agent.model_dump() for url, agent in registered_agents.items()} save_to_json(agents_data, REGISTERED_AGENTS_FILE) await ctx.info(f"Successfully registered agent: {agent_card.name}") return { "status": "success", "agent": agent_info.model_dump(), } except Exception as e: return { "status": "error", "message": f"Failed to register agent: {str(e)}", }
• a2a_mcp_server.py:262-307 (helper)

  Supporting function that fetches the AgentCard from the provided agent URL, trying both the root endpoint and /.well-known/agent.json, with fallback to a minimal AgentCard. Called directly within the register_agent handler.
  async def fetch_agent_card(url: str) -> AgentCard: """ Fetch the agent card from the agent's URL. First try the main URL, then the well-known location. """ async with httpx.AsyncClient() as client: # First try the main endpoint try: response = await client.get(url) if response.status_code == 200: try: data = response.json() if isinstance(data, dict) and "name" in data and "url" in data: return AgentCard(**data) except json.JSONDecodeError: pass # Not a valid JSON response, try the well-known URL except Exception: pass # Connection error, try the well-known URL # Try the well-known location well_known_url = f"{url.rstrip('/')}/.well-known/agent.json" try: response = await client.get(well_known_url) if response.status_code == 200: try: data = response.json() return AgentCard(**data) except json.JSONDecodeError: raise ValueError(f"Invalid JSON in agent card from {well_known_url}") except httpx.RequestError as e: raise ValueError(f"Failed to fetch agent card from {well_known_url}: {str(e)}") # If we can't get the agent card, create a minimal one with default values return AgentCard( name="Unknown Agent", url=url, version="0.1.0", capabilities=AgentCapabilities(streaming=False), skills=[ AgentSkill( id="unknown", name="Unknown Skill", description="Unknown agent capabilities", ) ], )
• a2a_mcp_server.py:118-123 (schema)

  Pydantic BaseModel used to structure and validate agent information stored by register_agent. Includes URL, name, and description fields with descriptions for MCP tool schema inference.
  class AgentInfo(BaseModel): """Information about an A2A agent.""" url: str = Field(description="URL of the A2A agent") name: str = Field(description="Name of the A2A agent") description: str = Field(description="Description of the A2A agent")
• a2a_mcp_server.py:310-310 (registration)

  The @mcp.tool() decorator registers the register_agent function as an MCP tool with FastMCP, making it available via the MCP protocol.
  @mcp.tool()