ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-23 05:54:19 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat: Add flexible parameter forwarding to OAuth proxy (#1771)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-09-06 15:50:29 -04:00 • committed by GitHub
commit 3dd7373680
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 283 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

64
docs/servers/auth/oauth-proxy.mdx
View file

@ -147,8 +147,72 @@ mcp = FastMCP(name="My Server", auth=auth)
<ParamField body="valid_scopes" type="list[str] | None">
List of all possible valid scopes for the OAuth provider. These are advertised to clients through the `/.well-known` endpoints. Defaults to `required_scopes` from your TokenVerifier if not specified.
</ParamField>
<ParamField body="extra_authorize_params" type="dict[str, str] | None">
Additional parameters to forward to the upstream authorization endpoint. Useful for provider-specific parameters that aren't part of the standard OAuth2 flow.
For example, Auth0 requires an `audience` parameter to issue JWT tokens:
```python
extra_authorize_params={"audience": "https://api.example.com"}
```
These parameters are added to every authorization request sent to the upstream provider.
</ParamField>
<ParamField body="extra_token_params" type="dict[str, str] | None">
Additional parameters to forward to the upstream token endpoint during code exchange and token refresh. Useful for provider-specific requirements during token operations.
For example, some providers require additional context during token exchange:
```python
extra_token_params={"audience": "https://api.example.com"}
```
These parameters are included in all token requests to the upstream provider.
</ParamField>
</Card>
### 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:

62
src/fastmcp/server/auth/oauth_proxy.py
View file

@ -250,6 +250,10 @@ class OAuthProxy(OAuthProvider):
forward_pkce: bool = True,
# Token endpoint authentication
token_endpoint_auth_method: str | None = None,
# Extra parameters to forward to authorization endpoint
extra_authorize_params: dict[str, str] | None = None,
# Extra parameters to forward to token endpoint
extra_token_params: dict[str, str] | None = None,
):
"""Initialize the OAuth proxy provider.
@ -278,6 +282,11 @@ class OAuthProxy(OAuthProvider):
token_endpoint_auth_method: Token endpoint authentication method for upstream server.
Common values: "client_secret_basic", "client_secret_post", "none".
If None, authlib will use its default (typically "client_secret_basic").
extra_authorize_params: Additional parameters to forward to the upstream authorization endpoint.
Useful for provider-specific parameters like Auth0's "audience".
Example: {"audience": "https://api.example.com"}
extra_token_params: Additional parameters to forward to the upstream token endpoint.
Useful for provider-specific parameters during token exchange.
"""
# Always enable DCR since we implement it locally for MCP clients
client_registration_options = ClientRegistrationOptions(
@ -319,6 +328,10 @@ class OAuthProxy(OAuthProvider):
# Token endpoint authentication
self._token_endpoint_auth_method = token_endpoint_auth_method
# Extra parameters for authorization and token endpoints
self._extra_authorize_params = extra_authorize_params or {}
self._extra_token_params = extra_token_params or {}
# Local state for DCR and token bookkeeping
self._clients: dict[str, OAuthClientInformationFull] = {}
self._access_tokens: dict[str, AccessToken] = {}
@ -485,6 +498,24 @@ class OAuthProxy(OAuthProvider):
txn_id,
)
# Forward resource parameter if provided (RFC 8707)
if params.resource:
query_params["resource"] = params.resource
logger.debug(
"Forwarding resource indicator '%s' to upstream for transaction %s",
params.resource,
txn_id,
)
# Add any extra authorization parameters configured for this proxy
if self._extra_authorize_params:
query_params.update(self._extra_authorize_params)
logger.debug(
"Adding extra authorization parameters for transaction %s: %s",
txn_id,
list(self._extra_authorize_params.keys()),
)
# Build the upstream authorization URL
separator = "&" if "?" in self._upstream_authorization_endpoint else "?"
upstream_url = f"{self._upstream_authorization_endpoint}{separator}{urlencode(query_params)}"
@ -870,26 +901,35 @@ class OAuthProxy(OAuthProvider):
f"Exchanging IdP code for tokens with redirect_uri: {idp_redirect_uri}"
)
# Build token exchange parameters
token_params = {
"url": self._upstream_token_endpoint,
"code": idp_code,
"redirect_uri": idp_redirect_uri,
}
# Include proxy's code_verifier if we forwarded PKCE
proxy_code_verifier = transaction.get("proxy_code_verifier")
if proxy_code_verifier:
token_params["code_verifier"] = proxy_code_verifier
logger.debug(
"Including proxy code_verifier in token exchange for transaction %s",
txn_id,
)
idp_tokens: dict[str, Any] = await oauth_client.fetch_token( # type: ignore[misc]
url=self._upstream_token_endpoint,
code=idp_code,
redirect_uri=idp_redirect_uri,
code_verifier=proxy_code_verifier,
)
else:
idp_tokens: dict[str, Any] = await oauth_client.fetch_token( # type: ignore[misc]
url=self._upstream_token_endpoint,
code=idp_code,
redirect_uri=idp_redirect_uri,
# Add any extra token parameters configured for this proxy
if self._extra_token_params:
token_params.update(self._extra_token_params)
logger.debug(
"Adding extra token parameters for transaction %s: %s",
txn_id,
list(self._extra_token_params.keys()),
)
idp_tokens: dict[str, Any] = await oauth_client.fetch_token(
**token_params
) # type: ignore[misc]
logger.debug(
f"Successfully exchanged IdP code for tokens (transaction: {txn_id}, PKCE: {bool(proxy_code_verifier)})"
)

168
tests/server/auth/test_oauth_proxy.py
View file

@ -793,3 +793,171 @@ class TestOAuthProxyE2E:
txn_id = query_params["state"][0]
transaction = proxy._oauth_transactions[txn_id]
assert "proxy_code_verifier" in transaction
class TestParameterForwarding:
"""Tests for forwarding custom parameters to upstream OAuth provider."""
@pytest.fixture
def proxy_with_extra_params(self, jwt_verifier):
"""Create OAuthProxy with extra parameters configured."""
return OAuthProxy(
upstream_authorization_endpoint="https://oauth.example.com/authorize",
upstream_token_endpoint="https://oauth.example.com/token",
upstream_client_id="upstream-client",
upstream_client_secret="upstream-secret",
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
extra_authorize_params={"audience": "https://api.example.com"},
extra_token_params={"audience": "https://api.example.com"},
)
@pytest.fixture
def proxy_without_extra_params(self, jwt_verifier):
"""Create OAuthProxy without extra parameters."""
return OAuthProxy(
upstream_authorization_endpoint="https://oauth.example.com/authorize",
upstream_token_endpoint="https://oauth.example.com/token",
upstream_client_id="upstream-client",
upstream_client_secret="upstream-secret",
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
)
async def test_resource_parameter_forwarding(self, proxy_without_extra_params):
"""Test that RFC 8707 resource parameter is forwarded from client request."""
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="client_challenge",
scopes=["read"],
resource="https://api.example.com/v1", # RFC 8707 resource indicator
)
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"
async def test_extra_authorize_params(self, proxy_with_extra_params):
"""Test that extra authorization parameters are included."""
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="client_challenge",
scopes=["read"],
)
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"
async def test_resource_and_extra_params_together(self, proxy_with_extra_params):
"""Test that both resource and extra params can be used together."""
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="client_challenge",
scopes=["read"],
resource="https://resource.example.com", # Client-specified resource
)
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"
async def test_no_extra_params_when_not_configured(
self, proxy_without_extra_params
):
"""Test that no extra params are added when not configured."""
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="client_challenge",
scopes=["read"],
# No resource parameter
)
redirect_url = await proxy_without_extra_params.authorize(client, params)
query_params = parse_qs(urlparse(redirect_url).query)
# No audience parameter should be present (not configured)
assert "audience" not in query_params
# No resource parameter should be present (not provided by client)
assert "resource" not in query_params
async def test_multiple_extra_params(self, jwt_verifier):
"""Test multiple extra parameters can be configured and forwarded."""
proxy = OAuthProxy(
upstream_authorization_endpoint="https://oauth.example.com/authorize",
upstream_token_endpoint="https://oauth.example.com/token",
upstream_client_id="upstream-client",
upstream_client_secret="upstream-secret",
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
extra_authorize_params={
"audience": "https://api.example.com",
"prompt": "consent",
"max_age": "3600",
},
)
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="client_challenge",
scopes=["read"],
)
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"