A2A MCP Server

Instructions

Implementation Reference

• a2a_mcp_server.py:310-348 (handler)

  The @mcp.tool() decorated function that implements and registers the register_agent tool. It fetches the agent card, stores agent info in a global dict, saves to disk, and returns success/error status.
  @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)

  Helper function called by register_agent to fetch and parse the AgentCard from the agent's URL (main or /.well-known/agent.json), falling back to a minimal card if fetch fails.
  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 store parsed agent information (url, name, description) in the registered_agents dictionary.
  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")