A2A MCP Server

Instructions

Add or remove an A2A server URL.

Args:
    action: Either "add" or "remove"
    name: Name of the server
    url: URL of the server (required for "add" action)
    
Returns:
    Dict with status and message

Implementation Reference

• a2a_mcp_server/server.py:85-149 (handler)

  Main handler function for the 'a2a_server_registry' tool. Handles adding or removing A2A server URLs from the registry, updates cache asynchronously, and logs operations. Includes input validation and error handling.

  @mcp.tool()
  async def a2a_server_registry(action: Literal["add", "remove"], name: str, url: Optional[str] = None) -> Dict[str, Any]:
      """
      Add or remove an A2A server URL.
      
      Args:
          action: Either "add" or "remove"
          name: Name of the server
          url: URL of the server (required for "add" action)
          
      Returns:
          Dict with status and message
      """
      try:
          logger.debug(f"a2a_server_registry called with action={action}, name={name}, url={url}")
          logger.debug(f"Registry before: {state.registry}")
          
          if action == "add":
              if not url:
                  logger.warning("URL is required for add action")
                  return {"status": "error", "message": "URL is required for add action"}
              
              # Add to registry
              state.registry[name] = url
              logger.info(f"Added A2A server: {name} -> {url}")
              logger.debug(f"Registry after: {state.registry}")
              
              # Update the cache asynchronously
              asyncio.create_task(update_agent_cache())
              
              return {
                  "status": "success", 
                  "message": f"Added A2A server: {name}",
                  "registry": state.registry
              }
          
          elif action == "remove":
              if name in state.registry:
                  # Remove from registry
                  del state.registry[name]
                  
                  # Remove from cache if present
                  if name in state.cache:
                      del state.cache[name]
                  
                  logger.info(f"Removed A2A server: {name}")
                  logger.debug(f"Registry after: {state.registry}")
                  
                  return {
                      "status": "success", 
                      "message": f"Removed A2A server: {name}",
                      "registry": state.registry
                  }
              else:
                  logger.warning(f"Server {name} not found in registry")
                  return {
                      "status": "error", 
                      "message": f"Server {name} not found in registry"
                  }
          
          logger.error(f"Invalid action: {action}")
          return {"status": "error", "message": "Invalid action. Use 'add' or 'remove'"}
      except Exception as e:
          logger.exception(f"Error in a2a_server_registry: {str(e)}")
          return {"status": "error", "message": f"Error: {str(e)}"}

• a2a_mcp_server/server.py:85-85 (registration)

  MCP tool registration decorator for a2a_server_registry.

  @mcp.tool()

• a2a_mcp_server/server.py:86-97 (schema)

  Function signature and docstring defining the input schema (action, name, url) and output (dict with status/message/registry).

  async def a2a_server_registry(action: Literal["add", "remove"], name: str, url: Optional[str] = None) -> Dict[str, Any]:
      """
      Add or remove an A2A server URL.
      
      Args:
          action: Either "add" or "remove"
          name: Name of the server
          url: URL of the server (required for "add" action)
          
      Returns:
          Dict with status and message
      """

• a2a_mcp_server/server.py:37-41 (helper)

  Persistent state class holding the registry (name to URL) and agent card cache, used by the a2a_server_registry tool.

  class ServerState:
      def __init__(self):
          self.registry = {}  # name -> url
          self.cache = {}  # name -> AgentCard