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