a2a_server_registry
Manage A2A server URLs by adding or removing entries to configure communication with external agents through the MCP protocol.
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| action | Yes | ||
| name | Yes | ||
| url | No |
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