Agent MCP Gateway

Instructions

Implementation Reference

• src/gateway.py:291-431 (handler)

  The main handler function for the 'get_server_tools' MCP tool. Retrieves tools from the specified downstream server via ProxyManager, applies policy-based filtering using PolicyEngine, supports filtering by explicit names, wildcard patterns, and token budgets, and returns a structured response with available tools.

  @gateway.tool
  async def get_server_tools(
      agent_id: Annotated[Optional[str], "Your agent identifier (leave empty if not provided to you)"] = None,
      server: Annotated[str, "Server name from list_servers"] = "",
      names: Annotated[Optional[str], "Filter: comma-separated tool names"] = None,
      pattern: Annotated[Optional[str], "Filter: wildcard pattern (e.g., 'get_*')"] = None,
      max_schema_tokens: Annotated[Optional[int], "Limit total tokens in returned schemas"] = None
  ) -> dict:
      """Discover tools available on a downstream MCP server accessed through this gateway. Returns only tools you have permission to use (filtered by policy rules). Use the returned tool definitions to call execute_tool."""
      # Defensive check (middleware should have resolved agent_id)
      if agent_id is None:
          raise ToolError("Internal error: agent_id not resolved by middleware")
  
      # Check for config changes (fallback mechanism for when file watching doesn't work)
      if _check_config_changes_fn:
          try:
              _check_config_changes_fn()
          except Exception:
              pass  # Don't let config check errors break tool execution
  
      # Parse comma-separated names string into list
      names_list: Optional[list[str]] = None
      if names is not None and names.strip():
          # Split by comma and trim whitespace from each name
          names_list = [name.strip() for name in names.split(",") if name.strip()]
          # If we ended up with an empty list after filtering, treat as None
          if not names_list:
              names_list = None
  
      # Get configurations from module-level storage
      policy_engine = _policy_engine
      proxy_manager = _proxy_manager
  
      if not policy_engine:
          return GetServerToolsResponse(
              tools=[],
              server=server,
              total_available=0,
              returned=0,
              tokens_used=None,
              error="PolicyEngine not initialized in gateway state"
          ).model_dump()
  
      if not proxy_manager:
          return GetServerToolsResponse(
              tools=[],
              server=server,
              total_available=0,
              returned=0,
              tokens_used=None,
              error="ProxyManager not initialized in gateway state"
          ).model_dump()
  
      # Validate agent can access server
      if not policy_engine.can_access_server(agent_id, server):
          return GetServerToolsResponse(
              tools=[],
              server=server,
              total_available=0,
              returned=0,
              tokens_used=None,
              error=f"Access denied: Agent '{agent_id}' cannot access server '{server}'"
          ).model_dump()
  
      # Get tools from downstream server
      try:
          all_tools = await proxy_manager.list_tools(server)
      except KeyError:
          return GetServerToolsResponse(
              tools=[],
              server=server,
              total_available=0,
              returned=0,
              tokens_used=None,
              error=f"Server '{server}' not found in configured servers"
          ).model_dump()
      except RuntimeError as e:
          return GetServerToolsResponse(
              tools=[],
              server=server,
              total_available=0,
              returned=0,
              tokens_used=None,
              error=f"Server unavailable: {str(e)}"
          ).model_dump()
      except Exception as e:
          return GetServerToolsResponse(
              tools=[],
              server=server,
              total_available=0,
              returned=0,
              tokens_used=None,
              error=f"Failed to retrieve tools: {str(e)}"
          ).model_dump()
  
      total_available = len(all_tools)
  
      # Filter tools based on criteria
      filtered_tools = []
      token_count = 0
  
      for tool in all_tools:
          tool_name = tool.name if hasattr(tool, 'name') else str(tool)
  
          # Filter by explicit names list
          if names_list is not None and tool_name not in names_list:
              continue
  
          # Filter by wildcard pattern
          if pattern is not None and not _matches_pattern(tool_name, pattern):
              continue
  
          # Filter by policy permissions
          if not policy_engine.can_access_tool(agent_id, server, tool_name):
              continue
  
          # Check token budget limit
          if max_schema_tokens is not None:
              tool_tokens = _estimate_tool_tokens(tool)
              if token_count + tool_tokens > max_schema_tokens:
                  # Stop adding tools - budget exceeded
                  break
              token_count += tool_tokens
  
          # Convert tool to ToolDefinition
          tool_definition = ToolDefinition(
              name=tool_name,
              description=tool.description if hasattr(tool, 'description') and tool.description else "",
              inputSchema=tool.inputSchema if hasattr(tool, 'inputSchema') else {}
          )
  
          filtered_tools.append(tool_definition)
  
      return GetServerToolsResponse(
          tools=filtered_tools,
          server=server,
          total_available=total_available,
          returned=len(filtered_tools),
          tokens_used=token_count if max_schema_tokens is not None else None
      ).model_dump()

• src/gateway.py:24-39 (schema)

  Pydantic models defining the structure of tool definitions (ToolDefinition) and the response format (GetServerToolsResponse) for the get_server_tools tool. Input parameters are defined via Annotated types in the handler function signature.

  class ToolDefinition(BaseModel):
      """Tool definition from downstream server."""
      name: Annotated[str, Field(description="Tool name (use in execute_tool)")]
      description: Annotated[str, Field(description="What this tool does")]
      inputSchema: Annotated[dict, Field(description="JSON Schema defining required/optional parameters for execute_tool args")]
  
  
  class GetServerToolsResponse(BaseModel):
      """Response from get_server_tools."""
      tools: Annotated[list[ToolDefinition], Field(description="Tool definitions you can access")]
      server: Annotated[str, Field(description="Queried server name")]
      total_available: Annotated[int, Field(description="Total tools on server (may exceed returned if filtered by permissions/criteria)")]
      returned: Annotated[int, Field(description="Count of tools returned (less than total_available is normal due to filtering)")]
      tokens_used: Annotated[Optional[int], Field(description="Tokens used in schemas (if max_schema_tokens was set)")] = None
      error: Annotated[Optional[str], Field(description="Error message if request failed")] = None

• src/gateway.py:291-291 (registration)

  FastMCP decorator that registers the get_server_tools function as a tool on the gateway server.

  @gateway.tool

• src/gateway.py:225-250 (helper)

  Helper function used by get_server_tools to match tool names against wildcard patterns (e.g., 'get_*') using fnmatch.

  def _matches_pattern(tool_name: str, pattern: str) -> bool:
      """Check if tool name matches wildcard pattern.
  
      Uses glob-style pattern matching:
      - * matches any sequence of characters
      - ? matches any single character
      - [seq] matches any character in seq
      - [!seq] matches any character not in seq
  
      Args:
          tool_name: Name of the tool to match
          pattern: Pattern with wildcards (e.g., "get_*", "*_user")
  
      Returns:
          True if tool_name matches pattern, False otherwise
  
      Example:
          >>> _matches_pattern("get_user", "get_*")
          True
          >>> _matches_pattern("delete_user", "get_*")
          False
          >>> _matches_pattern("list_items", "*_items")
          True
      """
      return fnmatch.fnmatch(tool_name, pattern)

• src/gateway.py:252-289 (helper)

  Helper function used by get_server_tools to estimate the token count of a tool definition for enforcing max_schema_tokens budget limits.

  def _estimate_tool_tokens(tool: Any) -> int:
      """Estimate token count for a tool definition.
  
      Estimates tokens based on name, description, and input schema JSON length.
      Uses rough approximation: characters / 4 = tokens (typical for English text).
  
      Args:
          tool: Tool object with name, description, and inputSchema attributes
  
      Returns:
          Estimated token count for the tool definition
  
      Example:
          >>> tool = Tool(name="get_user", description="Get user by ID", inputSchema={...})
          >>> _estimate_tool_tokens(tool)
          42
      """
      # Count name length
      name_len = len(tool.name) if hasattr(tool, 'name') and tool.name else 0
  
      # Count description length
      desc_len = len(tool.description) if hasattr(tool, 'description') and tool.description else 0
  
      # Count input schema length (convert to string for estimation)
      schema_len = 0
      if hasattr(tool, 'inputSchema') and tool.inputSchema:
          # Convert schema dict to string for rough character count
          import json
          try:
              schema_len = len(json.dumps(tool.inputSchema))
          except Exception:
              # If serialization fails, use a default estimate
              schema_len = 100
  
      # Total characters / 4 = rough token estimate
      total_chars = name_len + desc_len + schema_len
      return max(1, total_chars // 4)