From 72b27a284a7b8349676dcd0fd7032d13ac399959 Mon Sep 17 00:00:00 2001
From: Anurag DS <86948851+Anurag-gg@users.noreply.github.com>
Date: Sat, 11 Oct 2025 18:47:14 +0530
Subject: [PATCH 1/4] fixed a comment in File class (#2066)
---
src/fastmcp/utilities/types.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/fastmcp/utilities/types.py b/src/fastmcp/utilities/types.py
index 0bdea648c..a41ce6ccc 100644
--- a/src/fastmcp/utilities/types.py
+++ b/src/fastmcp/utilities/types.py
@@ -293,7 +293,7 @@ class Audio:
class File:
- """Helper class for returning audio from tools."""
+ """Helper class for returning file data from tools."""
def __init__(
self,
From 7b7e277d5ddcf1dd0c1dbf3eaa3461c1ca077efe Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Sat, 11 Oct 2025 10:53:54 -0400
Subject: [PATCH 2/4] Prevent confused deputy attacks in OAuth proxy (#2056)
---
docs/servers/auth/oauth-proxy.mdx | 146 ++--
docs/servers/auth/oidc-proxy.mdx | 12 +-
src/fastmcp/client/oauth_callback.py | 207 ++----
src/fastmcp/server/auth/oauth_proxy.py | 697 +++++++++++++++---
src/fastmcp/utilities/ui.py | 470 ++++++++++++
.../auth/test_github_provider_integration.py | 75 +-
tests/server/auth/providers/test_azure.py | 47 +-
tests/server/auth/test_oauth_consent_flow.py | 657 +++++++++++++++++
tests/server/auth/test_oauth_proxy.py | 203 +++--
9 files changed, 2077 insertions(+), 437 deletions(-)
create mode 100644 src/fastmcp/utilities/ui.py
create mode 100644 tests/server/auth/test_oauth_consent_flow.py
diff --git a/docs/servers/auth/oauth-proxy.mdx b/docs/servers/auth/oauth-proxy.mdx
index a5109ea0a..c06276e26 100644
--- a/docs/servers/auth/oauth-proxy.mdx
+++ b/docs/servers/auth/oauth-proxy.mdx
@@ -10,9 +10,9 @@ import { VersionBadge } from "/snippets/version-badge.mdx";
-OAuth Proxy enables FastMCP servers to authenticate with OAuth providers that **don't support Dynamic Client Registration (DCR)**. This includes virtually all traditional OAuth providers: GitHub, Google, Azure, AWS, Discord, Facebook, and most enterprise identity systems. For providers that do support DCR (like Descope and WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
+The OAuth proxy enables FastMCP servers to authenticate with OAuth providers that **don't support Dynamic Client Registration (DCR)**. This includes virtually all traditional OAuth providers: GitHub, Google, Azure, AWS, Discord, Facebook, and most enterprise identity systems. For providers that do support DCR (like Descope and WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
-MCP clients expect to register automatically and obtain credentials on the fly, but traditional providers require manual app registration through their developer consoles. OAuth Proxy bridges this gap by presenting a DCR-compliant interface to MCP clients while using your pre-registered credentials with the upstream provider. When a client attempts to register, the proxy returns your fixed credentials. When a client initiates authorization, the proxy handles the complexity of callback forwarding—storing the client's dynamic callback URL, using its own fixed callback with the provider, then forwarding back to the client after token exchange.
+MCP clients expect to register automatically and obtain credentials on the fly, but traditional providers require manual app registration through their developer consoles. The OAuth proxy bridges this gap by presenting a DCR-compliant interface to MCP clients while using your pre-registered credentials with the upstream provider. When a client attempts to register, the proxy returns your fixed credentials. When a client initiates authorization, the proxy handles the complexity of callback forwarding—storing the client's dynamic callback URL, using its own fixed callback with the provider, then forwarding back to the client after token exchange.
This approach enables any MCP client (whether using random localhost ports or fixed URLs like Claude.ai) to authenticate with any traditional OAuth provider, all while maintaining full OAuth 2.1 and PKCE security.
@@ -20,7 +20,7 @@ This approach enables any MCP client (whether using random localhost ports or fi
For providers that support OIDC discovery (Auth0, Google with OIDC
configuration, Azure AD), consider using [`OIDC
Proxy`](/servers/auth/oidc-proxy) for automatic configuration. OIDC Proxy
- extends OAuth Proxy to automatically discover endpoints from the provider's
+ extends the OAuth proxy to automatically discover endpoints from the provider's
`/.well-known/openid-configuration` URL, simplifying setup.
@@ -28,7 +28,7 @@ This approach enables any MCP client (whether using random localhost ports or fi
### Provider Setup Requirements
-Before using OAuth Proxy, you need to register your application with your OAuth provider:
+Before using the OAuth proxy, you need to register your application with your OAuth provider:
1. **Register your application** in the provider's developer console (GitHub Settings, Google Cloud Console, Azure Portal, etc.)
2. **Configure the redirect URI** as your FastMCP server URL plus your chosen callback path:
@@ -41,12 +41,12 @@ Before using OAuth Proxy, you need to register your application with your OAuth
The redirect URI you configure with your provider must exactly match your
FastMCP server's URL plus the callback path. If you customize `redirect_path`
- in OAuth Proxy, update your provider's redirect URI accordingly.
+ in the OAuth proxy, update your provider's redirect URI accordingly.
### Basic Setup
-Here's how to implement OAuth Proxy with any provider:
+Here's how to implement the OAuth proxy with any provider:
```python
from fastmcp import FastMCP
@@ -203,48 +203,6 @@ auth = OAuthProxy(..., client_storage=InMemoryStorage())
-### Provider-Specific Parameters
-
-Some OAuth providers require additional parameters beyond the standard OAuth2 flow. Use `extra_authorize_params` and `extra_token_params` to handle these requirements:
-
-#### Auth0 Example
-
-Auth0 requires an `audience` parameter to issue JWT tokens instead of opaque tokens:
-
-```python
-auth = OAuthProxy(
- upstream_authorization_endpoint="https://your-domain.auth0.com/authorize",
- upstream_token_endpoint="https://your-domain.auth0.com/oauth/token",
- upstream_client_id="your-auth0-client-id",
- upstream_client_secret="your-auth0-client-secret",
-
- # Auth0 requires audience for JWT tokens
- extra_authorize_params={
- "audience": "https://your-api-identifier.com"
- },
- extra_token_params={
- "audience": "https://your-api-identifier.com"
- },
-
- token_verifier=JWTVerifier(
- jwks_uri="https://your-domain.auth0.com/.well-known/jwks.json",
- issuer="https://your-domain.auth0.com/",
- audience="https://your-api-identifier.com"
- ),
-
- base_url="https://your-server.com"
-)
-```
-
-#### RFC 8707 Resource Indicators
-
-MCP clients can specify target resources using the standard `resource` parameter (RFC 8707). This is automatically forwarded when present:
-
-```python
-# Client code (automatic - no server configuration needed)
-# The resource parameter is passed through from AuthorizationParams
-```
-
### Using Built-in Providers
FastMCP includes pre-configured providers for common services:
@@ -263,21 +221,57 @@ mcp = FastMCP(name="My Server", auth=auth)
Available providers include `GitHubProvider`, `GoogleProvider`, and others. These handle token verification automatically.
+### Token Verification
+
+The OAuth proxy requires a compatible `TokenVerifier` to validate tokens from your provider. Different providers use different token formats:
+
+- **JWT tokens** (Google, Azure): Use `JWTVerifier` with the provider's JWKS endpoint
+- **Opaque tokens** (GitHub, Discord): Use provider-specific verifiers or implement custom validation
+
+See the [Token Verification guide](/servers/auth/token-verification) for detailed setup instructions for your provider.
+
### Scope Configuration
-OAuth scopes are configured through your `TokenVerifier`. Set `required_scopes` to automatically request the permissions your application needs:
+OAuth scopes control what permissions your application requests from users. They're configured through your `TokenVerifier` (required for the OAuth proxy to validate tokens from your provider). Set `required_scopes` to automatically request the permissions your application needs:
```python
JWTVerifier(..., required_scopes = ["read:user", "write:data"])
```
-Dynamic clients created by the proxy will automatically include these scopes in their authorization requests.
+Dynamic clients created by the proxy will automatically include these scopes in their authorization requests. See the [Token Verification](#token-verification) section below for detailed setup.
-## How It Works
+### Custom Parameters
+
+Some OAuth providers require additional parameters beyond the standard OAuth2 flow. Use `extra_authorize_params` and `extra_token_params` to pass provider-specific requirements. For example, Auth0 requires an `audience` parameter to issue JWT tokens instead of opaque tokens:
+
+```python
+auth = OAuthProxy(
+ upstream_authorization_endpoint="https://your-domain.auth0.com/authorize",
+ upstream_token_endpoint="https://your-domain.auth0.com/oauth/token",
+ upstream_client_id="your-auth0-client-id",
+ upstream_client_secret="your-auth0-client-secret",
+
+ # Auth0-specific audience parameter
+ extra_authorize_params={"audience": "https://your-api-identifier.com"},
+ extra_token_params={"audience": "https://your-api-identifier.com"},
+
+ token_verifier=JWTVerifier(
+ jwks_uri="https://your-domain.auth0.com/.well-known/jwks.json",
+ issuer="https://your-domain.auth0.com/",
+ audience="https://your-api-identifier.com"
+ ),
+ base_url="https://your-server.com"
+)
+```
+
+The proxy also automatically forwards RFC 8707 `resource` parameters from MCP clients to upstream providers that support them.
+
+## OAuth Flow
```mermaid
sequenceDiagram
participant Client as MCP Client (localhost:random)
+ participant User as User
participant Proxy as FastMCP OAuth Proxy (server:8000)
participant Provider as OAuth Provider (GitHub, etc.)
@@ -285,25 +279,27 @@ sequenceDiagram
Client->>Proxy: 1. POST /register redirect_uri: localhost:54321/callback
Proxy-->>Client: 2. Returns fixed upstream credentials
- Note over Client, Proxy: Authorization with PKCE & Callback Forwarding
+ Note over Client, User: Authorization with User Consent
Client->>Proxy: 3. GET /authorize redirect_uri=localhost:54321/callback code_challenge=CLIENT_CHALLENGE
Note over Proxy: Store transaction with client PKCE Generate proxy PKCE pair
- Proxy->>Provider: 4. Redirect to provider redirect_uri=server:8000/auth/callback code_challenge=PROXY_CHALLENGE
+ Proxy->>User: 4. Show consent page (client details, redirect URI, scopes)
+ User->>Proxy: 5. Approve/deny consent
+ Proxy->>Provider: 6. Redirect to provider redirect_uri=server:8000/auth/callback code_challenge=PROXY_CHALLENGE
Note over Provider, Proxy: Provider Callback
- Provider->>Proxy: 5. GET /auth/callback with authorization code
- Proxy->>Provider: 6. Exchange code for tokens code_verifier=PROXY_VERIFIER
- Provider-->>Proxy: 7. Access & refresh tokens
+ Provider->>Proxy: 7. GET /auth/callback with authorization code
+ Proxy->>Provider: 8. Exchange code for tokens code_verifier=PROXY_VERIFIER
+ Provider-->>Proxy: 9. Access & refresh tokens
Note over Proxy, Client: Client Callback Forwarding
- Proxy->>Client: 8. Redirect to localhost:54321/callback with new authorization code
+ Proxy->>Client: 10. Redirect to localhost:54321/callback with new authorization code
Note over Client, Proxy: Token Exchange
- Client->>Proxy: 9. POST /token with code code_verifier=CLIENT_VERIFIER
- Proxy-->>Client: 10. Returns stored provider tokens
+ Client->>Proxy: 11. POST /token with code code_verifier=CLIENT_VERIFIER
+ Proxy-->>Client: 12. Returns stored provider tokens
```
-The flow diagram above illustrates the complete OAuth Proxy pattern. Let's understand each phase:
+The flow diagram above illustrates the complete OAuth proxy pattern. Let's understand each phase:
### Registration Phase
@@ -315,9 +311,10 @@ The client initiates OAuth by redirecting to the proxy's `/authorize` endpoint.
1. Stores the client's transaction with its PKCE challenge
2. Generates its own PKCE parameters for upstream security
-3. Redirects to the upstream provider using the fixed callback URL
+3. Shows the user a consent page with the client's details, redirect URI, and requested scopes
+4. If the user approves (or the client was previously approved), redirects to the upstream provider using the fixed callback URL
-This dual-PKCE approach maintains end-to-end security at both the client-to-proxy and proxy-to-provider layers.
+This dual-PKCE approach maintains end-to-end security at both the client-to-proxy and proxy-to-provider layers. The consent step protects against confused deputy attacks by ensuring you explicitly approve each client before it can complete authorization.
### Callback Phase
@@ -336,7 +333,7 @@ This entire flow is transparent to the MCP client—it experiences a standard OA
### PKCE Forwarding
-OAuth Proxy automatically handles PKCE (Proof Key for Code Exchange) when working with providers that support or require it. The proxy generates its own PKCE parameters to send upstream while separately validating the client's PKCE, ensuring end-to-end security at both layers.
+The OAuth proxy automatically handles PKCE (Proof Key for Code Exchange) when working with providers that support or require it. The proxy generates its own PKCE parameters to send upstream while separately validating the client's PKCE, ensuring end-to-end security at both layers.
This is enabled by default via the `forward_pkce` parameter and works seamlessly with providers like Google, Azure AD, and GitHub. Only disable it for legacy providers that don't support PKCE:
@@ -350,7 +347,7 @@ auth = OAuthProxy(
### Redirect URI Validation
-While OAuth Proxy accepts all redirect URIs by default (for DCR compatibility), you can restrict which clients can connect by specifying allowed patterns:
+While the OAuth proxy accepts all redirect URIs by default (for DCR compatibility), you can restrict which clients can connect by specifying allowed patterns:
```python
# Allow only localhost clients (common for development)
@@ -375,20 +372,29 @@ auth = OAuthProxy(
Check your server logs for "Client registered with redirect_uri" messages to identify what URLs your clients use.
-## Token Verification
+## Security
-OAuth Proxy requires a compatible `TokenVerifier` to validate tokens from your provider. Different providers use different token formats:
+### Confused Deputy Attacks
-- **JWT tokens** (Google, Azure): Use `JWTVerifier` with the provider's JWKS endpoint
-- **Opaque tokens** (GitHub, Discord): Use provider-specific verifiers or implement custom validation
+
-See the [Token Verification guide](/servers/auth/token-verification) for detailed setup instructions for your provider.
+A confused deputy attack allows a malicious client to steal your authorization by tricking you into granting it access under your identity.
+
+The OAuth proxy works by bridging DCR clients to traditional auth providers, which means that multiple MCP clients connect through a single upstream OAuth application. An attacker can exploit this shared application by registering a malicious client with their own redirect URI, then sending you an authorization link. When you click it, your browser goes through the OAuth flow—but since you may have already authorized this OAuth app before, the provider might auto-approve the request. The authorization code then gets sent to the attacker's redirect URI instead of a legitimate client, giving them access under your credentials.
+
+#### Mitigation
+
+FastMCP's OAuth proxy requires you to explicitly consent whenever any new or unrecognized client attempts to connect to your server. Before any authorization happens, you see a consent page showing the client's details, redirect URI, and requested scopes. This gives you the opportunity to review and deny suspicious requests. Once you approve a client, it's remembered so you don't see the consent page again for that client. The consent mechanism is implemented with CSRF tokens and cryptographically signed cookies to prevent tampering.
+
+**Learn more:**
+- [MCP Security Best Practices](https://modelcontextprotocol.io/specification/2025-06-18/basic/security_best_practices#confused-deputy-problem) - Official specification guidance
+- [Confused Deputy Attacks Explained](https://den.dev/blog/mcp-confused-deputy-api-management/) - Detailed walkthrough by Den Delimarsky
## Environment Configuration
-For production deployments, configure OAuth Proxy through environment variables instead of hardcoding credentials:
+For production deployments, configure the OAuth proxy through environment variables instead of hardcoding credentials:
```bash
# Specify the provider implementation
diff --git a/docs/servers/auth/oidc-proxy.mdx b/docs/servers/auth/oidc-proxy.mdx
index e08d0c358..22f5e2ee8 100644
--- a/docs/servers/auth/oidc-proxy.mdx
+++ b/docs/servers/auth/oidc-proxy.mdx
@@ -10,15 +10,15 @@ import { VersionBadge } from "/snippets/version-badge.mdx";
-OIDC Proxy enables FastMCP servers to authenticate with OIDC providers that **don't support Dynamic Client Registration (DCR)** out of the box. This includes OAuth providers like: Auth0, Google, Azure, AWS, etc. For providers that do support DCR (like WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
+The OIDC proxy enables FastMCP servers to authenticate with OIDC providers that **don't support Dynamic Client Registration (DCR)** out of the box. This includes OAuth providers like: Auth0, Google, Azure, AWS, etc. For providers that do support DCR (like WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
-The OIDC Proxy is built upon [`OAuthProxy`](/servers/auth/oauth-proxy) so it has all the same functionality under the covers.
+The OIDC proxy is built upon [`OAuthProxy`](/servers/auth/oauth-proxy) so it has all the same functionality under the covers.
## Implementation
### Provider Setup Requirements
-Before using OIDC Proxy, you need to register your application with your OAuth provider:
+Before using the OIDC proxy, you need to register your application with your OAuth provider:
1. **Register your application** in the provider's developer console (Auth0 Applications, Google Cloud Console, Azure Portal, etc.)
2. **Configure the redirect URI** as your FastMCP server URL plus your chosen callback path:
@@ -30,12 +30,12 @@ Before using OIDC Proxy, you need to register your application with your OAuth p
The redirect URI you configure with your provider must exactly match your
FastMCP server's URL plus the callback path. If you customize `redirect_path`
- in OAuth Proxy, update your provider's redirect URI accordingly.
+ in the OIDC proxy, update your provider's redirect URI accordingly.
### Basic Setup
-Here's how to implement OIDC Proxy with any provider:
+Here's how to implement the OIDC proxy with any provider:
```python
from fastmcp import FastMCP
@@ -172,7 +172,7 @@ Dynamic clients created by the proxy will automatically include these scopes in
-For production deployments, configure OIDC Proxy through environment variables instead of hardcoding credentials:
+For production deployments, configure the OIDC proxy through environment variables instead of hardcoding credentials:
```bash
# Specify the provider implementation
diff --git a/src/fastmcp/client/oauth_callback.py b/src/fastmcp/client/oauth_callback.py
index c6da794e8..cf1166a80 100644
--- a/src/fastmcp/client/oauth_callback.py
+++ b/src/fastmcp/client/oauth_callback.py
@@ -12,12 +12,21 @@ from dataclasses import dataclass
from starlette.applications import Starlette
from starlette.requests import Request
-from starlette.responses import HTMLResponse
from starlette.routing import Route
from uvicorn import Config, Server
from fastmcp.utilities.http import find_available_port
from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.ui import (
+ HELPER_TEXT_STYLES,
+ INFO_BOX_STYLES,
+ STATUS_MESSAGE_STYLES,
+ create_info_box,
+ create_logo,
+ create_page,
+ create_secure_html_response,
+ create_status_message,
+)
logger = get_logger(__name__)
@@ -29,155 +38,41 @@ def create_callback_html(
server_url: str | None = None,
) -> str:
"""Create a styled HTML response for OAuth callbacks."""
- logo_url = "https://gofastmcp.com/assets/brand/blue-logo.png"
-
# Build the main status message
- if is_success:
- status_title = "Authentication successful"
- status_icon = "✓"
- icon_bg = "#10b98120"
- else:
- status_title = "Authentication failed"
- status_icon = "✕"
- icon_bg = "#ef444420"
+ status_title = (
+ "Authentication successful" if is_success else "Authentication failed"
+ )
# Add detail info box for both success and error cases
detail_info = ""
if is_success and server_url:
- detail_info = f"""
-
-
-
"""
+ # Additional styles needed for this page
+ additional_styles = STATUS_MESSAGE_STYLES + INFO_BOX_STYLES + HELPER_TEXT_STYLES
+
+ return create_page(
+ content=content,
+ title=title,
+ additional_styles=additional_styles,
+ )
+
@dataclass
class CallbackResponse:
@@ -221,32 +116,34 @@ def create_oauth_callback_server(
if callback_response.error:
error_desc = callback_response.error_description or "Unknown error"
+ # Create user-friendly error messages
+ if callback_response.error == "access_denied":
+ user_message = "Access was denied by the authorization server."
+ else:
+ user_message = f"Authorization failed: {error_desc}"
+
# Resolve future with exception if provided
if response_future and not response_future.done():
- response_future.set_exception(
- RuntimeError(
- f"OAuth error: {callback_response.error} - {error_desc}"
- )
- )
+ response_future.set_exception(RuntimeError(user_message))
- return HTMLResponse(
+ return create_secure_html_response(
create_callback_html(
- f"FastMCP OAuth Error: {callback_response.error} {error_desc}",
+ user_message,
is_success=False,
),
status_code=400,
)
if not callback_response.code:
+ user_message = "No authorization code was received from the server."
+
# Resolve future with exception if provided
if response_future and not response_future.done():
- response_future.set_exception(
- RuntimeError("OAuth callback missing authorization code")
- )
+ response_future.set_exception(RuntimeError(user_message))
- return HTMLResponse(
+ return create_secure_html_response(
create_callback_html(
- "FastMCP OAuth Error: No authorization code received",
+ user_message,
is_success=False,
),
status_code=400,
@@ -254,17 +151,17 @@ def create_oauth_callback_server(
# Check for missing state parameter (indicates OAuth flow issue)
if callback_response.state is None:
+ user_message = (
+ "The OAuth server did not return the expected state parameter."
+ )
+
# Resolve future with exception if provided
if response_future and not response_future.done():
- response_future.set_exception(
- RuntimeError(
- "OAuth server did not return state parameter - authentication failed"
- )
- )
+ response_future.set_exception(RuntimeError(user_message))
- return HTMLResponse(
+ return create_secure_html_response(
create_callback_html(
- "FastMCP OAuth Error: Authentication failed The OAuth server did not return the expected state parameter",
+ user_message,
is_success=False,
),
status_code=400,
@@ -276,7 +173,7 @@ def create_oauth_callback_server(
(callback_response.code, callback_response.state)
)
- return HTMLResponse(
+ return create_secure_html_response(
create_callback_html("", is_success=True, server_url=server_url)
)
diff --git a/src/fastmcp/server/auth/oauth_proxy.py b/src/fastmcp/server/auth/oauth_proxy.py
index 998aa05a6..2781f3516 100644
--- a/src/fastmcp/server/auth/oauth_proxy.py
+++ b/src/fastmcp/server/auth/oauth_proxy.py
@@ -18,12 +18,15 @@ production use with enterprise identity providers.
from __future__ import annotations
+import base64
import hashlib
+import hmac
+import json
import secrets
import time
from base64 import urlsafe_b64encode
from typing import TYPE_CHECKING, Any, Final
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlparse
import httpx
from authlib.common.security import generate_token
@@ -48,14 +51,26 @@ from mcp.server.auth.settings import (
RevocationOptions,
)
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
-from pydantic import AnyHttpUrl, AnyUrl, Field, SecretStr
+from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, SecretStr
from starlette.requests import Request
-from starlette.responses import RedirectResponse
+from starlette.responses import HTMLResponse, RedirectResponse
from starlette.routing import Route
from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
-from fastmcp.server.auth.redirect_validation import validate_redirect_uri
+from fastmcp.server.auth.redirect_validation import (
+ validate_redirect_uri,
+)
from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.ui import (
+ BUTTON_STYLES,
+ DETAIL_BOX_STYLES,
+ INFO_BOX_STYLES,
+ TOOLTIP_STYLES,
+ create_detail_box,
+ create_logo,
+ create_page,
+ create_secure_html_response,
+)
if TYPE_CHECKING:
pass
@@ -63,6 +78,62 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
+# -------------------------------------------------------------------------
+# Constants
+# -------------------------------------------------------------------------
+
+# Default token expiration times
+DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS: Final[int] = 60 * 60 # 1 hour
+DEFAULT_AUTH_CODE_EXPIRY_SECONDS: Final[int] = 5 * 60 # 5 minutes
+
+# HTTP client timeout
+HTTP_TIMEOUT_SECONDS: Final[int] = 30
+
+
+# -------------------------------------------------------------------------
+# Pydantic Models
+# -------------------------------------------------------------------------
+
+
+class OAuthTransaction(BaseModel):
+ """OAuth transaction state for consent flow.
+
+ Stored server-side to track active authorization flows with client context.
+ Includes CSRF tokens for consent protection per MCP security best practices.
+ """
+
+ txn_id: str
+ client_id: str
+ client_redirect_uri: str
+ client_state: str
+ code_challenge: str | None
+ code_challenge_method: str
+ scopes: list[str]
+ created_at: float
+ resource: str | None = None
+ proxy_code_verifier: str | None = None
+ csrf_token: str | None = None
+ csrf_expires_at: float | None = None
+
+
+class ClientCode(BaseModel):
+ """Client authorization code with PKCE and upstream tokens.
+
+ Stored server-side after upstream IdP callback. Contains the upstream
+ tokens bound to the client's PKCE challenge for secure token exchange.
+ """
+
+ code: str
+ client_id: str
+ redirect_uri: str
+ code_challenge: str | None
+ code_challenge_method: str
+ scopes: list[str]
+ idp_tokens: dict[str, Any]
+ expires_at: float
+ created_at: float
+
+
class ProxyDCRClient(OAuthClientInformationFull):
"""Client for DCR proxy with configurable redirect URI validation.
@@ -90,6 +161,7 @@ class ProxyDCRClient(OAuthClientInformationFull):
"""
allowed_redirect_uri_patterns: list[str] | None = Field(default=None)
+ client_name: str | None = Field(default=None)
def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
"""Validate redirect URI against allowed patterns.
@@ -112,12 +184,110 @@ class ProxyDCRClient(OAuthClientInformationFull):
return super().validate_redirect_uri(redirect_uri)
-# Default token expiration times
-DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS: Final[int] = 60 * 60 # 1 hour
-DEFAULT_AUTH_CODE_EXPIRY_SECONDS: Final[int] = 5 * 60 # 5 minutes
+# -------------------------------------------------------------------------
+# Helper Functions
+# -------------------------------------------------------------------------
-# HTTP client timeout
-HTTP_TIMEOUT_SECONDS: Final[int] = 30
+
+def create_consent_html(
+ client_id: str,
+ redirect_uri: str,
+ scopes: list[str],
+ txn_id: str,
+ csrf_token: str,
+ client_name: str | None = None,
+ title: str = "Authorization Consent",
+) -> str:
+ """Create a styled HTML consent page for OAuth authorization requests."""
+ # Format scopes for display
+ scopes_display = ", ".join(scopes) if scopes else "None"
+
+ # Build warning box with client name if available
+ client_display = client_name or client_id
+ warning_box = f"""
+
+
{client_display} is requesting access to this FastMCP server.
+
Review the details below before approving.
+
+ """
+
+ # Build detail box with client information
+ detail_rows = []
+ if client_name:
+ detail_rows.append(("Client Name", client_name))
+ detail_rows.extend(
+ [
+ ("Client ID", client_id),
+ ("Redirect URI", redirect_uri),
+ ("Requested Scopes", scopes_display),
+ ]
+ )
+ detail_box = create_detail_box(detail_rows)
+
+ # Build form with buttons
+ form = f"""
+
+ """
+
+ # Build help link with tooltip
+ help_link = """
+
+
+ Why am I seeing this?
+
+ This FastMCP server requires your consent to allow a new client
+ to connect. This protects you from confused deputy
+ attacks, where malicious clients could impersonate you
+ and steal access.
+ """
+
+
+def create_info_box(
+ content: str, is_error: bool = False, centered: bool = False
+) -> str:
+ """
+ Create an info box.
+
+ Args:
+ content: HTML content for the info box
+ is_error: True for error styling, False for normal
+ centered: True to center the text, False for left-aligned
+
+ Returns:
+ HTML for info box
+ """
+ classes = ["info-box"]
+ if is_error:
+ classes.append("error")
+ if centered:
+ classes.append("centered")
+ class_str = " ".join(classes)
+ return f'
{content}
'
+
+
+def create_detail_box(rows: list[tuple[str, str]]) -> str:
+ """
+ Create a detail box with key-value pairs.
+
+ Args:
+ rows: List of (label, value) tuples
+
+ Returns:
+ HTML for detail box
+ """
+ rows_html = "\n".join(
+ f"""
+
+
{label}:
+
{value}
+
+ """
+ for label, value in rows
+ )
+
+ return f'
{rows_html}
'
+
+
+def create_button_group(buttons: list[tuple[str, str, str]]) -> str:
+ """
+ Create a group of buttons.
+
+ Args:
+ buttons: List of (text, value, css_class) tuples
+
+ Returns:
+ HTML for button group
+ """
+ buttons_html = "\n".join(
+ f''
+ for text, value, css_class in buttons
+ )
+
+ return f'
{buttons_html}
'
+
+
+def create_secure_html_response(html: str, status_code: int = 200) -> HTMLResponse:
+ """
+ Create an HTMLResponse with security headers.
+
+ Adds X-Frame-Options: DENY to prevent clickjacking attacks per MCP security best practices.
+
+ Args:
+ html: HTML content to return
+ status_code: HTTP status code
+
+ Returns:
+ HTMLResponse with security headers
+ """
+ return HTMLResponse(
+ content=html,
+ status_code=status_code,
+ headers={"X-Frame-Options": "DENY"},
+ )
diff --git a/tests/integration_tests/auth/test_github_provider_integration.py b/tests/integration_tests/auth/test_github_provider_integration.py
index 1fbb14b8c..54d159fbd 100644
--- a/tests/integration_tests/auth/test_github_provider_integration.py
+++ b/tests/integration_tests/auth/test_github_provider_integration.py
@@ -11,6 +11,7 @@ with the following configuration:
"""
import os
+import re
from collections.abc import Generator
from urllib.parse import parse_qs, urlparse
@@ -84,6 +85,8 @@ def create_github_server_with_mock_callback(
import secrets
import time
+ from fastmcp.server.auth.oauth_proxy import ClientCode
+
# Generate a fake authorization code
fake_code = secrets.token_urlsafe(32)
@@ -94,17 +97,21 @@ def create_github_server_with_mock_callback(
"expires_in": 3600,
}
- # Store the mock tokens in the proxy's client codes
- auth._client_codes[fake_code] = {
- "client_id": client.client_id,
- "redirect_uri": str(params.redirect_uri),
- "code_challenge": params.code_challenge,
- "code_challenge_method": getattr(params, "code_challenge_method", "S256"),
- "scopes": params.scopes or [],
- "idp_tokens": mock_tokens,
- "expires_at": int(time.time() + 300), # 5 minutes
- "created_at": time.time(),
- }
+ # Store the mock tokens in the proxy's code storage
+ await auth._code_store.put(
+ key=fake_code,
+ value=ClientCode(
+ code=fake_code,
+ client_id=client.client_id,
+ redirect_uri=str(params.redirect_uri),
+ code_challenge=params.code_challenge,
+ code_challenge_method=getattr(params, "code_challenge_method", "S256"),
+ scopes=params.scopes or [],
+ idp_tokens=mock_tokens,
+ expires_at=int(time.time() + 300), # 5 minutes
+ created_at=time.time(),
+ ),
+ )
# Return the redirect to the client's callback with the fake code
callback_params = {
@@ -204,11 +211,12 @@ async def test_github_oauth_credentials_available():
async def test_github_oauth_authorization_redirect(github_server: str):
- """Test that GitHub OAuth authorization redirects to GitHub correctly.
+ """Test that GitHub OAuth authorization redirects to GitHub correctly through consent flow.
Since HeadlessOAuth can't handle real GitHub redirects, we test that:
1. DCR client registration works
- 2. Authorization endpoint redirects to GitHub with correct parameters
+ 2. Authorization endpoint redirects to consent page
+ 3. Consent approval redirects to GitHub with correct parameters
"""
# Extract base URL
parsed = urlparse(github_server)
@@ -235,7 +243,7 @@ async def test_github_oauth_authorization_redirect(github_server: str):
client_id = client_info["client_id"]
assert client_id is not None
- # Step 2: Test authorization endpoint redirects to GitHub
+ # Step 2: Test authorization endpoint redirects to consent page
auth_url = f"{base_url}/authorize"
auth_params = {
"response_type": "code",
@@ -250,9 +258,44 @@ async def test_github_oauth_authorization_redirect(github_server: str):
auth_url, params=auth_params, follow_redirects=False
)
- # Should redirect to GitHub
+ # Should redirect to consent page (confused deputy protection)
assert auth_response.status_code == 302
- redirect_location = auth_response.headers["location"]
+ consent_location = auth_response.headers["location"]
+ assert "/consent" in consent_location
+
+ # Step 3: Visit consent page to get CSRF token
+ consent_response = await http_client.get(
+ consent_location, follow_redirects=False
+ )
+ assert consent_response.status_code == 200
+
+ # Extract CSRF token from consent page HTML
+ csrf_match = re.search(
+ r'name="csrf_token"\s+value="([^"]+)"', consent_response.text
+ )
+ assert csrf_match, "CSRF token not found in consent page"
+ csrf_token = csrf_match.group(1)
+
+ # Extract txn_id from consent URL
+ txn_id_match = re.search(r"txn_id=([^&]+)", consent_location)
+ assert txn_id_match, "txn_id not found in consent URL"
+ txn_id = txn_id_match.group(1)
+
+ # Step 4: Approve consent
+ approve_response = await http_client.post(
+ f"{base_url}/consent/submit",
+ data={
+ "action": "approve",
+ "txn_id": txn_id,
+ "csrf_token": csrf_token,
+ },
+ cookies=consent_response.cookies,
+ follow_redirects=False,
+ )
+
+ # Should redirect to GitHub
+ assert approve_response.status_code in (302, 303)
+ redirect_location = approve_response.headers["location"]
# Parse redirect URL - should be GitHub
redirect_parsed = urlparse(redirect_location)
diff --git a/tests/server/auth/providers/test_azure.py b/tests/server/auth/providers/test_azure.py
index ec360403a..970a5c3b3 100644
--- a/tests/server/auth/providers/test_azure.py
+++ b/tests/server/auth/providers/test_azure.py
@@ -213,6 +213,14 @@ class TestAzureProvider:
base_url="https://srv.example",
)
+ await provider.register_client(
+ OAuthClientInformationFull(
+ client_id="dummy",
+ client_secret="secret",
+ redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+ )
+ )
+
client = OAuthClientInformationFull(
client_id="dummy",
client_secret="secret",
@@ -230,13 +238,19 @@ class TestAzureProvider:
url = await provider.authorize(client, params)
+ # Extract transaction ID from consent redirect
parsed = urlparse(url)
qs = parse_qs(parsed.query)
- assert "resource" not in qs
- scope_value = qs.get("scope", [""])[0]
- scope_parts = scope_value.split(" ") if scope_value else []
- assert "api://my-api/read" in scope_parts
- assert "api://my-api/profile" in scope_parts
+ assert "txn_id" in qs, "Should redirect to consent page with transaction ID"
+ txn_id = qs["txn_id"][0]
+
+ # Verify transaction contains correct parameters (resource filtered, scopes prefixed)
+ transaction = await provider._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert "api://my-api/read" in transaction.scopes
+ assert "api://my-api/profile" in transaction.scopes
+ # Azure provider filters resource parameter (not stored in transaction)
+ assert transaction.resource is None
@pytest.mark.asyncio
async def test_authorize_appends_unprefixed_additional_scopes(self):
@@ -251,6 +265,14 @@ class TestAzureProvider:
additional_authorize_scopes=["Mail.Read", "User.Read"],
)
+ await provider.register_client(
+ OAuthClientInformationFull(
+ client_id="dummy",
+ client_secret="secret",
+ redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+ )
+ )
+
client = OAuthClientInformationFull(
client_id="dummy",
client_secret="secret",
@@ -267,10 +289,15 @@ class TestAzureProvider:
url = await provider.authorize(client, params)
+ # Extract transaction ID from consent redirect
parsed = urlparse(url)
qs = parse_qs(parsed.query)
- scope_value = qs.get("scope", [""])[0]
- scope_parts = scope_value.split(" ") if scope_value else []
- assert "api://my-api/read" in scope_parts
- assert "Mail.Read" in scope_parts
- assert "User.Read" in scope_parts
+ assert "txn_id" in qs, "Should redirect to consent page with transaction ID"
+ txn_id = qs["txn_id"][0]
+
+ # Verify transaction contains correct scopes (prefixed + unprefixed additional)
+ transaction = await provider._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert "api://my-api/read" in transaction.scopes
+ assert "Mail.Read" in transaction.scopes
+ assert "User.Read" in transaction.scopes
diff --git a/tests/server/auth/test_oauth_consent_flow.py b/tests/server/auth/test_oauth_consent_flow.py
new file mode 100644
index 000000000..41eff87ca
--- /dev/null
+++ b/tests/server/auth/test_oauth_consent_flow.py
@@ -0,0 +1,657 @@
+"""Tests for OAuth Proxy consent flow with server-side storage.
+
+This test suite verifies:
+1. OAuth transactions are stored in server-side storage (not in-memory)
+2. Authorization codes are stored in server-side storage
+3. Consent flow redirects correctly through /consent endpoint
+4. CSRF protection works with cookies
+5. State persists across storage backends
+6. Security headers (X-Frame-Options) are set correctly
+7. Cookie signing and tampering detection
+8. Auto-approve behavior with valid cookies
+"""
+
+import re
+import secrets
+import time
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+from key_value.aio.stores.memory import MemoryStore
+from mcp.server.auth.provider import AuthorizationParams
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
+from starlette.applications import Starlette
+from starlette.testclient import TestClient
+
+from fastmcp.server.auth.auth import TokenVerifier
+from fastmcp.server.auth.oauth_proxy import OAuthProxy
+
+
+class MockTokenVerifier(TokenVerifier):
+ """Mock token verifier for testing."""
+
+ def __init__(self):
+ self.required_scopes = ["read", "write"]
+
+ async def verify_token(self, token: str):
+ """Mock token verification."""
+ from fastmcp.server.auth.auth import AccessToken
+
+ return AccessToken(
+ token=token,
+ client_id="mock-client",
+ scopes=self.required_scopes,
+ expires_at=int(time.time() + 3600),
+ )
+
+
+class _Verifier(TokenVerifier):
+ """Minimal token verifier for security tests."""
+
+ def __init__(self):
+ self.required_scopes = ["read"]
+
+ async def verify_token(self, token: str):
+ from fastmcp.server.auth.auth import AccessToken
+
+ return AccessToken(
+ token=token, client_id="c", scopes=self.required_scopes, expires_at=None
+ )
+
+
+@pytest.fixture
+def storage():
+ """Create a fresh in-memory storage for each test."""
+ return MemoryStore()
+
+
+@pytest.fixture
+def oauth_proxy_with_storage(storage):
+ """Create OAuth proxy with explicit storage backend."""
+ return OAuthProxy(
+ upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+ upstream_token_endpoint="https://github.com/login/oauth/access_token",
+ upstream_client_id="test-upstream-client",
+ upstream_client_secret="test-upstream-secret",
+ token_verifier=MockTokenVerifier(),
+ base_url="https://myserver.com",
+ redirect_path="/auth/callback",
+ client_storage=storage, # Use our test storage
+ )
+
+
+@pytest.fixture
+def oauth_proxy_https():
+ """OAuthProxy configured with HTTPS base_url for __Host- cookies."""
+ return OAuthProxy(
+ upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+ upstream_token_endpoint="https://github.com/login/oauth/access_token",
+ upstream_client_id="client-id",
+ upstream_client_secret="client-secret",
+ token_verifier=_Verifier(),
+ base_url="https://myserver.example",
+ client_storage=MemoryStore(),
+ )
+
+
+async def _start_flow(
+ proxy: OAuthProxy, client_id: str, redirect: str
+) -> tuple[str, str]:
+ """Register client and start auth; returns (txn_id, consent_url)."""
+ await proxy.register_client(
+ OAuthClientInformationFull(
+ client_id=client_id,
+ client_secret="s",
+ redirect_uris=[AnyUrl(redirect)],
+ )
+ )
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl(redirect),
+ redirect_uri_provided_explicitly=True,
+ state="client-state-xyz",
+ code_challenge="challenge",
+ code_challenge_method="S256",
+ scopes=["read"],
+ )
+ consent_url = await proxy.authorize(
+ OAuthClientInformationFull(
+ client_id=client_id,
+ client_secret="s",
+ redirect_uris=[AnyUrl(redirect)],
+ ),
+ params,
+ )
+ qs = parse_qs(urlparse(consent_url).query)
+ return qs["txn_id"][0], consent_url
+
+
+def _extract_csrf(html: str) -> str | None:
+ """Extract CSRF token from HTML form."""
+ m = re.search(r"name=\"csrf_token\"\s+value=\"([^\"]+)\"", html)
+ return m.group(1) if m else None
+
+
+class TestServerSideStorage:
+ """Tests verifying OAuth state is stored in AsyncKeyValue storage."""
+
+ async def test_transaction_stored_in_storage_not_memory(
+ self, oauth_proxy_with_storage, storage
+ ):
+ """Verify OAuth transactions are stored in AsyncKeyValue, not in-memory dict."""
+ # Register client
+ client = OAuthClientInformationFull(
+ client_id="test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ # Start authorization flow
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:54321/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="client-state-123",
+ code_challenge="challenge-abc",
+ code_challenge_method="S256",
+ scopes=["read", "write"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+
+ # Extract transaction ID from consent redirect
+ parsed = urlparse(redirect_url)
+ assert "/consent" in parsed.path, "Should redirect to consent page"
+
+ query_params = parse_qs(parsed.query)
+ txn_id = query_params["txn_id"][0]
+
+ # Verify transaction is NOT in the old in-memory dict
+ # (the attribute should not exist or should be empty)
+ assert (
+ not hasattr(oauth_proxy_with_storage, "_oauth_transactions")
+ or len(getattr(oauth_proxy_with_storage, "_oauth_transactions", {})) == 0
+ )
+
+ # Verify transaction IS in storage backend
+ transaction = await storage.get(collection="mcp-oauth-transactions", key=txn_id)
+ assert transaction is not None, "Transaction should be in storage"
+
+ # Verify transaction has expected structure
+ assert transaction["client_id"] == "test-client"
+ assert transaction["client_redirect_uri"] == "http://localhost:54321/callback"
+ assert transaction["client_state"] == "client-state-123"
+ assert transaction["code_challenge"] == "challenge-abc"
+ assert transaction["scopes"] == ["read", "write"]
+
+ async def test_authorization_code_stored_in_storage(
+ self, oauth_proxy_with_storage, storage
+ ):
+ """Verify authorization codes are stored in AsyncKeyValue storage."""
+ # Register client
+ client = OAuthClientInformationFull(
+ client_id="test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ # Create a test app with OAuth routes
+ app = Starlette(routes=oauth_proxy_with_storage.get_routes())
+
+ with TestClient(app) as test_client:
+ # Start authorization flow
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:54321/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="client-state",
+ code_challenge="challenge-xyz",
+ code_challenge_method="S256",
+ scopes=["read"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+
+ # Extract txn_id from consent redirect
+ parsed = urlparse(redirect_url)
+ query_params = parse_qs(parsed.query)
+ txn_id = query_params["txn_id"][0]
+
+ # Simulate consent approval
+ # First, get the consent page to establish CSRF cookie
+ consent_response = test_client.get(
+ f"/consent?txn_id={txn_id}", follow_redirects=False
+ )
+
+ # Extract CSRF token from response (it's in the HTML form)
+ csrf_token = None
+ if consent_response.status_code == 200:
+ # For this test, we'll generate a CSRF token manually
+ # In production, this comes from the consent page HTML
+ csrf_token = secrets.token_urlsafe(32)
+
+ # Approve consent with CSRF token
+ approval_response = test_client.post(
+ "/consent",
+ data={"action": "approve", "txn": txn_id, "csrf_token": csrf_token},
+ cookies=consent_response.cookies,
+ follow_redirects=False,
+ )
+
+ # After approval, authorization code should be in storage
+ # The code is returned in the redirect URL
+ if approval_response.status_code in (302, 303):
+ location = approval_response.headers.get("location", "")
+ callback_params = parse_qs(urlparse(location).query)
+
+ if "code" in callback_params:
+ auth_code = callback_params["code"][0]
+
+ # Verify code is NOT in old in-memory dict
+ assert (
+ not hasattr(oauth_proxy_with_storage, "_client_codes")
+ or len(getattr(oauth_proxy_with_storage, "_client_codes", {}))
+ == 0
+ )
+
+ # Verify code IS in storage
+ code_data = await storage.get(
+ collection="mcp-authorization-codes", key=auth_code
+ )
+ assert code_data is not None, (
+ "Authorization code should be in storage"
+ )
+ assert code_data["client_id"] == "test-client"
+ assert code_data["scopes"] == ["read"]
+
+ async def test_storage_collections_are_isolated(self, oauth_proxy_with_storage):
+ """Verify that transactions, codes, and clients use separate collections."""
+ # Register a client
+ client = OAuthClientInformationFull(
+ client_id="isolation-test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ # Start authorization to create transaction
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:12345/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="test-state",
+ code_challenge="test-challenge",
+ code_challenge_method="S256",
+ scopes=["read"],
+ )
+
+ await oauth_proxy_with_storage.authorize(client, params)
+
+ # Get all collections from storage
+ storage = oauth_proxy_with_storage._client_storage
+
+ # Verify client is in client collection
+ client_data = await storage.get(
+ collection="mcp-oauth-proxy-clients", key="isolation-test-client"
+ )
+ assert client_data is not None
+
+ # Verify we can list transactions separately
+ # (This tests that collections are properly namespaced)
+ transactions = await storage.keys(collection="mcp-oauth-transactions")
+
+ assert len(transactions) > 0, "Should have at least one transaction"
+
+ # Verify transaction keys don't collide with client keys
+ for txn_key in transactions:
+ assert txn_key != "isolation-test-client"
+
+
+class TestConsentFlowRedirects:
+ """Tests for consent flow redirect behavior."""
+
+ async def test_authorize_redirects_to_consent_page(self, oauth_proxy_with_storage):
+ """Verify authorize() redirects to /consent instead of upstream."""
+ client = OAuthClientInformationFull(
+ client_id="consent-test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:8080/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:8080/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="test-state",
+ code_challenge="",
+ scopes=["read"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+
+ # Should redirect to consent page, not upstream
+ assert "/consent" in redirect_url
+ assert "github.com" not in redirect_url
+ assert "?txn_id=" in redirect_url
+
+ async def test_consent_page_contains_transaction_id(self, oauth_proxy_with_storage):
+ """Verify consent page receives and displays transaction ID."""
+ client = OAuthClientInformationFull(
+ client_id="txn-test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:9090/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:9090/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="test-state",
+ code_challenge="test-challenge",
+ scopes=["read", "write"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+
+ # Extract txn_id parameter
+ parsed = urlparse(redirect_url)
+ query = parse_qs(parsed.query)
+
+ assert "txn_id" in query
+ txn_id = query["txn_id"][0]
+ assert len(txn_id) > 0
+
+ # Create test client
+ app = Starlette(routes=oauth_proxy_with_storage.get_routes())
+
+ with TestClient(app) as test_client:
+ # Request consent page
+ response = test_client.get(
+ f"/consent?txn_id={txn_id}", follow_redirects=False
+ )
+
+ assert response.status_code == 200
+ # Consent page should contain transaction reference
+ assert txn_id.encode() in response.content or b"consent" in response.content
+
+
+class TestCSRFProtection:
+ """Tests for CSRF protection in consent flow."""
+
+ async def test_consent_requires_csrf_token(self, oauth_proxy_with_storage):
+ """Verify consent submission requires valid CSRF token."""
+ client = OAuthClientInformationFull(
+ client_id="csrf-test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:7070/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:7070/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="test-state",
+ code_challenge="",
+ scopes=["read"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+ parsed = urlparse(redirect_url)
+ query = parse_qs(parsed.query)
+ txn_id = query["txn_id"][0]
+
+ app = Starlette(routes=oauth_proxy_with_storage.get_routes())
+
+ with TestClient(app) as test_client:
+ # Try to submit consent WITHOUT CSRF token
+ response = test_client.post(
+ "/consent/submit",
+ data={"action": "approve", "txn_id": txn_id},
+ # No CSRF token!
+ follow_redirects=False,
+ )
+
+ # Should reject or require CSRF
+ # (Implementation may vary - checking for error response)
+ assert response.status_code in (
+ 400,
+ 403,
+ 302,
+ ) # Error or redirect to error
+
+ async def test_consent_cookie_established_on_page_visit(
+ self, oauth_proxy_with_storage
+ ):
+ """Verify consent page establishes CSRF cookie."""
+ client = OAuthClientInformationFull(
+ client_id="cookie-test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:6060/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:6060/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="test-state",
+ code_challenge="",
+ scopes=["read"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+ parsed = urlparse(redirect_url)
+ query = parse_qs(parsed.query)
+ txn_id = query["txn_id"][0]
+
+ app = Starlette(routes=oauth_proxy_with_storage.get_routes())
+
+ with TestClient(app) as test_client:
+ # Visit consent page
+ response = test_client.get(
+ f"/consent?txn_id={txn_id}", follow_redirects=False
+ )
+
+ # Should set cookies for CSRF protection
+ assert response.status_code == 200
+ # Cookie may be set via Set-Cookie header
+ cookies = response.cookies
+ # Look for any CSRF-related cookie (implementation dependent)
+ assert len(cookies) > 0 or "csrf" in response.text.lower(), (
+ "Consent page should establish CSRF protection"
+ )
+
+
+class TestStoragePersistence:
+ """Tests for state persistence across storage backends."""
+
+ async def test_transaction_persists_after_retrieval(self, oauth_proxy_with_storage):
+ """Verify transaction can be retrieved multiple times (until deleted)."""
+ client = OAuthClientInformationFull(
+ client_id="persist-test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:5050/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:5050/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="persist-state",
+ code_challenge="persist-challenge",
+ code_challenge_method="S256",
+ scopes=["read"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+ parsed = urlparse(redirect_url)
+ query = parse_qs(parsed.query)
+ txn_id = query["txn_id"][0]
+
+ storage = oauth_proxy_with_storage._client_storage
+
+ # Retrieve transaction multiple times
+ txn1 = await storage.get(collection="mcp-oauth-transactions", key=txn_id)
+ assert txn1 is not None
+
+ txn2 = await storage.get(collection="mcp-oauth-transactions", key=txn_id)
+ assert txn2 is not None
+
+ # Should be the same data
+ assert txn1["client_id"] == txn2["client_id"]
+ assert txn1["client_state"] == txn2["client_state"]
+
+ async def test_storage_uses_pydantic_adapter(self, oauth_proxy_with_storage):
+ """Verify that PydanticAdapter serializes/deserializes correctly."""
+ from fastmcp.server.auth.oauth_proxy import OAuthTransaction
+
+ client = OAuthClientInformationFull(
+ client_id="pydantic-test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:4040/callback")],
+ )
+ await oauth_proxy_with_storage.register_client(client)
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:4040/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="pydantic-state",
+ code_challenge="pydantic-challenge",
+ code_challenge_method="S256",
+ scopes=["read", "write"],
+ )
+
+ redirect_url = await oauth_proxy_with_storage.authorize(client, params)
+ parsed = urlparse(redirect_url)
+ query = parse_qs(parsed.query)
+ txn_id = query["txn_id"][0]
+
+ # Retrieve using PydanticAdapter (which is what the proxy uses)
+ transaction_store = oauth_proxy_with_storage._transaction_store
+ txn_model = await transaction_store.get(key=txn_id)
+
+ # Should be a Pydantic model instance
+ assert isinstance(txn_model, OAuthTransaction)
+ assert txn_model.client_id == "pydantic-test-client"
+ assert txn_model.client_state == "pydantic-state"
+ assert txn_model.code_challenge == "pydantic-challenge"
+ assert txn_model.scopes == ["read", "write"]
+
+
+class TestConsentSecurity:
+ """Tests for consent page security features."""
+
+ async def test_consent_sets_xfo_header(self, oauth_proxy_https):
+ """Verify consent page sets X-Frame-Options header to prevent clickjacking."""
+ txn_id, _ = await _start_flow(
+ oauth_proxy_https, "client-a", "http://localhost:5001/callback"
+ )
+ app = Starlette(routes=oauth_proxy_https.get_routes())
+ with TestClient(app) as c:
+ r = c.get(f"/consent?txn_id={txn_id}")
+ assert r.status_code == 200
+ assert r.headers.get("X-Frame-Options") == "DENY"
+
+ async def test_deny_sets_cookie_and_redirects_with_error(self, oauth_proxy_https):
+ """Verify denying consent sets signed cookie and redirects with error."""
+ client_redirect = "http://localhost:5002/callback"
+ txn_id, _ = await _start_flow(oauth_proxy_https, "client-b", client_redirect)
+ app = Starlette(routes=oauth_proxy_https.get_routes())
+ with TestClient(app) as c:
+ consent = c.get(f"/consent?txn_id={txn_id}")
+ csrf = _extract_csrf(consent.text)
+ assert csrf
+ # Persist consent page cookies on client instance to avoid per-request deprecation
+ for k, v in consent.cookies.items():
+ c.cookies.set(k, v)
+ r = c.post(
+ "/consent/submit",
+ data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
+ follow_redirects=False,
+ )
+ assert r.status_code in (302, 303)
+ loc = r.headers.get("location", "")
+ parsed = urlparse(loc)
+ assert parsed.scheme == "http" and parsed.netloc.startswith("localhost")
+ q = parse_qs(parsed.query)
+ assert q.get("error") == ["access_denied"]
+ assert q.get("state") == ["client-state-xyz"]
+ # Signed denied cookie should be set
+ assert "MCP_DENIED_CLIENTS" in ";\n".join(
+ r.headers.get("set-cookie", "").splitlines()
+ )
+
+ async def test_approve_sets_cookie_and_redirects_to_upstream(
+ self, oauth_proxy_https
+ ):
+ """Verify approving consent sets signed cookie and redirects to upstream."""
+ txn_id, _ = await _start_flow(
+ oauth_proxy_https, "client-c", "http://localhost:5003/callback"
+ )
+ app = Starlette(routes=oauth_proxy_https.get_routes())
+ with TestClient(app) as c:
+ consent = c.get(f"/consent?txn_id={txn_id}")
+ csrf = _extract_csrf(consent.text)
+ assert csrf
+ for k, v in consent.cookies.items():
+ c.cookies.set(k, v)
+ r = c.post(
+ "/consent/submit",
+ data={"action": "approve", "txn_id": txn_id, "csrf_token": csrf},
+ follow_redirects=False,
+ )
+ assert r.status_code in (302, 303)
+ loc = r.headers.get("location", "")
+ assert loc.startswith("https://github.com/login/oauth/authorize")
+ assert f"state={txn_id}" in loc
+ # Signed approved cookie should be set with __Host- prefix for HTTPS
+ set_cookie = ";\n".join(r.headers.get("set-cookie", "").splitlines())
+ assert "__Host-MCP_APPROVED_CLIENTS" in set_cookie
+
+ async def test_tampered_cookie_is_ignored(self, oauth_proxy_https):
+ """Verify tampered approval cookie is ignored and consent page shown."""
+ txn_id, _ = await _start_flow(
+ oauth_proxy_https, "client-d", "http://localhost:5004/callback"
+ )
+ app = Starlette(routes=oauth_proxy_https.get_routes())
+ with TestClient(app) as c:
+ # Create a tampered cookie (invalid signature)
+ # Value format: payload.signature; using wrong signature to force failure
+ tampered_value = "W10=.invalidsig"
+ c.cookies.set("__Host-MCP_APPROVED_CLIENTS", tampered_value)
+ r = c.get(f"/consent?txn_id={txn_id}", follow_redirects=False)
+ # Should not auto-redirect to upstream; should show consent page
+ assert r.status_code == 200
+ # httpx returns a URL object; compare path or stringify
+ assert urlparse(str(r.request.url)).path == "/consent"
+
+ async def test_autoapprove_cookie_skips_consent(self, oauth_proxy_https):
+ """Verify valid approval cookie auto-approves and redirects to upstream."""
+ client_id = "client-e"
+ redirect = "http://localhost:5005/callback"
+ txn_id, _ = await _start_flow(oauth_proxy_https, client_id, redirect)
+ app = Starlette(routes=oauth_proxy_https.get_routes())
+ with TestClient(app) as c:
+ # Approve once to set approved cookie
+ consent = c.get(f"/consent?txn_id={txn_id}")
+ csrf = _extract_csrf(consent.text)
+ for k, v in consent.cookies.items():
+ c.cookies.set(k, v)
+ r = c.post(
+ "/consent/submit",
+ data={"action": "approve", "txn_id": txn_id, "csrf_token": csrf},
+ follow_redirects=False,
+ )
+ # Extract approved cookie value
+ set_cookie = ";\n".join(r.headers.get("set-cookie", "").splitlines())
+ m = re.search(r"__Host-MCP_APPROVED_CLIENTS=([^;]+)", set_cookie)
+ assert m, "approved cookie should be set"
+ approved_cookie = m.group(1)
+
+ # Start a new flow for the same client and redirect
+ new_txn, _ = await _start_flow(oauth_proxy_https, client_id, redirect)
+ # Should auto-redirect to upstream when visiting consent due to cookie
+ c.cookies.set("__Host-MCP_APPROVED_CLIENTS", approved_cookie)
+ r2 = c.get(f"/consent?txn_id={new_txn}", follow_redirects=False)
+ assert r2.status_code in (302, 303)
+ assert r2.headers.get("location", "").startswith(
+ "https://github.com/login/oauth/authorize"
+ )
diff --git a/tests/server/auth/test_oauth_proxy.py b/tests/server/auth/test_oauth_proxy.py
index c171547eb..13ceb5185 100644
--- a/tests/server/auth/test_oauth_proxy.py
+++ b/tests/server/auth/test_oauth_proxy.py
@@ -441,13 +441,16 @@ class TestOAuthProxyAuthorization:
"""Tests for OAuth proxy authorization flow."""
async def test_authorize_creates_transaction(self, oauth_proxy):
- """Test that authorize creates transaction and returns upstream URL."""
+ """Test that authorize creates transaction and redirects to consent."""
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:54321/callback")],
)
+ # Register client first (required for consent flow)
+ await oauth_proxy.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:54321/callback"),
redirect_uri_provided_explicitly=True,
@@ -463,18 +466,18 @@ class TestOAuthProxyAuthorization:
parsed = urlparse(redirect_url)
query_params = parse_qs(parsed.query)
- # Verify upstream URL structure
- assert "github.com/login/oauth/authorize" in redirect_url
- assert query_params["client_id"][0] == "test-client-id"
- assert query_params["response_type"][0] == "code"
- assert "state" in query_params # Transaction ID
+ # Should redirect to consent page
+ assert "/consent" in redirect_url
+ assert "txn_id" in query_params
- # Verify transaction was stored
- txn_id = query_params["state"][0]
- assert txn_id in oauth_proxy._oauth_transactions
- transaction = oauth_proxy._oauth_transactions[txn_id]
- assert transaction["client_id"] == "test-client"
- assert transaction["code_challenge"] == "challenge-abc"
+ # Verify transaction was stored with correct data
+ txn_id = query_params["txn_id"][0]
+ transaction = await oauth_proxy._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert transaction.client_id == "test-client"
+ assert transaction.code_challenge == "challenge-abc"
+ assert transaction.client_state == "client-state-123"
+ assert transaction.scopes == ["read", "write"]
class TestOAuthProxyPKCE:
@@ -512,6 +515,9 @@ class TestOAuthProxyPKCE:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy_with_pkce.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -523,16 +529,19 @@ class TestOAuthProxyPKCE:
redirect_url = await proxy_with_pkce.authorize(client, params)
query_params = parse_qs(urlparse(redirect_url).query)
- # Proxy should forward its own PKCE
- assert "code_challenge" in query_params
- assert query_params["code_challenge"][0] != "client_challenge"
- assert query_params["code_challenge_method"] == ["S256"]
+ # Should redirect to consent page
+ assert "/consent" in redirect_url
+ assert "txn_id" in query_params
# Transaction should store both challenges
- txn_id = query_params["state"][0]
- transaction = proxy_with_pkce._oauth_transactions[txn_id]
- assert transaction["code_challenge"] == "client_challenge" # Client's
- assert "proxy_code_verifier" in transaction # Proxy's verifier
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy_with_pkce._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert transaction.code_challenge == "client_challenge" # Client's
+ assert transaction.proxy_code_verifier is not None # Proxy's verifier
+ # Proxy code challenge is computed from verifier when building upstream URL
+ # Just verify the verifier exists and is different from client's challenge
+ assert len(transaction.proxy_code_verifier) > 0
async def test_pkce_forwarding_disabled(self, proxy_without_pkce):
"""Test that PKCE is not forwarded when disabled."""
@@ -542,6 +551,9 @@ class TestOAuthProxyPKCE:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy_without_pkce.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -553,15 +565,16 @@ class TestOAuthProxyPKCE:
redirect_url = await proxy_without_pkce.authorize(client, params)
query_params = parse_qs(urlparse(redirect_url).query)
- # No PKCE forwarded to upstream
- assert "code_challenge" not in query_params
- assert "code_challenge_method" not in query_params
+ # Should redirect to consent page
+ assert "/consent" in redirect_url
+ assert "txn_id" in query_params
- # Client's challenge still stored
- txn_id = query_params["state"][0]
- transaction = proxy_without_pkce._oauth_transactions[txn_id]
- assert transaction["code_challenge"] == "client_challenge"
- assert "proxy_code_verifier" not in transaction
+ # Client's challenge still stored, but no proxy PKCE
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy_without_pkce._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert transaction.code_challenge == "client_challenge"
+ assert transaction.proxy_code_verifier is None # No proxy PKCE when disabled
class TestOAuthProxyTokenEndpointAuth:
@@ -682,6 +695,9 @@ class TestOAuthProxyE2E:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy.register_client(client_info)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -690,28 +706,22 @@ class TestOAuthProxyE2E:
scopes=["read"],
)
- # Get authorization URL
+ # Get authorization URL (now returns consent redirect)
auth_url = await proxy.authorize(client_info, params)
- # Verify mock provider was called
- assert mock_oauth_provider.authorize_endpoint in auth_url
-
- # Verify state is present (transaction ID)
+ # Should redirect to consent page
+ assert "/consent" in auth_url
query_params = parse_qs(urlparse(auth_url).query)
- assert "state" in query_params
+ assert "txn_id" in query_params
- # Simulate authorization callback
- async with httpx.AsyncClient() as http_client:
- # This would normally redirect, but our mock returns the code
- response = await http_client.get(auth_url, follow_redirects=False)
-
- # Extract code from redirect location
- location = response.headers.get("location", "")
- callback_params = parse_qs(urlparse(location).query)
- auth_code = callback_params.get("code", [None])[0]
-
- assert auth_code is not None
- assert mock_oauth_provider.authorize_called
+ # Verify transaction was created with correct configuration
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert transaction.client_id == "test-client"
+ assert transaction.scopes == ["read"]
+ # Transaction ID itself is used as upstream state parameter
+ assert transaction.txn_id == txn_id
@pytest.mark.asyncio
async def test_token_refresh_with_mock_provider(self, mock_oauth_provider):
@@ -790,6 +800,9 @@ class TestOAuthProxyE2E:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -803,14 +816,20 @@ class TestOAuthProxyE2E:
auth_url = await proxy.authorize(client, params)
query_params = parse_qs(urlparse(auth_url).query)
- # Verify PKCE was forwarded (proxy's challenge, not client's)
- assert "code_challenge" in query_params
- assert query_params["code_challenge"][0] != "client_challenge_value"
+ # Should redirect to consent page
+ assert "/consent" in auth_url
+ assert "txn_id" in query_params
- # Transaction should have proxy's verifier
- txn_id = query_params["state"][0]
- transaction = proxy._oauth_transactions[txn_id]
- assert "proxy_code_verifier" in transaction
+ # Transaction should have proxy's PKCE verifier (different from client's)
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert (
+ transaction.code_challenge == "client_challenge_value"
+ ) # Client's challenge
+ assert transaction.proxy_code_verifier is not None # Proxy generated its own
+ # Proxy code challenge is computed from verifier when needed
+ assert len(transaction.proxy_code_verifier) > 0
class TestParameterForwarding:
@@ -850,6 +869,9 @@ class TestParameterForwarding:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy_without_extra_params.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -862,9 +884,17 @@ class TestParameterForwarding:
redirect_url = await proxy_without_extra_params.authorize(client, params)
query_params = parse_qs(urlparse(redirect_url).query)
- # Resource parameter should be forwarded to upstream
- assert "resource" in query_params
- assert query_params["resource"][0] == "https://api.example.com/v1"
+ # Should redirect to consent page
+ assert "/consent" in redirect_url
+ assert "txn_id" in query_params
+
+ # Resource parameter should be stored in transaction for upstream forwarding
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy_without_extra_params._transaction_store.get(
+ key=txn_id
+ )
+ assert transaction is not None
+ assert transaction.resource == "https://api.example.com/v1"
async def test_extra_authorize_params(self, proxy_with_extra_params):
"""Test that extra authorization parameters are included."""
@@ -874,6 +904,9 @@ class TestParameterForwarding:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy_with_extra_params.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -885,9 +918,19 @@ class TestParameterForwarding:
redirect_url = await proxy_with_extra_params.authorize(client, params)
query_params = parse_qs(urlparse(redirect_url).query)
- # Extra audience parameter should be included
- assert "audience" in query_params
- assert query_params["audience"][0] == "https://api.example.com"
+ # Should redirect to consent page
+ assert "/consent" in redirect_url
+ assert "txn_id" in query_params
+
+ # Extra audience parameter is configured at proxy level (not per-transaction)
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy_with_extra_params._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ # Verify proxy has extra params configured
+ assert (
+ proxy_with_extra_params._extra_authorize_params.get("audience")
+ == "https://api.example.com"
+ )
async def test_resource_and_extra_params_together(self, proxy_with_extra_params):
"""Test that both resource and extra params can be used together."""
@@ -897,6 +940,9 @@ class TestParameterForwarding:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy_with_extra_params.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -909,11 +955,19 @@ class TestParameterForwarding:
redirect_url = await proxy_with_extra_params.authorize(client, params)
query_params = parse_qs(urlparse(redirect_url).query)
- # Both resource and audience should be present
- assert "resource" in query_params
- assert query_params["resource"][0] == "https://resource.example.com"
- assert "audience" in query_params
- assert query_params["audience"][0] == "https://api.example.com"
+ # Should redirect to consent page
+ assert "/consent" in redirect_url
+ assert "txn_id" in query_params
+
+ # Resource stored in transaction, extra params configured at proxy level
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy_with_extra_params._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ assert transaction.resource == "https://resource.example.com"
+ assert (
+ proxy_with_extra_params._extra_authorize_params.get("audience")
+ == "https://api.example.com"
+ )
async def test_no_extra_params_when_not_configured(
self, proxy_without_extra_params
@@ -964,6 +1018,9 @@ class TestParameterForwarding:
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
+ # Register client first
+ await proxy.register_client(client)
+
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
@@ -975,10 +1032,20 @@ class TestParameterForwarding:
redirect_url = await proxy.authorize(client, params)
query_params = parse_qs(urlparse(redirect_url).query)
- # All extra parameters should be included
- assert query_params["audience"][0] == "https://api.example.com"
- assert query_params["prompt"][0] == "consent"
- assert query_params["max_age"][0] == "3600"
+ # Should redirect to consent page
+ assert "/consent" in redirect_url
+ assert "txn_id" in query_params
+
+ # All extra parameters configured at proxy level
+ txn_id = query_params["txn_id"][0]
+ transaction = await proxy._transaction_store.get(key=txn_id)
+ assert transaction is not None
+ # Verify proxy has all extra params configured
+ assert (
+ proxy._extra_authorize_params.get("audience") == "https://api.example.com"
+ )
+ assert proxy._extra_authorize_params.get("prompt") == "consent"
+ assert proxy._extra_authorize_params.get("max_age") == "3600"
@pytest.mark.asyncio
async def test_token_endpoint_invalid_client_error(self, jwt_verifier):
From c42c2b937edf9e5a13344b9d9bee5954604896a1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Niko=20D=C3=BCrr?=
<48563891+tradeqvest@users.noreply.github.com>
Date: Sat, 11 Oct 2025 23:17:53 +0200
Subject: [PATCH 3/4] Fix: add missing json_response parameter to
`run_http_async` (#2071)
---
src/fastmcp/server/server.py | 3 +++
1 file changed, 3 insertions(+)
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 91367360c..67cdbd9c0 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -1564,6 +1564,7 @@ class FastMCP(Generic[LifespanResultT]):
path: str | None = None,
uvicorn_config: dict[str, Any] | None = None,
middleware: list[ASGIMiddleware] | None = None,
+ json_response: bool | None = None,
stateless_http: bool | None = None,
) -> None:
"""Run the server using HTTP transport.
@@ -1576,6 +1577,7 @@ class FastMCP(Generic[LifespanResultT]):
path: Path for the endpoint (defaults to settings.streamable_http_path or settings.sse_path)
uvicorn_config: Additional configuration for the Uvicorn server
middleware: A list of middleware to apply to the app
+ json_response: Whether to use JSON response format (defaults to settings.json_response)
stateless_http: Whether to use stateless HTTP (defaults to settings.stateless_http)
"""
host = host or self._deprecated_settings.host
@@ -1588,6 +1590,7 @@ class FastMCP(Generic[LifespanResultT]):
path=path,
transport=transport,
middleware=middleware,
+ json_response=json_response,
stateless_http=stateless_http,
)
From 1f11b5e641d99aaf94fea6e522af291e5e587c8b Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Sat, 11 Oct 2025 18:28:21 -0400
Subject: [PATCH 4/4] Refactor mounting logic from managers to server (#2069)
---
.cursor/worktrees.json | 6 +
docs/servers/composition.mdx | 58 +++
.../component_manager/component_service.py | 12 +-
src/fastmcp/prompts/prompt_manager.py | 121 +----
src/fastmcp/resources/resource_manager.py | 175 +------
src/fastmcp/server/server.py | 435 +++++++++++++++---
src/fastmcp/tools/tool_manager.py | 126 +----
src/fastmcp/utilities/inspect.py | 37 +-
tests/client/test_client.py | 7 +-
.../server/openapi/test_advanced_behavior.py | 9 +-
tests/server/openapi/test_configuration.py | 29 +-
.../openapi/test_description_propagation.py | 3 +-
tests/server/test_mount.py | 122 ++++-
tests/server/test_server.py | 9 +-
tests/tools/test_tool_manager.py | 24 +-
tests/utilities/test_inspect.py | 192 ++++++++
16 files changed, 840 insertions(+), 525 deletions(-)
create mode 100644 .cursor/worktrees.json
diff --git a/.cursor/worktrees.json b/.cursor/worktrees.json
new file mode 100644
index 000000000..3321da678
--- /dev/null
+++ b/.cursor/worktrees.json
@@ -0,0 +1,6 @@
+{
+ "setup-worktree": [
+ "uv sync",
+ "uv run pre-commit install"
+ ]
+}
diff --git a/docs/servers/composition.mdx b/docs/servers/composition.mdx
index 63ac1f773..4ca1895ce 100644
--- a/docs/servers/composition.mdx
+++ b/docs/servers/composition.mdx
@@ -263,6 +263,64 @@ main_server.mount(remote_proxy, prefix="remote")
+## Tag Filtering with Composition
+
+
+
+When using `include_tags` or `exclude_tags` on a parent server, these filters apply **recursively** to all components, including those from mounted or imported servers. This allows you to control which components are exposed at the parent level, regardless of how your application is composed.
+
+```python
+import asyncio
+from fastmcp import FastMCP, Client
+
+# Create a subserver with tools tagged for different environments
+api_server = FastMCP(name="APIServer")
+
+@api_server.tool(tags={"production"})
+def prod_endpoint() -> str:
+ """Production-ready endpoint."""
+ return "Production data"
+
+@api_server.tool(tags={"development"})
+def dev_endpoint() -> str:
+ """Development-only endpoint."""
+ return "Debug data"
+
+# Mount the subserver with production tag filtering at parent level
+prod_app = FastMCP(name="ProductionApp", include_tags={"production"})
+prod_app.mount(api_server, prefix="api")
+
+# Test the filtering
+async def test_filtering():
+ async with Client(prod_app) as client:
+ tools = await client.list_tools()
+ print("Available tools:", [t.name for t in tools])
+ # Shows: ['api_prod_endpoint']
+ # The 'api_dev_endpoint' is filtered out
+
+ # Calling the filtered tool raises an error
+ try:
+ await client.call_tool("api_dev_endpoint")
+ except Exception as e:
+ print(f"Filtered tool not accessible: {e}")
+
+if __name__ == "__main__":
+ asyncio.run(test_filtering())
+```
+
+### How Recursive Filtering Works
+
+Tag filters apply in the following order:
+
+1. **Child Server Filters**: Each mounted/imported server first applies its own `include_tags`/`exclude_tags` to its components.
+2. **Parent Server Filters**: The parent server then applies its own `include_tags`/`exclude_tags` to all components, including those from child servers.
+
+This ensures that parent server tag policies act as a global policy for everything the parent server exposes, no matter how your application is composed.
+
+
+This filtering applies to both **listing** (e.g., `list_tools()`) and **execution** (e.g., `call_tool()`). Filtered components are neither visible nor executable through the parent server.
+
+
## Resource Prefix Formats
diff --git a/src/fastmcp/contrib/component_manager/component_service.py b/src/fastmcp/contrib/component_manager/component_service.py
index b63dff339..b23f96420 100644
--- a/src/fastmcp/contrib/component_manager/component_service.py
+++ b/src/fastmcp/contrib/component_manager/component_service.py
@@ -41,7 +41,7 @@ class ComponentService:
return tool
# 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._tool_manager._mounted_servers):
+ for mounted in reversed(self._server._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
@@ -70,7 +70,7 @@ class ComponentService:
return tool
# 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._tool_manager._mounted_servers):
+ for mounted in reversed(self._server._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
@@ -103,7 +103,7 @@ class ComponentService:
return template
# 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._resource_manager._mounted_servers):
+ for mounted in reversed(self._server._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
@@ -146,7 +146,7 @@ class ComponentService:
return template
# 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._resource_manager._mounted_servers):
+ for mounted in reversed(self._server._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
@@ -185,7 +185,7 @@ class ComponentService:
return prompt
# 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._prompt_manager._mounted_servers):
+ for mounted in reversed(self._server._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
@@ -213,7 +213,7 @@ class ComponentService:
return prompt
# 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._prompt_manager._mounted_servers):
+ for mounted in reversed(self._server._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
diff --git a/src/fastmcp/prompts/prompt_manager.py b/src/fastmcp/prompts/prompt_manager.py
index 56da9289c..1563a1831 100644
--- a/src/fastmcp/prompts/prompt_manager.py
+++ b/src/fastmcp/prompts/prompt_manager.py
@@ -2,7 +2,7 @@ from __future__ import annotations as _annotations
import warnings
from collections.abc import Awaitable, Callable
-from typing import TYPE_CHECKING, Any
+from typing import Any
from mcp import GetPromptResult
@@ -12,9 +12,6 @@ from fastmcp.prompts.prompt import FunctionPrompt, Prompt, PromptResult
from fastmcp.settings import DuplicateBehavior
from fastmcp.utilities.logging import get_logger
-if TYPE_CHECKING:
- from fastmcp.server.server import MountedServer
-
logger = get_logger(__name__)
@@ -27,7 +24,6 @@ class PromptManager:
mask_error_details: bool | None = None,
):
self._prompts: dict[str, Prompt] = {}
- self._mounted_servers: list[MountedServer] = []
self.mask_error_details = mask_error_details or settings.mask_error_details
# Default to "warn" if None is provided
@@ -42,54 +38,6 @@ class PromptManager:
self.duplicate_behavior = duplicate_behavior
- def mount(self, server: MountedServer) -> None:
- """Adds a mounted server as a source for prompts."""
- self._mounted_servers.append(server)
-
- async def _load_prompts(
- self, *, apply_filtering: bool = False
- ) -> dict[str, Prompt]:
- """
- The single, consolidated recursive method for fetching prompts. The 'apply_filtering'
- parameter determines the communication path.
-
- - apply_filtering=False: Manager-to-manager path for complete, unfiltered inventory
- - apply_filtering=True: Server-to-server path for filtered MCP requests
- """
- all_prompts: dict[str, Prompt] = {}
-
- for mounted in self._mounted_servers:
- try:
- if apply_filtering:
- # Use the server-to-server filtered path
- child_results = await mounted.server._list_prompts_middleware()
- else:
- # Use the manager-to-manager unfiltered path
- child_results = await mounted.server._prompt_manager.list_prompts()
-
- # The combination logic is the same for both paths
- child_dict = {p.key: p for p in child_results}
- if mounted.prefix:
- for prompt in child_dict.values():
- prefixed_prompt = prompt.model_copy(
- key=f"{mounted.prefix}_{prompt.key}"
- )
- all_prompts[prefixed_prompt.key] = prefixed_prompt
- else:
- all_prompts.update(child_dict)
- except Exception as e:
- # Skip failed mounts silently, matches existing behavior
- logger.warning(
- f"Failed to get prompts from server: {mounted.server.name!r}, mounted at: {mounted.prefix!r}: {e}"
- )
- if settings.mounted_components_raise_on_load_error:
- raise
- continue
-
- # Finally, add local prompts, which always take precedence
- all_prompts.update(self._prompts)
- return all_prompts
-
async def has_prompt(self, key: str) -> bool:
"""Check if a prompt exists."""
prompts = await self.get_prompts()
@@ -104,16 +52,9 @@ class PromptManager:
async def get_prompts(self) -> dict[str, Prompt]:
"""
- Gets the complete, unfiltered inventory of all prompts.
+ Gets the complete, unfiltered inventory of local prompts.
"""
- return await self._load_prompts(apply_filtering=False)
-
- async def list_prompts(self) -> list[Prompt]:
- """
- Lists all prompts, applying protocol filtering.
- """
- prompts_dict = await self._load_prompts(apply_filtering=True)
- return list(prompts_dict.values())
+ return dict(self._prompts)
def add_prompt_from_fn(
self,
@@ -162,46 +103,16 @@ class PromptManager:
Internal API for servers: Finds and renders a prompt, respecting the
filtered protocol path.
"""
- # 1. Check local prompts first. The server will have already applied its filter.
- if name in self._prompts:
- prompt = await self.get_prompt(name)
- if not prompt:
- raise NotFoundError(f"Unknown prompt: {name}")
-
- try:
- messages = await prompt.render(arguments)
- return GetPromptResult(
- description=prompt.description, messages=messages
- )
-
- # Pass through PromptErrors as-is
- except PromptError as e:
- logger.exception(f"Error rendering prompt {name!r}")
- raise e
-
- # Handle other exceptions
- except Exception as e:
- logger.exception(f"Error rendering prompt {name!r}")
- if self.mask_error_details:
- # Mask internal details
- raise PromptError(f"Error rendering prompt {name!r}") from e
- else:
- # Include original error details
- raise PromptError(f"Error rendering prompt {name!r}: {e}") from e
-
- # 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._mounted_servers):
- prompt_key = name
- if mounted.prefix:
- if name.startswith(f"{mounted.prefix}_"):
- prompt_key = name.removeprefix(f"{mounted.prefix}_")
- else:
- continue
- try:
- return await mounted.server._get_prompt_middleware(
- prompt_key, arguments
- )
- except NotFoundError:
- continue
-
- raise NotFoundError(f"Unknown prompt: {name}")
+ prompt = await self.get_prompt(name)
+ try:
+ messages = await prompt.render(arguments)
+ return GetPromptResult(description=prompt.description, messages=messages)
+ except PromptError as e:
+ logger.exception(f"Error rendering prompt {name!r}")
+ raise e
+ except Exception as e:
+ logger.exception(f"Error rendering prompt {name!r}")
+ if self.mask_error_details:
+ raise PromptError(f"Error rendering prompt {name!r}") from e
+ else:
+ raise PromptError(f"Error rendering prompt {name!r}: {e}") from e
diff --git a/src/fastmcp/resources/resource_manager.py b/src/fastmcp/resources/resource_manager.py
index d66a49ccf..07331ae18 100644
--- a/src/fastmcp/resources/resource_manager.py
+++ b/src/fastmcp/resources/resource_manager.py
@@ -5,7 +5,7 @@ from __future__ import annotations
import inspect
import warnings
from collections.abc import Callable
-from typing import TYPE_CHECKING, Any
+from typing import Any
from pydantic import AnyUrl
@@ -19,9 +19,6 @@ from fastmcp.resources.template import (
from fastmcp.settings import DuplicateBehavior
from fastmcp.utilities.logging import get_logger
-if TYPE_CHECKING:
- from fastmcp.server.server import MountedServer
-
logger = get_logger(__name__)
@@ -43,7 +40,6 @@ class ResourceManager:
"""
self._resources: dict[str, Resource] = {}
self._templates: dict[str, ResourceTemplate] = {}
- self._mounted_servers: list[MountedServer] = []
self.mask_error_details = mask_error_details or settings.mask_error_details
# Default to "warn" if None is provided
@@ -57,143 +53,13 @@ class ResourceManager:
)
self.duplicate_behavior = duplicate_behavior
- def mount(self, server: MountedServer) -> None:
- """Adds a mounted server as a source for resources and templates."""
- self._mounted_servers.append(server)
-
async def get_resources(self) -> dict[str, Resource]:
"""Get all registered resources, keyed by URI."""
- return await self._load_resources(apply_filtering=False)
+ return dict(self._resources)
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
"""Get all registered templates, keyed by URI template."""
- return await self._load_resource_templates(apply_filtering=False)
-
- async def _load_resources(
- self, *, apply_filtering: bool = False
- ) -> dict[str, Resource]:
- """
- The single, consolidated recursive method for fetching resources. The 'apply_filtering'
- parameter determines the communication path.
-
- - apply_filtering=False: Manager-to-manager path for complete, unfiltered inventory
- - apply_filtering=True: Server-to-server path for filtered MCP requests
- """
- all_resources: dict[str, Resource] = {}
-
- for mounted in self._mounted_servers:
- try:
- if apply_filtering:
- # Use the server-to-server filtered path
- child_resources_list = (
- await mounted.server._list_resources_middleware()
- )
- child_resources = {
- resource.key: resource for resource in child_resources_list
- }
- else:
- # Use the manager-to-manager unfiltered path
- child_resources = (
- await mounted.server._resource_manager.get_resources()
- )
-
- # Apply prefix if needed
- if mounted.prefix:
- from fastmcp.server.server import add_resource_prefix
-
- for uri, resource in child_resources.items():
- prefixed_uri = add_resource_prefix(
- uri, mounted.prefix, mounted.resource_prefix_format
- )
- # Create a copy of the resource with the prefixed key and name
- prefixed_resource = resource.model_copy(
- update={"name": f"{mounted.prefix}_{resource.name}"},
- key=prefixed_uri,
- )
- all_resources[prefixed_uri] = prefixed_resource
- else:
- all_resources.update(child_resources)
- except Exception as e:
- # Skip failed mounts silently, matches existing behavior
- logger.warning(
- f"Failed to get resources from server: {mounted.server.name!r}, mounted at: {mounted.prefix!r}: {e}"
- )
- if settings.mounted_components_raise_on_load_error:
- raise
- continue
-
- # Finally, add local resources, which always take precedence
- all_resources.update(self._resources)
- return all_resources
-
- async def _load_resource_templates(
- self, *, apply_filtering: bool = False
- ) -> dict[str, ResourceTemplate]:
- """
- The single, consolidated recursive method for fetching templates. The 'apply_filtering'
- parameter determines the communication path.
-
- - apply_filtering=False: Manager-to-manager path for complete, unfiltered inventory
- - apply_filtering=True: Server-to-server path for filtered MCP requests
- """
- all_templates: dict[str, ResourceTemplate] = {}
-
- for mounted in self._mounted_servers:
- try:
- if apply_filtering:
- # Use the server-to-server filtered path
- child_templates = (
- await mounted.server._list_resource_templates_middleware()
- )
- else:
- # Use the manager-to-manager unfiltered path
- child_templates = (
- await mounted.server._resource_manager.list_resource_templates()
- )
- child_dict = {template.key: template for template in child_templates}
-
- # Apply prefix if needed
- if mounted.prefix:
- from fastmcp.server.server import add_resource_prefix
-
- for uri_template, template in child_dict.items():
- prefixed_uri_template = add_resource_prefix(
- uri_template, mounted.prefix, mounted.resource_prefix_format
- )
- # Create a copy of the template with the prefixed key and name
- prefixed_template = template.model_copy(
- update={"name": f"{mounted.prefix}_{template.name}"},
- key=prefixed_uri_template,
- )
- all_templates[prefixed_uri_template] = prefixed_template
- else:
- all_templates.update(child_dict)
- except Exception as e:
- # Skip failed mounts silently, matches existing behavior
- logger.warning(
- f"Failed to get templates from server: {mounted.server.name!r}, mounted at: {mounted.prefix!r}: {e}"
- )
- if settings.mounted_components_raise_on_load_error:
- raise
- continue
-
- # Finally, add local templates, which always take precedence
- all_templates.update(self._templates)
- return all_templates
-
- async def list_resources(self) -> list[Resource]:
- """
- Lists all resources, applying protocol filtering.
- """
- resources_dict = await self._load_resources(apply_filtering=True)
- return list(resources_dict.values())
-
- async def list_resource_templates(self) -> list[ResourceTemplate]:
- """
- Lists all templates, applying protocol filtering.
- """
- templates_dict = await self._load_resource_templates(apply_filtering=True)
- return list(templates_dict.values())
+ return dict(self._templates)
def add_resource_or_template_from_fn(
self,
@@ -387,12 +253,12 @@ class ResourceManager:
uri_str = str(uri)
logger.debug("Getting resource", extra={"uri": uri_str})
- # First check concrete resources (local and mounted)
+ # First check concrete resources
resources = await self.get_resources()
if resource := resources.get(uri_str):
return resource
- # Then check templates (local and mounted) - use the utility function to match against storage keys
+ # Then check templates
templates = await self.get_resource_templates()
for storage_key, template in templates.items():
# Try to match against the storage key (which might be a custom key)
@@ -430,9 +296,6 @@ class ResourceManager:
# 1. Check local resources first. The server will have already applied its filter.
if uri_str in self._resources:
resource = await self.get_resource(uri_str)
- if not resource:
- raise NotFoundError(f"Resource {uri_str!r} not found")
-
try:
return await resource.read()
@@ -477,32 +340,4 @@ class ResourceManager:
f"Error reading resource from template {uri_str!r}: {e}"
) from e
- # 2. Check mounted servers using the filtered protocol path.
- from fastmcp.server.server import has_resource_prefix, remove_resource_prefix
-
- for mounted in reversed(self._mounted_servers):
- key = uri_str
- try:
- if mounted.prefix:
- if has_resource_prefix(
- key,
- mounted.prefix,
- mounted.resource_prefix_format,
- ):
- key = remove_resource_prefix(
- key,
- mounted.prefix,
- mounted.resource_prefix_format,
- )
- else:
- continue
-
- try:
- result = await mounted.server._read_resource_middleware(key)
- return result[0].content
- except NotFoundError:
- continue
- except NotFoundError:
- continue
-
raise NotFoundError(f"Resource {uri_str!r} not found.")
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 67cdbd9c0..81290e5cd 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -410,8 +410,24 @@ class FastMCP(Generic[LifespanResultT]):
self.middleware.append(middleware)
async def get_tools(self) -> dict[str, Tool]:
- """Get all registered tools, indexed by registered key."""
- return await self._tool_manager.get_tools()
+ """Get all tools (unfiltered), including mounted servers, indexed by key."""
+ all_tools = dict(await self._tool_manager.get_tools())
+
+ for mounted in self._mounted_servers:
+ try:
+ child_tools = await mounted.server.get_tools()
+ for key, tool in child_tools.items():
+ new_key = f"{mounted.prefix}_{key}" if mounted.prefix else key
+ all_tools[new_key] = tool.model_copy(key=new_key)
+ except Exception as e:
+ logger.warning(
+ f"Failed to get tools from mounted server {mounted.server.name!r}: {e}"
+ )
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return all_tools
async def get_tool(self, key: str) -> Tool:
tools = await self.get_tools()
@@ -420,8 +436,37 @@ class FastMCP(Generic[LifespanResultT]):
return tools[key]
async def get_resources(self) -> dict[str, Resource]:
- """Get all registered resources, indexed by registered key."""
- return await self._resource_manager.get_resources()
+ """Get all resources (unfiltered), including mounted servers, indexed by key."""
+ all_resources = dict(await self._resource_manager.get_resources())
+
+ for mounted in self._mounted_servers:
+ try:
+ child_resources = await mounted.server.get_resources()
+ for key, resource in child_resources.items():
+ new_key = (
+ add_resource_prefix(
+ key, mounted.prefix, mounted.resource_prefix_format
+ )
+ if mounted.prefix
+ else key
+ )
+ update = (
+ {"name": f"{mounted.prefix}_{resource.name}"}
+ if mounted.prefix and resource.name
+ else {}
+ )
+ all_resources[new_key] = resource.model_copy(
+ key=new_key, update=update
+ )
+ except Exception as e:
+ logger.warning(
+ f"Failed to get resources from mounted server {mounted.server.name!r}: {e}"
+ )
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return all_resources
async def get_resource(self, key: str) -> Resource:
resources = await self.get_resources()
@@ -430,8 +475,37 @@ class FastMCP(Generic[LifespanResultT]):
return resources[key]
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
- """Get all registered resource templates, indexed by registered key."""
- return await self._resource_manager.get_resource_templates()
+ """Get all resource templates (unfiltered), including mounted servers, indexed by key."""
+ all_templates = dict(await self._resource_manager.get_resource_templates())
+
+ for mounted in self._mounted_servers:
+ try:
+ child_templates = await mounted.server.get_resource_templates()
+ for key, template in child_templates.items():
+ new_key = (
+ add_resource_prefix(
+ key, mounted.prefix, mounted.resource_prefix_format
+ )
+ if mounted.prefix
+ else key
+ )
+ update = (
+ {"name": f"{mounted.prefix}_{template.name}"}
+ if mounted.prefix and template.name
+ else {}
+ )
+ all_templates[new_key] = template.model_copy(
+ key=new_key, update=update
+ )
+ except Exception as e:
+ logger.warning(
+ f"Failed to get resource templates from mounted server {mounted.server.name!r}: {e}"
+ )
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return all_templates
async def get_resource_template(self, key: str) -> ResourceTemplate:
"""Get a registered resource template by key."""
@@ -441,10 +515,24 @@ class FastMCP(Generic[LifespanResultT]):
return templates[key]
async def get_prompts(self) -> dict[str, Prompt]:
- """
- List all available prompts.
- """
- return await self._prompt_manager.get_prompts()
+ """Get all prompts (unfiltered), including mounted servers, indexed by key."""
+ all_prompts = dict(await self._prompt_manager.get_prompts())
+
+ for mounted in self._mounted_servers:
+ try:
+ child_prompts = await mounted.server.get_prompts()
+ for key, prompt in child_prompts.items():
+ new_key = f"{mounted.prefix}_{key}" if mounted.prefix else key
+ all_prompts[new_key] = prompt.model_copy(key=new_key)
+ except Exception as e:
+ logger.warning(
+ f"Failed to get prompts from mounted server {mounted.server.name!r}: {e}"
+ )
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return all_prompts
async def get_prompt(self, key: str) -> Prompt:
prompts = await self.get_prompts()
@@ -560,16 +648,45 @@ class FastMCP(Generic[LifespanResultT]):
context: MiddlewareContext[mcp.types.ListToolsRequest],
) -> list[Tool]:
"""
- List all available tools
+ List all available tools.
"""
- tools = await self._tool_manager.list_tools() # type: ignore[reportPrivateUsage]
+ # 1. Get local tools and filter them
+ local_tools = await self._tool_manager.get_tools()
+ filtered_local = [
+ tool for tool in local_tools.values() if self._should_enable_component(tool)
+ ]
- mcp_tools: list[Tool] = []
- for tool in tools:
- if self._should_enable_component(tool):
- mcp_tools.append(tool)
+ # 2. Get tools from mounted servers
+ # Mounted servers apply their own filtering, but we also apply parent's filtering
+ # Use a dict to implement "later wins" deduplication by key
+ all_tools: dict[str, Tool] = {tool.key: tool for tool in filtered_local}
- return mcp_tools
+ for mounted in self._mounted_servers:
+ try:
+ child_tools = await mounted.server._list_tools_middleware()
+ for tool in child_tools:
+ # Apply parent server's filtering to mounted components
+ if not self._should_enable_component(tool):
+ continue
+
+ key = tool.key
+ if mounted.prefix:
+ key = f"{mounted.prefix}_{tool.key}"
+ tool = tool.model_copy(key=key)
+ # Later mounted servers override earlier ones
+ all_tools[key] = tool
+ except Exception as e:
+ server_name = getattr(
+ getattr(mounted, "server", None), "name", repr(mounted)
+ )
+ logger.warning(
+ f"Failed to list tools from mounted server {server_name!r}: {e}"
+ )
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return list(all_tools.values())
async def _list_resources_mcp(self) -> list[MCPResource]:
"""
@@ -611,16 +728,54 @@ class FastMCP(Generic[LifespanResultT]):
context: MiddlewareContext[dict[str, Any]],
) -> list[Resource]:
"""
- List all available resources
+ List all available resources.
"""
- resources = await self._resource_manager.list_resources() # type: ignore[reportPrivateUsage]
+ # 1. Filter local resources
+ local_resources = await self._resource_manager.get_resources()
+ filtered_local = [
+ resource
+ for resource in local_resources.values()
+ if self._should_enable_component(resource)
+ ]
- mcp_resources: list[Resource] = []
- for resource in resources:
- if self._should_enable_component(resource):
- mcp_resources.append(resource)
+ # 2. Get from mounted servers with resource prefix handling
+ # Mounted servers apply their own filtering, but we also apply parent's filtering
+ # Use a dict to implement "later wins" deduplication by key
+ all_resources: dict[str, Resource] = {
+ resource.key: resource for resource in filtered_local
+ }
- return mcp_resources
+ for mounted in self._mounted_servers:
+ try:
+ child_resources = await mounted.server._list_resources_middleware()
+ for resource in child_resources:
+ # Apply parent server's filtering to mounted components
+ if not self._should_enable_component(resource):
+ continue
+
+ key = resource.key
+ if mounted.prefix:
+ key = add_resource_prefix(
+ resource.key,
+ mounted.prefix,
+ mounted.resource_prefix_format,
+ )
+ resource = resource.model_copy(
+ key=key,
+ update={"name": f"{mounted.prefix}_{resource.name}"},
+ )
+ # Later mounted servers override earlier ones
+ all_resources[key] = resource
+ except Exception as e:
+ server_name = getattr(
+ getattr(mounted, "server", None), "name", repr(mounted)
+ )
+ logger.warning(f"Failed to list resources from {server_name!r}: {e}")
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return list(all_resources.values())
async def _list_resource_templates_mcp(self) -> list[MCPResourceTemplate]:
"""
@@ -665,16 +820,58 @@ class FastMCP(Generic[LifespanResultT]):
context: MiddlewareContext[dict[str, Any]],
) -> list[ResourceTemplate]:
"""
- List all available resource templates
+ List all available resource templates.
"""
- templates = await self._resource_manager.list_resource_templates() # type: ignore[reportPrivateUsage]
+ # 1. Filter local templates
+ local_templates = await self._resource_manager.get_resource_templates()
+ filtered_local = [
+ template
+ for template in local_templates.values()
+ if self._should_enable_component(template)
+ ]
- mcp_templates: list[ResourceTemplate] = []
- for template in templates:
- if self._should_enable_component(template):
- mcp_templates.append(template)
+ # 2. Get from mounted servers with resource prefix handling
+ # Mounted servers apply their own filtering, but we also apply parent's filtering
+ # Use a dict to implement "later wins" deduplication by key
+ all_templates: dict[str, ResourceTemplate] = {
+ template.key: template for template in filtered_local
+ }
- return mcp_templates
+ for mounted in self._mounted_servers:
+ try:
+ child_templates = (
+ await mounted.server._list_resource_templates_middleware()
+ )
+ for template in child_templates:
+ # Apply parent server's filtering to mounted components
+ if not self._should_enable_component(template):
+ continue
+
+ key = template.key
+ if mounted.prefix:
+ key = add_resource_prefix(
+ template.key,
+ mounted.prefix,
+ mounted.resource_prefix_format,
+ )
+ template = template.model_copy(
+ key=key,
+ update={"name": f"{mounted.prefix}_{template.name}"},
+ )
+ # Later mounted servers override earlier ones
+ all_templates[key] = template
+ except Exception as e:
+ server_name = getattr(
+ getattr(mounted, "server", None), "name", repr(mounted)
+ )
+ logger.warning(
+ f"Failed to list resource templates from {server_name!r}: {e}"
+ )
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return list(all_templates.values())
async def _list_prompts_mcp(self) -> list[MCPPrompt]:
"""
@@ -717,16 +914,49 @@ class FastMCP(Generic[LifespanResultT]):
context: MiddlewareContext[mcp.types.ListPromptsRequest],
) -> list[Prompt]:
"""
- List all available prompts
+ List all available prompts.
"""
- prompts = await self._prompt_manager.list_prompts() # type: ignore[reportPrivateUsage]
+ # 1. Filter local prompts
+ local_prompts = await self._prompt_manager.get_prompts()
+ filtered_local = [
+ prompt
+ for prompt in local_prompts.values()
+ if self._should_enable_component(prompt)
+ ]
- mcp_prompts: list[Prompt] = []
- for prompt in prompts:
- if self._should_enable_component(prompt):
- mcp_prompts.append(prompt)
+ # 2. Get from mounted servers
+ # Mounted servers apply their own filtering, but we also apply parent's filtering
+ # Use a dict to implement "later wins" deduplication by key
+ all_prompts: dict[str, Prompt] = {
+ prompt.key: prompt for prompt in filtered_local
+ }
- return mcp_prompts
+ for mounted in self._mounted_servers:
+ try:
+ child_prompts = await mounted.server._list_prompts_middleware()
+ for prompt in child_prompts:
+ # Apply parent server's filtering to mounted components
+ if not self._should_enable_component(prompt):
+ continue
+
+ key = prompt.key
+ if mounted.prefix:
+ key = f"{mounted.prefix}_{prompt.key}"
+ prompt = prompt.model_copy(key=key)
+ # Later mounted servers override earlier ones
+ all_prompts[key] = prompt
+ except Exception as e:
+ server_name = getattr(
+ getattr(mounted, "server", None), "name", repr(mounted)
+ )
+ logger.warning(
+ f"Failed to list prompts from mounted server {server_name!r}: {e}"
+ )
+ if fastmcp.settings.mounted_components_raise_on_load_error:
+ raise
+ continue
+
+ return list(all_prompts.values())
async def _call_tool_mcp(
self, key: str, arguments: dict[str, Any]
@@ -781,13 +1011,40 @@ class FastMCP(Generic[LifespanResultT]):
"""
Call a tool
"""
- tool = await self._tool_manager.get_tool(context.message.name)
- if not self._should_enable_component(tool):
- raise NotFoundError(f"Unknown tool: {context.message.name!r}")
+ tool_name = context.message.name
- return await self._tool_manager.call_tool(
- key=context.message.name, arguments=context.message.arguments or {}
- )
+ # Try mounted servers in reverse order (later wins)
+ for mounted in reversed(self._mounted_servers):
+ try_name = tool_name
+ if mounted.prefix:
+ if not tool_name.startswith(f"{mounted.prefix}_"):
+ continue
+ try_name = tool_name[len(mounted.prefix) + 1 :]
+
+ try:
+ # First, get the tool to check if parent's filter allows it
+ tool = await mounted.server._tool_manager.get_tool(try_name)
+ if not self._should_enable_component(tool):
+ # Parent filter blocks this tool, continue searching
+ continue
+
+ return await mounted.server._call_tool_middleware(
+ try_name, context.message.arguments or {}
+ )
+ except NotFoundError:
+ continue
+
+ # Try local tools last (mounted servers override local)
+ try:
+ tool = await self._tool_manager.get_tool(tool_name)
+ if self._should_enable_component(tool):
+ return await self._tool_manager.call_tool(
+ key=tool_name, arguments=context.message.arguments or {}
+ )
+ except NotFoundError:
+ pass
+
+ raise NotFoundError(f"Unknown tool: {tool_name!r}")
async def _read_resource_mcp(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
"""
@@ -837,17 +1094,46 @@ class FastMCP(Generic[LifespanResultT]):
"""
Read a resource
"""
- resource = await self._resource_manager.get_resource(context.message.uri)
- if not self._should_enable_component(resource):
- raise NotFoundError(f"Unknown resource: {str(context.message.uri)!r}")
+ uri_str = str(context.message.uri)
- content = await self._resource_manager.read_resource(context.message.uri)
- return [
- ReadResourceContents(
- content=content,
- mime_type=resource.mime_type,
- )
- ]
+ # Try mounted servers in reverse order (later wins)
+ for mounted in reversed(self._mounted_servers):
+ key = uri_str
+ if mounted.prefix:
+ if not has_resource_prefix(
+ key, mounted.prefix, mounted.resource_prefix_format
+ ):
+ continue
+ key = remove_resource_prefix(
+ key, mounted.prefix, mounted.resource_prefix_format
+ )
+
+ try:
+ # First, get the resource to check if parent's filter allows it
+ resource = await mounted.server._resource_manager.get_resource(key)
+ if not self._should_enable_component(resource):
+ # Parent filter blocks this resource, continue searching
+ continue
+ result = await mounted.server._read_resource_middleware(key)
+ return result
+ except NotFoundError:
+ continue
+
+ # Try local resources last (mounted servers override local)
+ try:
+ resource = await self._resource_manager.get_resource(uri_str)
+ if self._should_enable_component(resource):
+ content = await self._resource_manager.read_resource(uri_str)
+ return [
+ ReadResourceContents(
+ content=content,
+ mime_type=resource.mime_type,
+ )
+ ]
+ except NotFoundError:
+ pass
+
+ raise NotFoundError(f"Unknown resource: {uri_str!r}")
async def _get_prompt_mcp(
self, name: str, arguments: dict[str, Any] | None = None
@@ -893,13 +1179,39 @@ class FastMCP(Generic[LifespanResultT]):
self,
context: MiddlewareContext[mcp.types.GetPromptRequestParams],
) -> GetPromptResult:
- prompt = await self._prompt_manager.get_prompt(context.message.name)
- if not self._should_enable_component(prompt):
- raise NotFoundError(f"Unknown prompt: {context.message.name!r}")
+ name = context.message.name
- return await self._prompt_manager.render_prompt(
- name=context.message.name, arguments=context.message.arguments
- )
+ # Try mounted servers in reverse order (later wins)
+ for mounted in reversed(self._mounted_servers):
+ try_name = name
+ if mounted.prefix:
+ if not name.startswith(f"{mounted.prefix}_"):
+ continue
+ try_name = name[len(mounted.prefix) + 1 :]
+
+ try:
+ # First, get the prompt to check if parent's filter allows it
+ prompt = await mounted.server._prompt_manager.get_prompt(try_name)
+ if not self._should_enable_component(prompt):
+ # Parent filter blocks this prompt, continue searching
+ continue
+ return await mounted.server._get_prompt_middleware(
+ try_name, context.message.arguments
+ )
+ except NotFoundError:
+ continue
+
+ # Try local prompts last (mounted servers override local)
+ try:
+ prompt = await self._prompt_manager.get_prompt(name)
+ if self._should_enable_component(prompt):
+ return await self._prompt_manager.render_prompt(
+ name=name, arguments=context.message.arguments
+ )
+ except NotFoundError:
+ pass
+
+ raise NotFoundError(f"Unknown prompt: {name!r}")
def add_tool(self, tool: Tool) -> Tool:
"""Add a tool to the server.
@@ -1904,9 +2216,6 @@ class FastMCP(Generic[LifespanResultT]):
resource_prefix_format=self.resource_prefix_format,
)
self._mounted_servers.append(mounted_server)
- self._tool_manager.mount(mounted_server)
- self._resource_manager.mount(mounted_server)
- self._prompt_manager.mount(mounted_server)
async def import_server(
self,
diff --git a/src/fastmcp/tools/tool_manager.py b/src/fastmcp/tools/tool_manager.py
index 02797f55f..cd74dc18e 100644
--- a/src/fastmcp/tools/tool_manager.py
+++ b/src/fastmcp/tools/tool_manager.py
@@ -2,7 +2,7 @@ from __future__ import annotations
import warnings
from collections.abc import Callable
-from typing import TYPE_CHECKING, Any
+from typing import Any
from mcp.types import ToolAnnotations
@@ -16,9 +16,6 @@ from fastmcp.tools.tool_transform import (
)
from fastmcp.utilities.logging import get_logger
-if TYPE_CHECKING:
- from fastmcp.server.server import MountedServer
-
logger = get_logger(__name__)
@@ -32,7 +29,6 @@ class ToolManager:
transformations: dict[str, ToolTransformConfig] | None = None,
):
self._tools: dict[str, Tool] = {}
- self._mounted_servers: list[MountedServer] = []
self.mask_error_details = mask_error_details or settings.mask_error_details
self.transformations = transformations or {}
@@ -48,56 +44,12 @@ class ToolManager:
self.duplicate_behavior = duplicate_behavior
- def mount(self, server: MountedServer) -> None:
- """Adds a mounted server as a source for tools."""
- self._mounted_servers.append(server)
-
- async def _load_tools(self, *, apply_filtering: bool = False) -> dict[str, Tool]:
- """
- The single, consolidated recursive method for fetching tools. The 'apply_filtering'
- parameter determines the communication path.
-
- - apply_filtering=False: Manager-to-manager path for complete, unfiltered inventory
- - apply_filtering=True: Server-to-server path for filtered MCP requests
- """
- all_tools: dict[str, Tool] = {}
-
- for mounted in self._mounted_servers:
- try:
- if apply_filtering:
- # Use the server-to-server filtered path
- child_results = await mounted.server._list_tools_middleware()
- else:
- # Use the manager-to-manager unfiltered path
- child_results = await mounted.server._tool_manager.list_tools()
-
- # The combination logic is the same for both paths
- child_dict = {t.key: t for t in child_results}
- if mounted.prefix:
- for tool in child_dict.values():
- prefixed_tool = tool.model_copy(
- key=f"{mounted.prefix}_{tool.key}"
- )
- all_tools[prefixed_tool.key] = prefixed_tool
- else:
- all_tools.update(child_dict)
- except Exception as e:
- # Skip failed mounts silently, matches existing behavior
- logger.warning(
- f"Failed to get tools from server: {mounted.server.name!r}, mounted at: {mounted.prefix!r}: {e}"
- )
- if settings.mounted_components_raise_on_load_error:
- raise
- continue
-
- # Finally, add local tools, which always take precedence
- all_tools.update(self._tools)
-
+ async def _load_tools(self) -> dict[str, Tool]:
+ """Return this manager's local tools with transformations applied."""
transformed_tools = apply_transformations_to_tools(
- tools=all_tools,
+ tools=self._tools,
transformations=self.transformations,
)
-
return transformed_tools
async def has_tool(self, key: str) -> bool:
@@ -114,25 +66,9 @@ class ToolManager:
async def get_tools(self) -> dict[str, Tool]:
"""
- Gets the complete, unfiltered inventory of all tools.
+ Gets the complete, unfiltered inventory of local tools.
"""
- return await self._load_tools(apply_filtering=False)
-
- async def list_tools(self) -> list[Tool]:
- """
- Lists all tools, applying protocol filtering.
- """
- tools_dict = await self._load_tools(apply_filtering=True)
- return list(tools_dict.values())
-
- @property
- def _tools_transformed(self) -> list[str]:
- """Get the local tools."""
-
- return [
- transformation.name or tool_name
- for tool_name, transformation in self.transformations.items()
- ]
+ return await self._load_tools()
def add_tool_from_fn(
self,
@@ -214,41 +150,15 @@ class ToolManager:
Internal API for servers: Finds and calls a tool, respecting the
filtered protocol path.
"""
- # 1. Check local tools first. The server will have already applied its filter.
- if key in self._tools or key in self._tools_transformed:
- tool = await self.get_tool(key)
- if not tool:
- raise NotFoundError(f"Tool {key!r} not found")
-
- try:
- return await tool.run(arguments)
-
- # raise ToolErrors as-is
- except ToolError as e:
- logger.exception(f"Error calling tool {key!r}")
- raise e
-
- # Handle other exceptions
- except Exception as e:
- logger.exception(f"Error calling tool {key!r}")
- if self.mask_error_details:
- # Mask internal details
- raise ToolError(f"Error calling tool {key!r}") from e
- else:
- # Include original error details
- raise ToolError(f"Error calling tool {key!r}: {e}") from e
-
- # 2. Check mounted servers using the filtered protocol path.
- for mounted in reversed(self._mounted_servers):
- tool_key = key
- if mounted.prefix:
- if key.startswith(f"{mounted.prefix}_"):
- tool_key = key.removeprefix(f"{mounted.prefix}_")
- else:
- continue
- try:
- return await mounted.server._call_tool_middleware(tool_key, arguments)
- except NotFoundError:
- continue
-
- raise NotFoundError(f"Tool {key!r} not found.")
+ tool = await self.get_tool(key)
+ try:
+ return await tool.run(arguments)
+ except ToolError as e:
+ logger.exception(f"Error calling tool {key!r}")
+ raise e
+ except Exception as e:
+ logger.exception(f"Error calling tool {key!r}")
+ if self.mask_error_details:
+ raise ToolError(f"Error calling tool {key!r}") from e
+ else:
+ raise ToolError(f"Error calling tool {key!r}: {e}") from e
diff --git a/src/fastmcp/utilities/inspect.py b/src/fastmcp/utilities/inspect.py
index f3303a8f4..f68921674 100644
--- a/src/fastmcp/utilities/inspect.py
+++ b/src/fastmcp/utilities/inspect.py
@@ -104,21 +104,20 @@ async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
Returns:
FastMCPInfo dataclass containing the extracted information
"""
- # Get all the components using FastMCP2's direct methods
- tools_dict = await mcp.get_tools()
- prompts_dict = await mcp.get_prompts()
- resources_dict = await mcp.get_resources()
- templates_dict = await mcp.get_resource_templates()
+ # Get all components via middleware to respect filtering and preserve metadata
+ tools_list = await mcp._list_tools_middleware()
+ prompts_list = await mcp._list_prompts_middleware()
+ resources_list = await mcp._list_resources_middleware()
+ templates_list = await mcp._list_resource_templates_middleware()
# Extract detailed tool information
tool_infos = []
- for key, tool in tools_dict.items():
- # Convert to MCP tool to get input schema
- mcp_tool = tool.to_mcp_tool(name=key)
+ for tool in tools_list:
+ mcp_tool = tool.to_mcp_tool(name=tool.key)
tool_infos.append(
ToolInfo(
- key=key,
- name=tool.name or key,
+ key=tool.key,
+ name=tool.name or tool.key,
description=tool.description,
input_schema=mcp_tool.inputSchema if mcp_tool.inputSchema else {},
output_schema=tool.output_schema,
@@ -132,11 +131,11 @@ async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
# Extract detailed prompt information
prompt_infos = []
- for key, prompt in prompts_dict.items():
+ for prompt in prompts_list:
prompt_infos.append(
PromptInfo(
- key=key,
- name=prompt.name or key,
+ key=prompt.key,
+ name=prompt.name or prompt.key,
description=prompt.description,
arguments=[arg.model_dump() for arg in prompt.arguments]
if prompt.arguments
@@ -150,11 +149,11 @@ async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
# Extract detailed resource information
resource_infos = []
- for key, resource in resources_dict.items():
+ for resource in resources_list:
resource_infos.append(
ResourceInfo(
- key=key,
- uri=key, # For v2, key is the URI
+ key=resource.key,
+ uri=resource.key,
name=resource.name,
description=resource.description,
mime_type=resource.mime_type,
@@ -170,11 +169,11 @@ async def inspect_fastmcp_v2(mcp: FastMCP[Any]) -> FastMCPInfo:
# Extract detailed template information
template_infos = []
- for key, template in templates_dict.items():
+ for template in templates_list:
template_infos.append(
TemplateInfo(
- key=key,
- uri_template=key, # For v2, key is the URI template
+ key=template.key,
+ uri_template=template.key,
name=template.name,
description=template.description,
mime_type=template.mime_type,
diff --git a/tests/client/test_client.py b/tests/client/test_client.py
index 9a41c7600..009103233 100644
--- a/tests/client/test_client.py
+++ b/tests/client/test_client.py
@@ -941,12 +941,7 @@ class TestInferTransport:
transport = infer_transport(config)
assert isinstance(transport, MCPConfigTransport)
assert isinstance(transport.transport, FastMCPTransport)
- assert (
- len(
- cast(FastMCP, transport.transport.server)._tool_manager._mounted_servers
- )
- == 2
- )
+ assert len(cast(FastMCP, transport.transport.server)._mounted_servers) == 2
def test_infer_fastmcp_server(self, fastmcp_server):
"""FastMCP server instances should infer to FastMCPTransport."""
diff --git a/tests/server/openapi/test_advanced_behavior.py b/tests/server/openapi/test_advanced_behavior.py
index 81cd34faf..159c47900 100644
--- a/tests/server/openapi/test_advanced_behavior.py
+++ b/tests/server/openapi/test_advanced_behavior.py
@@ -102,7 +102,8 @@ class TestTagTransfer:
):
"""Test that tags from OpenAPI routes are correctly transferred to Tools."""
# Get internal tools directly (not the public API which returns MCP.Content)
- tools = await fastmcp_openapi_server._tool_manager.list_tools()
+ tools_dict = await fastmcp_openapi_server._tool_manager.get_tools()
+ tools = list(tools_dict.values())
# Find the create_user and update_user_name tools
create_user_tool = next(
@@ -201,7 +202,8 @@ class TestReprMethods:
async def test_openapi_tool_repr(self, fastmcp_openapi_server: FastMCPOpenAPI):
"""Test that OpenAPITool's __repr__ method works without recursion errors."""
- tools = await fastmcp_openapi_server._tool_manager.list_tools()
+ tools_dict = await fastmcp_openapi_server._tool_manager.get_tools()
+ tools = list(tools_dict.values())
tool = next(iter(tools))
# Verify repr doesn't cause recursion and contains expected elements
@@ -276,7 +278,8 @@ class TestEnumHandling:
)
# Get the tools from the server
- tools = await server._tool_manager.list_tools()
+ tools_dict = await server._tool_manager.get_tools()
+ tools = list(tools_dict.values())
# Find the read_item tool
read_item_tool = next((t for t in tools if t.name == "read_item_items"), None)
diff --git a/tests/server/openapi/test_configuration.py b/tests/server/openapi/test_configuration.py
index 8000ac35e..38e2c6f05 100644
--- a/tests/server/openapi/test_configuration.py
+++ b/tests/server/openapi/test_configuration.py
@@ -65,8 +65,8 @@ class TestRouteMapWildcard:
)
# All operations should be mapped to tools
- tools = await mcp._tool_manager.list_tools()
- tool_names = {tool.name for tool in tools}
+ tools_dict = await mcp._tool_manager.get_tools()
+ tool_names = {tool.name for tool in tools_dict.values()}
# Check that all 4 operations became tools
expected_tools = {"getUsers", "createUser", "getPosts", "createPost"}
@@ -382,8 +382,8 @@ class TestMCPNames:
)
# Check tools use custom names
- tools = await server._tool_manager.list_tools()
- tool_names = {tool.name for tool in tools}
+ tools_dict = await server._tool_manager.get_tools()
+ tool_names = {tool.name for tool in tools_dict.values()}
assert "admin_create_user" in tool_names
# Check resource templates use custom names
@@ -412,7 +412,8 @@ class TestMCPNames:
route_maps=GET_ROUTE_MAPS,
)
- tools = await server._tool_manager.list_tools()
+ tools_dict = await server._tool_manager.get_tools()
+ tools = list(tools_dict.values())
tool_names = {tool.name for tool in tools}
templates_dict = await server._resource_manager.get_resource_templates()
@@ -468,8 +469,8 @@ class TestMCPNames:
# Check all component types
all_names = []
- tools = await server._tool_manager.list_tools()
- all_names.extend(tool.name for tool in tools)
+ tools_dict = await server._tool_manager.get_tools()
+ all_names.extend(tool.name for tool in tools_dict.values())
resources_dict = await server._resource_manager.get_resources()
all_names.extend(resource.name for resource in resources_dict.values())
@@ -501,8 +502,8 @@ class TestMCPNames:
mcp_names=mcp_names,
)
- tools = await server._tool_manager.list_tools()
- tool_names = {tool.name for tool in tools}
+ tools_dict = await server._tool_manager.get_tools()
+ tool_names = {tool.name for tool in tools_dict.values()}
assert "openapi_user_list" in tool_names
async def test_mcp_names_with_from_fastapi_classmethod(self):
@@ -533,8 +534,8 @@ class TestMCPNames:
mcp_names=mcp_names,
)
- tools = await server._tool_manager.list_tools()
- tool_names = {tool.name for tool in tools}
+ tools_dict = await server._tool_manager.get_tools()
+ tool_names = {tool.name for tool in tools_dict.values()}
assert "fastapi_create_user" in tool_names
assert "fastapi_user_list" in tool_names
@@ -636,7 +637,8 @@ class TestRouteMapMCPTags:
)
# Get the POST tool
- tools = await server._tool_manager.list_tools()
+ tools_dict = await server._tool_manager.get_tools()
+ tools = list(tools_dict.values())
create_user_tool = next((t for t in tools if "create_user" in t.name), None)
assert create_user_tool is not None, "create_user tool not found"
@@ -752,7 +754,8 @@ class TestRouteMapMCPTags:
)
# Check tool tags
- tools = await server._tool_manager.list_tools()
+ tools_dict = await server._tool_manager.get_tools()
+ tools = list(tools_dict.values())
create_tool = next((t for t in tools if "create_user" in t.name), None)
assert create_tool is not None
assert "write-operation" in create_tool.tags
diff --git a/tests/server/openapi/test_description_propagation.py b/tests/server/openapi/test_description_propagation.py
index 04bf21c01..a851a19a8 100644
--- a/tests/server/openapi/test_description_propagation.py
+++ b/tests/server/openapi/test_description_propagation.py
@@ -567,7 +567,8 @@ class TestFastAPIDescriptionPropagation:
print(f" Template: {name}, Name attribute: {template.name}")
print("\nDEBUG - Tools created:")
- tools = await server._tool_manager.list_tools()
+ tools_dict = await server._tool_manager.get_tools()
+ tools = list(tools_dict.values())
for tool in tools:
print(f" Tool: {tool.name}")
diff --git a/tests/server/test_mount.py b/tests/server/test_mount.py
index 8fde83446..e3fd3abd1 100644
--- a/tests/server/test_mount.py
+++ b/tests/server/test_mount.py
@@ -329,18 +329,15 @@ class TestMultipleServerMount:
record.message for record in caplog.records if record.levelname == "WARNING"
]
assert any(
- "Failed to get tools from server: 'unreachable_proxy', mounted at: 'unreachable'"
- in msg
+ "Failed to list tools from mounted server 'unreachable_proxy'" in msg
for msg in warning_messages
)
assert any(
- "Failed to get resources from server: 'unreachable_proxy', mounted at: 'unreachable'"
- in msg
+ "Failed to list resources from 'unreachable_proxy'" in msg
for msg in warning_messages
)
assert any(
- "Failed to get prompts from server: 'unreachable_proxy', mounted at: 'unreachable'"
- in msg
+ "Failed to list prompts from mounted server 'unreachable_proxy'" in msg
for msg in warning_messages
)
@@ -871,7 +868,7 @@ class TestAsProxyKwarg:
sub = FastMCP("Sub")
mcp.mount(sub, "sub")
- assert mcp._tool_manager._mounted_servers[0].server is sub
+ assert mcp._mounted_servers[0].server is sub
async def test_as_proxy_false(self):
mcp = FastMCP("Main")
@@ -879,7 +876,7 @@ class TestAsProxyKwarg:
mcp.mount(sub, "sub", as_proxy=False)
- assert mcp._tool_manager._mounted_servers[0].server is sub
+ assert mcp._mounted_servers[0].server is sub
async def test_as_proxy_true(self):
mcp = FastMCP("Main")
@@ -887,8 +884,8 @@ class TestAsProxyKwarg:
mcp.mount(sub, "sub", as_proxy=True)
- assert mcp._tool_manager._mounted_servers[0].server is not sub
- assert isinstance(mcp._tool_manager._mounted_servers[0].server, FastMCPProxy)
+ assert mcp._mounted_servers[0].server is not sub
+ assert isinstance(mcp._mounted_servers[0].server, FastMCPProxy)
async def test_as_proxy_defaults_true_if_lifespan(self):
@asynccontextmanager
@@ -900,8 +897,8 @@ class TestAsProxyKwarg:
mcp.mount(sub, "sub")
- assert mcp._tool_manager._mounted_servers[0].server is not sub
- assert isinstance(mcp._tool_manager._mounted_servers[0].server, FastMCPProxy)
+ assert mcp._mounted_servers[0].server is not sub
+ assert isinstance(mcp._mounted_servers[0].server, FastMCPProxy)
async def test_as_proxy_ignored_for_proxy_mounts_default(self):
mcp = FastMCP("Main")
@@ -910,7 +907,7 @@ class TestAsProxyKwarg:
mcp.mount(sub_proxy, "sub")
- assert mcp._tool_manager._mounted_servers[0].server is sub_proxy
+ assert mcp._mounted_servers[0].server is sub_proxy
async def test_as_proxy_ignored_for_proxy_mounts_false(self):
mcp = FastMCP("Main")
@@ -919,7 +916,7 @@ class TestAsProxyKwarg:
mcp.mount(sub_proxy, "sub", as_proxy=False)
- assert mcp._tool_manager._mounted_servers[0].server is sub_proxy
+ assert mcp._mounted_servers[0].server is sub_proxy
async def test_as_proxy_ignored_for_proxy_mounts_true(self):
mcp = FastMCP("Main")
@@ -928,7 +925,7 @@ class TestAsProxyKwarg:
mcp.mount(sub_proxy, "sub", as_proxy=True)
- assert mcp._tool_manager._mounted_servers[0].server is sub_proxy
+ assert mcp._mounted_servers[0].server is sub_proxy
async def test_as_proxy_mounts_still_have_live_link(self):
mcp = FastMCP("Main")
@@ -1024,6 +1021,101 @@ class TestResourceNamePrefixing:
assert template.name == "prefix_user_template"
+class TestParentTagFiltering:
+ """Test that parent server tag filters apply recursively to mounted servers."""
+
+ async def test_parent_include_tags_filters_mounted_tools(self):
+ """Test that parent include_tags filters out non-matching mounted tools."""
+ parent = FastMCP("Parent", include_tags={"allowed"})
+ mounted = FastMCP("Mounted")
+
+ @mounted.tool(tags={"allowed"})
+ def allowed_tool() -> str:
+ return "allowed"
+
+ @mounted.tool(tags={"blocked"})
+ def blocked_tool() -> str:
+ return "blocked"
+
+ parent.mount(mounted)
+
+ async with Client(parent) as client:
+ tools = await client.list_tools()
+ tool_names = {t.name for t in tools}
+ assert "allowed_tool" in tool_names
+ assert "blocked_tool" not in tool_names
+
+ # Verify execution also respects filters
+ result = await client.call_tool("allowed_tool", {})
+ assert result.data == "allowed"
+
+ with pytest.raises(Exception, match="Unknown tool"):
+ await client.call_tool("blocked_tool", {})
+
+ async def test_parent_exclude_tags_filters_mounted_tools(self):
+ """Test that parent exclude_tags filters out matching mounted tools."""
+ parent = FastMCP("Parent", exclude_tags={"blocked"})
+ mounted = FastMCP("Mounted")
+
+ @mounted.tool(tags={"production"})
+ def production_tool() -> str:
+ return "production"
+
+ @mounted.tool(tags={"blocked"})
+ def blocked_tool() -> str:
+ return "blocked"
+
+ parent.mount(mounted)
+
+ async with Client(parent) as client:
+ tools = await client.list_tools()
+ tool_names = {t.name for t in tools}
+ assert "production_tool" in tool_names
+ assert "blocked_tool" not in tool_names
+
+ async def test_parent_filters_apply_to_mounted_resources(self):
+ """Test that parent tag filters apply to mounted resources."""
+ parent = FastMCP("Parent", include_tags={"allowed"})
+ mounted = FastMCP("Mounted")
+
+ @mounted.resource("resource://allowed", tags={"allowed"})
+ def allowed_resource() -> str:
+ return "allowed"
+
+ @mounted.resource("resource://blocked", tags={"blocked"})
+ def blocked_resource() -> str:
+ return "blocked"
+
+ parent.mount(mounted)
+
+ async with Client(parent) as client:
+ resources = await client.list_resources()
+ resource_uris = {str(r.uri) for r in resources}
+ assert "resource://allowed" in resource_uris
+ assert "resource://blocked" not in resource_uris
+
+ async def test_parent_filters_apply_to_mounted_prompts(self):
+ """Test that parent tag filters apply to mounted prompts."""
+ parent = FastMCP("Parent", exclude_tags={"blocked"})
+ mounted = FastMCP("Mounted")
+
+ @mounted.prompt(tags={"allowed"})
+ def allowed_prompt() -> str:
+ return "allowed"
+
+ @mounted.prompt(tags={"blocked"})
+ def blocked_prompt() -> str:
+ return "blocked"
+
+ parent.mount(mounted)
+
+ async with Client(parent) as client:
+ prompts = await client.list_prompts()
+ prompt_names = {p.name for p in prompts}
+ assert "allowed_prompt" in prompt_names
+ assert "blocked_prompt" not in prompt_names
+
+
class TestCustomRouteForwarding:
"""Test that custom HTTP routes from mounted servers are forwarded."""
diff --git a/tests/server/test_server.py b/tests/server/test_server.py
index 819cb2146..1b712be81 100644
--- a/tests/server/test_server.py
+++ b/tests/server/test_server.py
@@ -307,10 +307,11 @@ class TestToolDecorator:
def sample_tool(x: int) -> int:
return x * 2
- # Verify the tags were set correctly
- tools = await mcp._tool_manager.list_tools()
- assert len(tools) == 1
- assert tools[0].tags == {"example", "test-tag"}
+ # Verify the tags were set correctly (local inventory)
+ tools_dict = await mcp._tool_manager.get_tools()
+ assert len(tools_dict) == 1
+ only_tool = next(iter(tools_dict.values()))
+ assert only_tool.tags == {"example", "test-tag"}
async def test_add_tool_with_custom_name(self):
"""Test adding a tool with a custom name using server.add_tool()."""
diff --git a/tests/tools/test_tool_manager.py b/tests/tools/test_tool_manager.py
index 10d0b7972..e39e0a7b8 100644
--- a/tests/tools/test_tool_manager.py
+++ b/tests/tools/test_tool_manager.py
@@ -280,8 +280,8 @@ class TestListTools:
tool_manager.add_tool_transformation(
"add", ToolTransformConfig(name="add_transformed")
)
- tools = await tool_manager.list_tools()
- tools_by_name = {tool.name: tool for tool in tools}
+ tools_dict = await tool_manager.get_tools()
+ tools_by_name = {tool.name: tool for tool in tools_dict.values()}
assert "add_transformed" in tools_by_name
assert "add" not in tools_by_name
@@ -303,8 +303,8 @@ class TestListTools:
name="add_transformed", description=None, tags={"enabled_tools"}
),
)
- tools = await tool_manager.list_tools()
- tools_by_name = {tool.name: tool for tool in tools}
+ tools_dict = await tool_manager.get_tools()
+ tools_by_name = {tool.name: tool for tool in tools_dict.values()}
assert "add_transformed" in tools_by_name
assert "add" not in tools_by_name
assert tools_by_name["add_transformed"].description is None
@@ -1027,12 +1027,12 @@ class TestMountedComponentsRaiseOnLoadError:
# Create a failing mounted server by corrupting it
parent_mcp.mount(child_mcp, prefix="child")
- # Corrupt the child server to make it fail during tool loading
- child_mcp._tool_manager._mounted_servers.append("invalid") # type: ignore
+ # Corrupt the parent's mounted servers to make it fail during loading
+ parent_mcp._mounted_servers.append("invalid") # type: ignore
- # Should not raise, just warn
- tools = await parent_mcp._tool_manager.list_tools()
- assert isinstance(tools, list) # Should return empty list, not raise
+ # Should not raise, just warn; use server middleware path now
+ tools = await parent_mcp._list_tools_middleware()
+ assert isinstance(tools, list) # Should return list, not raise
async def test_mounted_components_raise_on_load_error_true(self):
"""Test that when enabled, mounted component load errors are raised."""
@@ -1041,8 +1041,8 @@ class TestMountedComponentsRaiseOnLoadError:
# Create a failing mounted server
parent_mcp.mount(child_mcp, prefix="child")
- # Corrupt the child server to make it fail during tool loading
- child_mcp._tool_manager._mounted_servers.append("invalid") # type: ignore
+ # Corrupt the parent's mounted servers to make it fail during loading
+ parent_mcp._mounted_servers.append("invalid") # type: ignore
# Use temporary settings context manager
with temporary_settings(mounted_components_raise_on_load_error=True):
@@ -1050,4 +1050,4 @@ class TestMountedComponentsRaiseOnLoadError:
with pytest.raises(
AttributeError, match="'str' object has no attribute 'server'"
):
- await parent_mcp._tool_manager.list_tools()
+ await parent_mcp._list_tools_middleware()
diff --git a/tests/utilities/test_inspect.py b/tests/utilities/test_inspect.py
index 420335c82..730e297b6 100644
--- a/tests/utilities/test_inspect.py
+++ b/tests/utilities/test_inspect.py
@@ -269,6 +269,198 @@ class TestGetFastMCPInfo:
assert info.resources[0].uri == str(resources[0].uri)
assert info.prompts[0].name == prompts[0].name
+ async def test_inspect_respects_tag_filtering(self):
+ """Test that inspect omits components filtered out by include_tags/exclude_tags.
+
+ Regression test for Issue #2032: inspect command was showing components
+ that were filtered out by tag rules, causing confusion when those
+ components weren't actually available to clients.
+ """
+ # Create server with include_tags that will filter out untagged components
+ mcp = FastMCP(
+ "FilteredServer",
+ include_tags={"fetch", "analyze", "create"},
+ )
+
+ # Add tools with and without matching tags
+ @mcp.tool(tags={"fetch"})
+ def tagged_tool() -> str:
+ """Tool with matching tag - should be visible."""
+ return "visible"
+
+ @mcp.tool
+ def untagged_tool() -> str:
+ """Tool without tags - should be filtered out."""
+ return "hidden"
+
+ # Add resources with and without matching tags
+ @mcp.resource("resource://tagged", tags={"analyze"})
+ def tagged_resource() -> str:
+ """Resource with matching tag - should be visible."""
+ return "visible resource"
+
+ @mcp.resource("resource://untagged")
+ def untagged_resource() -> str:
+ """Resource without tags - should be filtered out."""
+ return "hidden resource"
+
+ # Add templates with and without matching tags
+ @mcp.resource("resource://tagged/{id}", tags={"create"})
+ def tagged_template(id: str) -> str:
+ """Template with matching tag - should be visible."""
+ return f"visible template {id}"
+
+ @mcp.resource("resource://untagged/{id}")
+ def untagged_template(id: str) -> str:
+ """Template without tags - should be filtered out."""
+ return f"hidden template {id}"
+
+ # Add prompts with and without matching tags
+ @mcp.prompt(tags={"fetch"})
+ def tagged_prompt() -> list:
+ """Prompt with matching tag - should be visible."""
+ return [{"role": "user", "content": "visible prompt"}]
+
+ @mcp.prompt
+ def untagged_prompt() -> list:
+ """Prompt without tags - should be filtered out."""
+ return [{"role": "user", "content": "hidden prompt"}]
+
+ # Get inspect info
+ info = await inspect_fastmcp(mcp)
+
+ # Verify only tagged components are visible
+ assert len(info.tools) == 1
+ assert info.tools[0].name == "tagged_tool"
+
+ assert len(info.resources) == 1
+ assert info.resources[0].uri == "resource://tagged"
+
+ assert len(info.templates) == 1
+ assert info.templates[0].uri_template == "resource://tagged/{id}"
+
+ assert len(info.prompts) == 1
+ assert info.prompts[0].name == "tagged_prompt"
+
+ # Verify this matches what a client would see
+ async with Client(mcp) as client:
+ tools = await client.list_tools()
+ resources = await client.list_resources()
+ templates = await client.list_resource_templates()
+ prompts = await client.list_prompts()
+
+ assert len(info.tools) == len(tools)
+ assert len(info.resources) == len(resources)
+ assert len(info.templates) == len(templates)
+ assert len(info.prompts) == len(prompts)
+
+ async def test_inspect_respects_tag_filtering_with_mounted_servers(self):
+ """Test that inspect applies tag filtering to mounted servers.
+
+ Verifies that when a parent server has tag filters, those filters
+ are respected when inspecting components from mounted servers.
+ """
+ # Create a mounted server with various tagged and untagged components
+ mounted = FastMCP("MountedServer")
+
+ @mounted.tool(tags={"allowed"})
+ def allowed_tool() -> str:
+ return "allowed"
+
+ @mounted.tool(tags={"blocked"})
+ def blocked_tool() -> str:
+ return "blocked"
+
+ @mounted.tool
+ def untagged_tool() -> str:
+ return "untagged"
+
+ @mounted.resource("resource://allowed", tags={"allowed"})
+ def allowed_resource() -> str:
+ return "allowed resource"
+
+ @mounted.resource("resource://blocked", tags={"blocked"})
+ def blocked_resource() -> str:
+ return "blocked resource"
+
+ @mounted.prompt(tags={"allowed"})
+ def allowed_prompt() -> list:
+ return [{"role": "user", "content": "allowed"}]
+
+ @mounted.prompt(tags={"blocked"})
+ def blocked_prompt() -> list:
+ return [{"role": "user", "content": "blocked"}]
+
+ # Create parent server with tag filtering
+ parent = FastMCP("ParentServer", include_tags={"allowed"})
+ parent.mount(mounted)
+
+ # Get inspect info
+ info = await inspect_fastmcp(parent)
+
+ # Only components with "allowed" tag should be visible
+ tool_names = [t.name for t in info.tools]
+ assert "allowed_tool" in tool_names
+ assert "blocked_tool" not in tool_names
+ assert "untagged_tool" not in tool_names
+
+ resource_uris = [r.uri for r in info.resources]
+ assert "resource://allowed" in resource_uris
+ assert "resource://blocked" not in resource_uris
+
+ prompt_names = [p.name for p in info.prompts]
+ assert "allowed_prompt" in prompt_names
+ assert "blocked_prompt" not in prompt_names
+
+ # Verify this matches what a client would see
+ async with Client(parent) as client:
+ tools = await client.list_tools()
+ resources = await client.list_resources()
+ prompts = await client.list_prompts()
+
+ assert len(info.tools) == len(tools)
+ assert len(info.resources) == len(resources)
+ assert len(info.prompts) == len(prompts)
+
+ async def test_inspect_parent_filters_override_mounted_server_filters(self):
+ """Test that parent server tag filters apply to mounted servers.
+
+ Even if a mounted server has no tag filters of its own,
+ the parent server's filters should still apply.
+ """
+ # Create mounted server with NO tag filters (allows everything)
+ mounted = FastMCP("MountedServer")
+
+ @mounted.tool(tags={"production"})
+ def production_tool() -> str:
+ return "production"
+
+ @mounted.tool(tags={"development"})
+ def development_tool() -> str:
+ return "development"
+
+ @mounted.tool
+ def untagged_tool() -> str:
+ return "untagged"
+
+ # Create parent with exclude_tags - should filter mounted components
+ parent = FastMCP("ParentServer", exclude_tags={"development"})
+ parent.mount(mounted)
+
+ # Get inspect info
+ info = await inspect_fastmcp(parent)
+
+ # Only production and untagged should be visible
+ tool_names = [t.name for t in info.tools]
+ assert "production_tool" in tool_names
+ assert "untagged_tool" in tool_names
+ assert "development_tool" not in tool_names
+
+ # Verify this matches what a client would see
+ async with Client(parent) as client:
+ tools = await client.list_tools()
+ assert len(info.tools) == len(tools)
+
class TestFastMCP1xCompatibility:
"""Tests for FastMCP 1.x compatibility."""