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)