a2a_server_registry
Manage A2A server URLs by adding or removing them using specified actions; includes name and URL inputs to facilitate communication between Claude Desktop and A2A protocol agents.
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)The handler function implementing the a2a_server_registry tool. It handles 'add' and 'remove' actions for A2A server URLs, updates the persistent registry state, logs operations, triggers agent cache updates, and returns status messages.@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)The @mcp.tool() decorator registers the a2a_server_registry function as an MCP tool.@mcp.tool()
- a2a_mcp_server/server.py:37-41 (helper)Persistent state class storing the registry (name to URL mapping) used by the a2a_server_registry tool.class ServerState: def __init__(self): self.registry = {} # name -> url self.cache = {} # name -> AgentCard
- a2a_mcp_server/server.py:66-79 (helper)Helper function to update the agent card cache after registry changes, called by a2a_server_registry.async def update_agent_cache() -> None: """Update the cache of agent cards.""" logger.info("Updating agent cache...") new_cache = {} for name, url in state.registry.items(): card = await fetch_agent_card(url) if card: new_cache[name] = card logger.info(f"Added agent {name} to cache: {card.name}") # Update the state cache state.cache = new_cache logger.info(f"Agent cache updated with {len(state.cache)} agents")
- a2a_mcp_server/server.py:87-97 (schema)Docstring providing schema/documentation for the tool's inputs and outputs.""" 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 """