register_agent

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")