Agent MCP Gateway

Instructions

Implementation Reference

• src/gateway.py:164-222 (handler)

  The core handler function for the 'list_servers' MCP tool. Registered via @gateway.tool decorator on the FastMCP instance. Retrieves policy-allowed servers from ProxyManager, constructs ServerInfo objects based on include_metadata flag, and returns serialized list.
  @gateway.tool async def list_servers( agent_id: Annotated[Optional[str], "Your agent identifier (leave empty if not provided to you)"] = None, include_metadata: Annotated[bool, "Include technical details (transport, command, url)"] = False ) -> list[dict]: """Discover downstream MCP servers available through this gateway. Your access is determined by gateway policy rules. Workflow: 1) Call list_servers to discover servers, 2) Call get_server_tools to see available tools, 3) Call execute_tool to use them.""" # Defensive check (middleware should have resolved agent_id) if agent_id is None: raise ToolError("Internal error: agent_id not resolved by middleware") # Get configurations from module-level storage policy_engine = _policy_engine proxy_manager = _proxy_manager if not policy_engine: raise RuntimeError("PolicyEngine not initialized in gateway state") if not proxy_manager: raise RuntimeError("ProxyManager not initialized in gateway state") # Get servers this agent can access allowed_servers = policy_engine.get_allowed_servers(agent_id) # Get current server list from ProxyManager (reflects hot-reload changes) all_servers = proxy_manager.get_servers_config() # Build response server_list = [] # Handle wildcard access if allowed_servers == ["*"]: # Agent has wildcard access - return all servers allowed_servers = list(all_servers.keys()) for server_name in allowed_servers: if server_name in all_servers: server_config = all_servers[server_name] # Determine transport type transport = "stdio" if "command" in server_config else "http" # Build ServerInfo object - always include name and description server_info_kwargs = { "name": server_name, "description": server_config.get("description") # Include description always (None if not in config) } # Add technical metadata if requested if include_metadata: server_info_kwargs["transport"] = transport # Add transport-specific metadata if transport == "stdio": server_info_kwargs["command"] = server_config.get("command") elif transport == "http": server_info_kwargs["url"] = server_config.get("url") server_list.append(ServerInfo(**server_info_kwargs)) return [server.model_dump() for server in server_list]
• src/gateway.py:15-21 (schema)

  Pydantic BaseModel defining the structured output schema for the list_servers tool response. Used to serialize server information including name, description, and optional metadata.
  class ServerInfo(BaseModel): """Server information returned by list_servers.""" name: Annotated[str, Field(description="Server name (use in get_server_tools and execute_tool)")] description: Annotated[Optional[str], Field(description="What this server provides (from config or null if not configured)")] = None transport: Annotated[Optional[str], Field(description="How server communicates: stdio or http (only if include_metadata=true)")] = None command: Annotated[Optional[str], Field(description="Command that runs this server (only if include_metadata=true and transport=stdio)")] = None url: Annotated[Optional[str], Field(description="Server endpoint (only if include_metadata=true and transport=http)")] = None
• src/gateway.py:164-164 (registration)

  The @gateway.tool decorator registers the list_servers function as an MCP tool on the FastMCP 'gateway' instance.
  @gateway.tool