diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml
index 119d24e38..07f2d93df 100644
--- a/.github/workflows/run-tests.yml
+++ b/.github/workflows/run-tests.yml
@@ -76,3 +76,5 @@ jobs:
run: uv run pytest tests -m "integration"
env:
FASTMCP_GITHUB_TOKEN: ${{ secrets.FASTMCP_GITHUB_TOKEN }}
+ FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID: ${{ secrets.FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID }}
+ FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET: ${{ secrets.FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET }}
diff --git a/docs/docs.json b/docs/docs.json
index 22412156b..d65a65ad0 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -99,8 +99,9 @@
"icon": "shield-check",
"pages": [
"servers/auth/authentication",
- "servers/auth/remote-oauth",
"servers/auth/token-verification",
+ "servers/auth/remote-oauth",
+ "servers/auth/oauth-proxy",
"servers/auth/full-oauth-server"
]
},
@@ -122,10 +123,7 @@
{
"group": "Essentials",
"icon": "cube",
- "pages": [
- "clients/client",
- "clients/transports"
- ]
+ "pages": ["clients/client", "clients/transports"]
},
{
"group": "Core Operations",
@@ -151,10 +149,7 @@
{
"group": "Authentication",
"icon": "user-shield",
- "pages": [
- "clients/auth/oauth",
- "clients/auth/bearer"
- ]
+ "pages": ["clients/auth/oauth", "clients/auth/bearer"]
}
]
},
@@ -163,6 +158,8 @@
"pages": [
"integrations/anthropic",
"integrations/authkit",
+ "integrations/github",
+ "integrations/google",
"integrations/chatgpt",
"integrations/claude-code",
"integrations/claude-desktop",
@@ -200,17 +197,12 @@
},
{
"anchor": "What's New",
- "pages": [
- "updates",
- "changelog"
- ]
+ "pages": ["updates", "changelog"]
},
{
"anchor": "Community",
"icon": "users",
- "pages": [
- "community/showcase"
- ]
+ "pages": ["community/showcase"]
}
]
},
diff --git a/docs/integrations/github.mdx b/docs/integrations/github.mdx
new file mode 100644
index 000000000..fc73c5a25
--- /dev/null
+++ b/docs/integrations/github.mdx
@@ -0,0 +1,205 @@
+---
+title: GitHub OAuth 🤝 FastMCP
+sidebarTitle: GitHub OAuth
+description: Secure your FastMCP server with GitHub OAuth
+icon: github
+tag: NEW
+---
+
+import { VersionBadge } from "/snippets/version-badge.mdx"
+
+
+
+This guide shows you how to secure your FastMCP server using **GitHub OAuth**. Since GitHub doesn't support Dynamic Client Registration, this integration uses the [**OAuth Proxy**](/servers/auth/oauth-proxy) pattern to bridge GitHub's traditional OAuth with MCP's authentication requirements.
+
+## Configuration
+
+### Prerequisites
+
+Before you begin, you will need:
+1. A **[GitHub Account](https://github.com/)** with access to create OAuth Apps
+2. Your FastMCP server's URL (can be localhost for development, e.g., `http://localhost:8000`)
+
+### Step 1: Create a GitHub OAuth App
+
+Create an OAuth App in your GitHub settings to get the credentials needed for authentication:
+
+
+
+ Go to **Settings → Developer settings → OAuth Apps** in your GitHub account, or visit [github.com/settings/developers](https://github.com/settings/developers).
+
+ Click **"New OAuth App"** to create a new application.
+
+
+
+ Fill in the application details:
+
+ - **Application name**: Choose a name users will recognize (e.g., "My FastMCP Server")
+ - **Homepage URL**: Your application's homepage or documentation URL
+ - **Authorization callback URL**: Your server URL + `/oauth/callback` (e.g., `http://localhost:8000/oauth/callback`)
+
+
+ The callback URL must match exactly. The default path is `/oauth/callback`, but you can customize it using the `redirect_path` parameter. For local development, GitHub allows `http://localhost` URLs. For production, you must use HTTPS.
+
+
+
+ If you want to use a custom callback path (e.g., `/auth/github/callback`), make sure to set the same path in both your GitHub OAuth App settings and the `redirect_path` parameter when configuring the GitHubProvider.
+
+
+
+
+ After creating the app, you'll see:
+
+ - **Client ID**: A public identifier like `Ov23liAbcDefGhiJkLmN`
+ - **Client Secret**: Click "Generate a new client secret" and save the value securely
+
+
+ Store these credentials securely. Never commit them to version control. Use environment variables or a secrets manager in production.
+
+
+
+
+### Step 2: FastMCP Configuration
+
+Create your FastMCP server using the `GitHubProvider`, which handles GitHub's OAuth quirks automatically:
+
+```python server.py
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.github import GitHubProvider
+
+# The GitHubProvider handles GitHub's token format and validation
+auth_provider = GitHubProvider(
+ client_id="Ov23liAbcDefGhiJkLmN", # Your GitHub OAuth App Client ID
+ client_secret="github_pat_...", # Your GitHub OAuth App Client Secret
+ base_url="http://localhost:8000", # Must match your OAuth App configuration
+ # redirect_path="/oauth/callback" # Default value, customize if needed
+)
+
+mcp = FastMCP(name="GitHub Secured App", auth=auth_provider)
+
+# Add a protected tool to test authentication
+@mcp.tool
+async def get_user_info() -> dict:
+ """Returns information about the authenticated GitHub user."""
+ from fastmcp.server.dependencies import get_access_token
+
+ token = get_access_token()
+ # The GitHubProvider stores user data in token claims
+ return {
+ "github_user": token.claims.get("login"),
+ "name": token.claims.get("name"),
+ "email": token.claims.get("email")
+ }
+```
+
+## Testing
+
+### Running the Server
+
+Start your FastMCP server with HTTP transport to enable OAuth flows:
+
+```bash
+fastmcp run server.py --transport http --port 8000
+```
+
+Your server is now running and protected by GitHub OAuth authentication.
+
+### Testing with a Client
+
+Create a test client that authenticates with your GitHub-protected server:
+
+```python test_client.py
+from fastmcp import Client
+import asyncio
+
+async def main():
+ # The client will automatically handle GitHub OAuth
+ async with Client("http://localhost:8000/mcp/", auth="oauth") as client:
+ # First-time connection will open GitHub login in your browser
+ print("✓ Authenticated with GitHub!")
+
+ # Test the protected tool
+ result = await client.call_tool("get_user_info")
+ print(f"GitHub user: {result['github_user']}")
+
+if __name__ == "__main__":
+ asyncio.run(main())
+```
+
+When you run the client for the first time:
+1. Your browser will open to GitHub's authorization page
+2. After you authorize the app, you'll be redirected back
+3. The client receives the token and can make authenticated requests
+
+
+The client caches tokens locally, so you won't need to re-authenticate for subsequent runs unless the token expires or you explicitly clear the cache.
+
+
+## Environment Variables
+
+For production deployments, use environment variables instead of hardcoding credentials.
+
+
+To use the registered GitHub provider, you must set `FASTMCP_SERVER_AUTH=GITHUB`. Learn more about [registered providers](/servers/auth/authentication#registered-providers).
+
+
+### Provider Selection
+
+
+Set to `GITHUB` to use the registered GitHubProvider with default configuration.
+
+
+### GitHub-Specific Configuration
+
+
+
+Your GitHub OAuth App Client ID (e.g., `Ov23liAbcDefGhiJkLmN`)
+
+
+
+Your GitHub OAuth App Client Secret
+
+
+
+Public URL of your FastMCP server for OAuth callbacks
+
+
+
+Redirect path configured in your GitHub OAuth App
+
+
+
+Comma-separated list of required GitHub scopes (e.g., `user,repo`)
+
+
+
+HTTP request timeout for GitHub API calls
+
+
+
+Example `.env` file:
+```bash
+# Use the registered GitHub provider
+FASTMCP_SERVER_AUTH=GITHUB
+
+# GitHub OAuth credentials
+FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID=Ov23liAbcDefGhiJkLmN
+FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET=github_pat_...
+FASTMCP_SERVER_AUTH_GITHUB_BASE_URL=https://your-server.com
+FASTMCP_SERVER_AUTH_GITHUB_REQUIRED_SCOPES=user,repo
+```
+
+With environment variables set, your server code simplifies to:
+
+```python server.py
+from fastmcp import FastMCP
+
+# Authentication is automatically configured from environment
+mcp = FastMCP(name="GitHub Secured App")
+
+@mcp.tool
+async def list_repos() -> list[str]:
+ """List the authenticated user's repositories."""
+ # Your tool implementation here
+ pass
+```
diff --git a/docs/integrations/google.mdx b/docs/integrations/google.mdx
new file mode 100644
index 000000000..78b07c553
--- /dev/null
+++ b/docs/integrations/google.mdx
@@ -0,0 +1,215 @@
+---
+title: Google OAuth 🤝 FastMCP
+sidebarTitle: Google OAuth
+description: Secure your FastMCP server with Google OAuth
+icon: google
+tag: NEW
+---
+
+import { VersionBadge } from "/snippets/version-badge.mdx"
+
+
+
+This guide shows you how to secure your FastMCP server using **Google OAuth**. Since Google doesn't support Dynamic Client Registration, this integration uses the [**OAuth Proxy**](/servers/auth/oauth-proxy) pattern to bridge Google's traditional OAuth with MCP's authentication requirements.
+
+## Configuration
+
+### Prerequisites
+
+Before you begin, you will need:
+1. A **[Google Cloud Account](https://console.cloud.google.com/)** with access to create OAuth 2.0 Client IDs
+2. Your FastMCP server's URL (can be localhost for development, e.g., `http://localhost:8000`)
+
+### Step 1: Create a Google OAuth 2.0 Client ID
+
+Create an OAuth 2.0 Client ID in your Google Cloud Console to get the credentials needed for authentication:
+
+
+
+ Go to the [Google Cloud Console](https://console.cloud.google.com/apis/credentials) and select your project (or create a new one).
+
+ First, configure the OAuth consent screen by navigating to **APIs & Services → OAuth consent screen**. Choose "External" for testing or "Internal" for G Suite organizations.
+
+
+
+ Navigate to **APIs & Services → Credentials** and click **"+ CREATE CREDENTIALS"** → **"OAuth client ID"**.
+
+ Configure your OAuth client:
+
+ - **Application type**: Web application
+ - **Name**: Choose a descriptive name (e.g., "FastMCP Server")
+ - **Authorized JavaScript origins**: Add your server's base URL (e.g., `http://localhost:8000`)
+ - **Authorized redirect URIs**: Add your server URL + `/oauth/callback` (e.g., `http://localhost:8000/oauth/callback`)
+
+
+ The redirect URI must match exactly. The default path is `/oauth/callback`, but you can customize it using the `redirect_path` parameter. For local development, Google allows `http://localhost` URLs with various ports. For production, you must use HTTPS.
+
+
+
+ If you want to use a custom callback path (e.g., `/auth/google/callback`), make sure to set the same path in both your Google OAuth Client settings and the `redirect_path` parameter when configuring the GoogleProvider.
+
+
+
+
+ After creating the client, you'll receive:
+
+ - **Client ID**: A string ending in `.apps.googleusercontent.com`
+ - **Client Secret**: A string starting with `GOCSPX-`
+
+ Download the JSON credentials or copy these values securely.
+
+
+ Store these credentials securely. Never commit them to version control. Use environment variables or a secrets manager in production.
+
+
+
+
+### Step 2: FastMCP Configuration
+
+Create your FastMCP server using the `GoogleProvider`, which handles Google's OAuth flow automatically:
+
+```python server.py
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.google import GoogleProvider
+
+# The GoogleProvider handles Google's token format and validation
+auth_provider = GoogleProvider(
+ client_id="123456789.apps.googleusercontent.com", # Your Google OAuth Client ID
+ client_secret="GOCSPX-abc123...", # Your Google OAuth Client Secret
+ base_url="http://localhost:8000", # Must match your OAuth configuration
+ required_scopes=["openid", "email", "profile"], # Request user information
+ # redirect_path="/oauth/callback" # Default value, customize if needed
+)
+
+mcp = FastMCP(name="Google Secured App", auth=auth_provider)
+
+# Add a protected tool to test authentication
+@mcp.tool
+async def get_user_info() -> dict:
+ """Returns information about the authenticated Google user."""
+ from fastmcp.server.dependencies import get_access_token
+
+ token = get_access_token()
+ # The GoogleProvider stores user data in token claims
+ return {
+ "google_id": token.claims.get("sub"),
+ "email": token.claims.get("email"),
+ "name": token.claims.get("name"),
+ "picture": token.claims.get("picture"),
+ "locale": token.claims.get("locale")
+ }
+```
+
+## Testing
+
+### Running the Server
+
+Start your FastMCP server with HTTP transport to enable OAuth flows:
+
+```bash
+fastmcp run server.py --transport http --port 8000
+```
+
+Your server is now running and protected by Google OAuth authentication.
+
+### Testing with a Client
+
+Create a test client that authenticates with your Google-protected server:
+
+```python test_client.py
+from fastmcp import Client
+import asyncio
+
+async def main():
+ # The client will automatically handle Google OAuth
+ async with Client("http://localhost:8000/mcp/", auth="oauth") as client:
+ # First-time connection will open Google login in your browser
+ print("✓ Authenticated with Google!")
+
+ # Test the protected tool
+ result = await client.call_tool("get_user_info")
+ print(f"Google user: {result['email']}")
+ print(f"Name: {result['name']}")
+
+if __name__ == "__main__":
+ asyncio.run(main())
+```
+
+When you run the client for the first time:
+1. Your browser will open to Google's authorization page
+2. Sign in with your Google account and grant the requested permissions
+3. After authorization, you'll be redirected back
+4. The client receives the token and can make authenticated requests
+
+
+The client caches tokens locally, so you won't need to re-authenticate for subsequent runs unless the token expires or you explicitly clear the cache.
+
+
+## Environment Variables
+
+For production deployments, use environment variables instead of hardcoding credentials.
+
+
+To use the registered Google provider, you must set `FASTMCP_SERVER_AUTH=GOOGLE`. Learn more about [registered providers](/servers/auth/authentication#registered-providers).
+
+
+### Provider Selection
+
+
+Set to `GOOGLE` to use the registered GoogleProvider with default configuration.
+
+
+### Google-Specific Configuration
+
+
+
+Your Google OAuth 2.0 Client ID (e.g., `123456789.apps.googleusercontent.com`)
+
+
+
+Your Google OAuth 2.0 Client Secret (e.g., `GOCSPX-abc123...`)
+
+
+
+Public URL of your FastMCP server for OAuth callbacks
+
+
+
+Redirect path configured in your Google OAuth Client
+
+
+
+Comma-separated list of required Google scopes (e.g., `openid`)
+
+
+
+HTTP request timeout for Google API calls
+
+
+
+Example `.env` file:
+```bash
+# Use the registered Google provider
+FASTMCP_SERVER_AUTH=GOOGLE
+
+# Google OAuth credentials
+FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID=123456789.apps.googleusercontent.com
+FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET=GOCSPX-abc123...
+FASTMCP_SERVER_AUTH_GOOGLE_BASE_URL=https://your-server.com
+FASTMCP_SERVER_AUTH_GOOGLE_REQUIRED_SCOPES=openid,email,profile
+```
+
+With environment variables set, your server code simplifies to:
+
+```python server.py
+from fastmcp import FastMCP
+
+# Authentication is automatically configured from environment
+mcp = FastMCP(name="Google Secured App")
+
+@mcp.tool
+async def protected_tool(query: str) -> str:
+ """A tool that requires Google authentication to access."""
+ # Your tool implementation here
+ return f"Processing authenticated request: {query}"
+```
\ No newline at end of file
diff --git a/docs/servers/auth/authentication.mdx b/docs/servers/auth/authentication.mdx
index b00807eef..c8a5b6380 100644
--- a/docs/servers/auth/authentication.mdx
+++ b/docs/servers/auth/authentication.mdx
@@ -34,7 +34,7 @@ Traditional web authentication assumes a human user with a browser who can inter
These challenges mean that not all authentication approaches work well with MCP. The patterns that do work fall into three categories based on the level of authentication responsibility your server assumes.
-## Understanding Authentication Responsibility
+## Authentication Responsibility
Authentication responsibility exists on a spectrum. Your MCP server can validate tokens created elsewhere, coordinate with external identity providers, or handle the complete authentication lifecycle internally. Each approach involves different trade-offs between simplicity, security, and control.
@@ -66,9 +66,9 @@ Full OAuth implementation means building user interfaces for login and consent,
This pattern makes sense only when you need complete control over the authentication process, operate in air-gapped environments, or have specialized requirements that external providers cannot meet.
-## FastMCP Implementation
+## FastMCP Authentication Providers
-FastMCP translates these authentication responsibility levels into three concrete classes that handle the complexities of MCP protocol integration.
+FastMCP translates these authentication responsibility levels into a variety of concrete classes that handle the complexities of MCP protocol integration. You can build on these classes to handle the complexities of MCP protocol integration.
### TokenVerifier
@@ -97,11 +97,13 @@ This example configures token validation against a JWT issuer. The `JWTVerifier`
### RemoteAuthProvider
-`RemoteAuthProvider` combines token validation with OAuth discovery metadata, enabling MCP clients to automatically discover and authenticate with external identity providers.
+`RemoteAuthProvider` enables authentication with identity providers that **support Dynamic Client Registration (DCR)**, such as WorkOS AuthKit. With DCR, MCP clients can automatically register themselves with the identity provider and obtain credentials without any manual configuration.
-This class extends `TokenVerifier` functionality by adding OAuth 2.0 protected resource endpoints that advertise your authentication requirements. MCP clients can examine these endpoints to understand which identity providers you trust and how to obtain valid tokens.
+This class combines token validation with OAuth discovery metadata. It extends `TokenVerifier` functionality by adding OAuth 2.0 protected resource endpoints that advertise your authentication requirements. MCP clients examine these endpoints to understand which identity providers you trust and how to obtain valid tokens.
-The implementation handles the OAuth metadata generation required by the MCP specification while delegating actual token validation to an underlying `TokenVerifier`. This separation allows you to use different token validation strategies while maintaining consistent OAuth discovery behavior.
+The key requirement is that your identity provider must support DCR - the ability for clients to dynamically register and obtain credentials. This is what enables the seamless, automated authentication flow that MCP requires.
+
+For example, the built-in `AuthKitProvider` uses WorkOS AuthKit, which fully supports DCR:
```python
from fastmcp import FastMCP
@@ -117,10 +119,41 @@ mcp = FastMCP(name="Enterprise Server", auth=auth)
This example uses WorkOS AuthKit as the external identity provider. The `AuthKitProvider` automatically configures token validation against WorkOS and provides the OAuth metadata that MCP clients need for automatic authentication.
-`RemoteAuthProvider` excels for production applications that need professional identity management without implementation complexity.
+`RemoteAuthProvider` is ideal for production applications when your identity provider supports Dynamic Client Registration (DCR). This enables fully automated authentication without manual client configuration.
→ **Complete guide**: [Remote OAuth](/servers/auth/remote-oauth)
+### OAuthProxy
+
+
+
+`OAuthProxy` enables authentication with OAuth providers that **don't support Dynamic Client Registration (DCR)**, such as GitHub, Google, Azure, and most traditional enterprise identity systems.
+
+When identity providers require manual app registration and fixed credentials, `OAuthProxy` bridges the gap. It presents a DCR-compliant interface to MCP clients (accepting any registration request) while using your pre-registered credentials with the upstream provider. The proxy handles the complexity of callback forwarding, enabling dynamic client callbacks to work with providers that require fixed redirect URIs.
+
+This class solves the fundamental incompatibility between MCP's expectation of dynamic registration and traditional OAuth providers' requirement for manual app registration.
+
+For example, the built-in `GitHubProvider` extends `OAuthProxy` to work with GitHub's OAuth system:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.github import GitHubProvider
+
+auth = GitHubProvider(
+ client_id="Ov23li...", # Your GitHub OAuth App ID
+ client_secret="abc123...", # Your GitHub OAuth App Secret
+ base_url="https://your-server.com"
+)
+
+mcp = FastMCP(name="GitHub-Protected Server", auth=auth)
+```
+
+This example uses the GitHub provider, which extends `OAuthProxy` with GitHub-specific token validation. The proxy handles the complete OAuth flow while making GitHub's non-DCR authentication work seamlessly with MCP clients.
+
+`OAuthProxy` is essential when integrating with OAuth providers that don't support DCR. This includes most established providers like GitHub, Google, and Azure, which require manual app registration through their developer consoles.
+
+→ **Complete guide**: [OAuth Proxy](/servers/auth/oauth-proxy)
+
### OAuthProvider
`OAuthProvider` implements a complete OAuth 2.0 authorization server within your MCP server. This class handles the full authentication lifecycle from user credential verification to token management.
@@ -162,6 +195,37 @@ Environment-based configuration separates authentication settings from applicati
FastMCP automatically detects authentication configuration from environment variables when no explicit `auth` parameter is provided. The configuration system supports all authentication providers and their various options.
+#### Registered Providers
+
+FastMCP includes pre-configured providers for popular OAuth services that can be activated with a single environment variable:
+
+
+The authentication provider to use. Supported values:
+- `GITHUB` - GitHub OAuth (requires additional GitHub-specific env vars)
+- `GOOGLE` - Google OAuth (requires additional Google-specific env vars)
+- `JWT` - JWT token verification
+- `WORKOS` - WorkOS AuthKit
+- Custom provider class names
+
+
+When using registered providers like GitHub or Google, you'll need to set provider-specific environment variables:
+
+```bash
+# GitHub OAuth
+export FASTMCP_SERVER_AUTH=GITHUB
+export FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID="Ov23li..."
+export FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET="github_pat_..."
+
+# Google OAuth
+export FASTMCP_SERVER_AUTH=GOOGLE
+export FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID="123456.apps.googleusercontent.com"
+export FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET="GOCSPX-..."
+```
+
+#### Custom Provider Configuration
+
+For providers that aren't pre-registered, specify the provider class and its configuration:
+
```bash
export FASTMCP_SERVER_AUTH=JWT
export FASTMCP_SERVER_AUTH_JWT_JWKS_URI="https://auth.example.com/jwks"
@@ -184,7 +248,9 @@ This approach simplifies deployment pipelines and follows twelve-factor app prin
The authentication approach you choose depends on your existing infrastructure, security requirements, and operational constraints.
-**For most production applications, external identity providers offer the best balance of security, features, and simplicity.** This approach provides enterprise-grade authentication without implementation complexity and scales well as your application grows. The main trade-off is requiring users to sign up with your chosen identity provider, but this also brings benefits like professional user management, security monitoring, and compliance features.
+**For OAuth providers without DCR support (GitHub, Google, Azure, most enterprise systems), use OAuth Proxy.** These providers require manual app registration through their developer consoles. OAuth Proxy bridges the gap by presenting a DCR-compliant interface to MCP clients while using your fixed credentials with the provider. The proxy's callback forwarding pattern enables dynamic client ports to work with providers that require fixed redirect URIs.
+
+**For identity providers with DCR support (WorkOS AuthKit, modern auth platforms), use RemoteAuthProvider.** These providers allow clients to dynamically register and obtain credentials without manual configuration. This enables the fully automated authentication flow that MCP is designed for, providing the best user experience and simplest implementation.
**Token validation works well when you already have authentication infrastructure that issues structured tokens.** If your organization already uses JWT-based systems, API gateways, or enterprise SSO that can generate tokens, this approach integrates seamlessly while keeping your MCP server focused on its core functionality. The simplicity comes from leveraging existing investment in authentication infrastructure.
diff --git a/docs/servers/auth/oauth-proxy.mdx b/docs/servers/auth/oauth-proxy.mdx
new file mode 100644
index 000000000..d385f1119
--- /dev/null
+++ b/docs/servers/auth/oauth-proxy.mdx
@@ -0,0 +1,300 @@
+---
+title: OAuth Proxy
+sidebarTitle: OAuth Proxy
+description: Enable authentication with OAuth providers that don't support Dynamic Client Registration.
+icon: share
+tag: NEW
+---
+
+import { VersionBadge } from "/snippets/version-badge.mdx"
+
+
+
+OAuth Proxy enables your FastMCP server to authenticate with OAuth providers that **don't support Dynamic Client Registration (DCR)**. This includes virtually all traditional OAuth providers: GitHub, Google, Azure, Facebook, Discord, and most enterprise identity systems.
+
+While MCP clients expect to dynamically register and obtain credentials, these 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 to use OAuth Proxy vs RemoteAuthProvider:**
+- **OAuth Proxy**: For providers WITHOUT Dynamic Client Registration (GitHub, Google, Azure, etc.)
+- **RemoteAuthProvider**: For providers WITH Dynamic Client Registration (WorkOS AuthKit, etc.)
+
+OAuth Proxy makes traditional OAuth providers work seamlessly with MCP's automated authentication flow.
+
+
+## DCR vs Non-DCR Providers
+
+The key distinction in MCP authentication is whether your OAuth provider supports **Dynamic Client Registration (DCR)**:
+
+- **Providers WITH DCR** (WorkOS, some OIDC providers): Use [`RemoteAuthProvider`](/servers/auth/remote-oauth)
+ - Clients can register themselves automatically
+ - No manual app registration needed
+ - True dynamic authentication flow
+
+- **Providers WITHOUT DCR** (GitHub, Google, Azure, Discord, etc.): Use `OAuthProxy` (this guide)
+ - Requires manual app registration in provider's console
+ - You obtain fixed client ID and secret
+ - OAuth Proxy bridges the gap for MCP compatibility
+
+OAuth Proxy makes non-DCR providers work seamlessly with MCP by implementing a local DCR interface that always returns your pre-registered credentials.
+
+## Understanding the DCR Gap
+
+**Dynamic Client Registration (DCR)** allows OAuth clients to automatically register themselves with an authorization server and obtain credentials without manual intervention. The MCP specification is designed around this capability, expecting clients to register dynamically.
+
+However, most OAuth providers don't support DCR:
+
+| Provider Type | DCR Support | Registration Method | Examples |
+|--------------|-------------|--------------------|-----------|
+| Modern Auth Platforms | ✅ Yes | Automatic via API | WorkOS AuthKit, Some OIDC providers |
+| Traditional OAuth | ❌ No | Manual via console | GitHub, Google, Azure, Discord, Facebook |
+| Enterprise SSO | ❌ No | IT Administrator | Okta, AD FS, PingIdentity |
+
+Providers without DCR require you to:
+- Manually register applications through their developer console
+- Obtain fixed client IDs and secrets that never change
+- Pre-configure specific redirect URIs
+- Manage credentials through their web interface
+
+This creates a fundamental incompatibility: MCP clients expect to call a registration endpoint and receive credentials, but traditional providers only work with pre-registered apps. OAuth Proxy solves this by accepting any client registration request and returning your fixed upstream credentials.
+
+## How OAuth Proxy Works
+
+The OAuth Proxy implements an intelligent callback forwarding pattern that solves both the DCR problem and the redirect URI mismatch issue:
+
+```mermaid
+sequenceDiagram
+ participant Client as MCP Client
(localhost:random)
+ participant Proxy as FastMCP OAuth Proxy
(server:8000)
+ participant Provider as OAuth Provider
(GitHub, etc.)
+
+ Note over Client, Proxy: Dynamic Registration (Local)
+ Client->>Proxy: 1. POST /register
redirect_uri: localhost:54321/callback
+ Proxy-->>Client: 2. Returns fixed upstream credentials
+
+ Note over Client, Proxy: Authorization with Callback Forwarding
+ Client->>Proxy: 3. GET /authorize
redirect_uri=localhost:54321/callback
+ Note over Proxy: Store transaction with client callback
+ Proxy->>Provider: 4. Redirect to provider
redirect_uri=server:8000/oauth/callback
+
+ Note over Provider, Proxy: Provider Callback
+ Provider->>Proxy: 5. GET /oauth/callback
with authorization code
+ Proxy->>Provider: 6. Exchange code for tokens
+ Provider-->>Proxy: 7. Access & refresh tokens
+
+ Note over Proxy, Client: Client Callback Forwarding
+ Proxy->>Client: 8. Redirect to localhost:54321/callback
with new authorization code
+
+ Note over Client, Proxy: Token Exchange
+ Client->>Proxy: 9. POST /token with code
+ Proxy-->>Client: 10. Returns stored provider tokens
+```
+
+### The Callback Forwarding Pattern
+
+OAuth Proxy implements an innovative callback forwarding pattern that solves the redirect URI mismatch problem:
+
+**The Challenge:**
+- MCP clients listen on random localhost ports (e.g., `http://localhost:54321/callback`)
+- Each client session uses a different port
+- OAuth providers only accept pre-registered, fixed redirect URIs
+- Registering every possible localhost port is impossible
+
+**The Solution:**
+
+The proxy acts as an intermediary callback handler:
+
+1. **Dynamic Registration**: Client provides its localhost callback URL during registration
+2. **Transaction Tracking**: Proxy stores the client's callback URL with a transaction ID
+3. **Fixed Provider Callback**: Proxy uses its own fixed callback URL with the provider
+4. **Server-Side Token Exchange**: Proxy receives the provider's callback and exchanges the authorization code for tokens
+5. **Client Forwarding**: Proxy redirects to the client's original localhost callback with a new authorization code
+6. **Token Delivery**: Client exchanges this new code with the proxy to receive the provider's tokens
+
+This pattern maintains full OAuth 2.1 security (including PKCE) while enabling dynamic client ports to work with fixed provider callbacks. The client never knows it's talking to a proxy - it experiences a standard DCR flow.
+
+## Basic Implementation
+
+The `OAuthProxy` class provides the complete proxy implementation:
+
+
+
+ URL of your OAuth provider's authorization endpoint (e.g., `https://github.com/login/oauth/authorize`)
+
+
+
+ URL of your OAuth provider's token endpoint (e.g., `https://github.com/login/oauth/access_token`)
+
+
+
+ Client ID from your registered OAuth application
+
+
+
+ Client secret from your registered OAuth application
+
+
+
+ A [`TokenVerifier`](/servers/auth/token-verification) instance to validate the provider's tokens
+
+
+
+ Public URL of your FastMCP server (e.g., `https://your-server.com`)
+
+
+
+ Path for OAuth callbacks. Must match the redirect URI configured in your OAuth application
+
+
+
+ Optional URL of provider's token revocation endpoint
+
+
+
+ Issuer URL for OAuth metadata (defaults to base_url)
+
+
+
+ Optional URL to your service documentation
+
+
+
+ Resource server URL (defaults to base_url)
+
+
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.proxy import OAuthProxy
+from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+# Configure token validation for your provider
+token_verifier = JWTVerifier(
+ jwks_uri="https://your-provider.com/.well-known/jwks.json",
+ issuer="https://your-provider.com",
+ audience="your-app-id"
+)
+
+# Create the OAuth proxy (accepts strings for URLs)
+auth = OAuthProxy(
+ # Upstream provider endpoints
+ upstream_authorization_endpoint="https://your-provider.com/oauth/authorize",
+ upstream_token_endpoint="https://your-provider.com/oauth/token",
+
+ # Your registered app credentials
+ upstream_client_id="your-registered-client-id",
+ upstream_client_secret="your-registered-client-secret",
+
+ # Token validation
+ token_verifier=token_verifier,
+
+ # Your FastMCP server URL (string automatically converted to AnyHttpUrl)
+ base_url="https://your-server.com",
+
+ # Optional: customize callback path (defaults to "/oauth/callback")
+ redirect_path="/oauth/callback"
+)
+
+mcp = FastMCP(name="My Server", auth=auth)
+```
+
+### OAuth Provider Configuration
+
+When registering your application with your OAuth provider, configure the redirect/callback URL as:
+
+```
+https://your-server.com/oauth/callback
+```
+
+For local development with providers that support it (like GitHub):
+
+```
+http://localhost:8000/oauth/callback
+```
+
+The proxy automatically:
+- Implements DCR by returning your fixed credentials to any client that registers
+- Handles callback forwarding between dynamic client callbacks and your fixed provider callback
+- Exchanges authorization codes server-side for enhanced security
+- Validates tokens using your provider's public keys or API
+- Maintains PKCE security throughout the flow
+
+## Client Compatibility
+
+
+The OAuth Proxy's callback forwarding enables **any MCP client** to authenticate with **any OAuth provider**, regardless of redirect URI restrictions. Clients can use dynamic localhost ports while providers see their expected fixed callbacks.
+
+
+This breakthrough means that MCP clients no longer need to worry about registering specific callback URLs with OAuth providers. The proxy handles the complexity of bridging dynamic client callbacks with the fixed URLs that providers require. The entire flow maintains OAuth 2.1 and PKCE (RFC-7636) compliance for security.
+
+## Token Verification Strategies
+
+Different OAuth providers use different token formats, requiring appropriate verification strategies:
+
+### Provider Token Types
+
+| Provider | Token Type | Verification Method | Built-in Support |
+|----------|-----------|--------------------|-----------------|
+| GitHub | Opaque | API validation (`/user` endpoint) | ✅ `GitHubProvider` |
+| Google | JWT | JWKS signature verification | ✅ `GoogleProvider` |
+| Azure AD | JWT | JWKS signature verification | Configure `JWTVerifier` |
+| Discord | Opaque | API validation | ✅ `DiscordOAuthProxyProvider` |
+| Custom | Varies | Implement `TokenVerifier` | Extend base class |
+
+### Using Built-in Providers
+
+FastMCP includes pre-configured providers that handle token verification automatically:
+
+```python
+from fastmcp.server.auth.providers.github import GitHubProvider
+
+# GitHub provider with automatic API-based token validation
+auth = GitHubProvider(
+ client_id="your-github-client-id",
+ client_secret="your-github-client-secret",
+ base_url="https://your-server.com"
+)
+```
+
+### Custom Token Verification
+
+For providers without built-in support, implement a [`TokenVerifier`](/servers/auth/token-verification):
+
+- **JWT tokens**: Use `JWTVerifier` with the provider's JWKS endpoint
+- **Opaque tokens**: Extend `TokenVerifier` to validate via the provider's API
+- **Hybrid approaches**: Combine multiple verification methods as needed
+
+## Environment Configuration
+
+OAuth Proxy supports environment-based configuration for production deployments:
+
+```bash
+# Provider selection
+export FASTMCP_SERVER_AUTH=OAUTH_PROXY
+
+# OAuth endpoints
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_UPSTREAM_AUTHORIZATION_ENDPOINT="https://github.com/login/oauth/authorize"
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_UPSTREAM_TOKEN_ENDPOINT="https://github.com/login/oauth/access_token"
+
+# Credentials (use secrets management in production)
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_UPSTREAM_CLIENT_ID="Ov23li..."
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_UPSTREAM_CLIENT_SECRET="abc123..."
+
+# Token validation
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_TOKEN_VERIFIER="JWT"
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_JWKS_URI="https://provider.com/.well-known/jwks.json"
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_ISSUER="https://provider.com"
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_AUDIENCE="your-app-id"
+
+# Server URL
+export FASTMCP_SERVER_AUTH_OAUTH_PROXY_BASE_URL="https://your-server.com"
+```
+
+With environment variables configured, your code becomes:
+
+```python
+from fastmcp import FastMCP
+
+# Authentication automatically configured from environment
+mcp = FastMCP(name="My Server")
+```
\ No newline at end of file
diff --git a/docs/servers/auth/remote-oauth.mdx b/docs/servers/auth/remote-oauth.mdx
index 0e86b8974..b230346eb 100644
--- a/docs/servers/auth/remote-oauth.mdx
+++ b/docs/servers/auth/remote-oauth.mdx
@@ -10,12 +10,30 @@ import { VersionBadge } from "/snippets/version-badge.mdx"
-Remote OAuth integration allows your FastMCP server to leverage external identity providers while maintaining the automated authentication flows that MCP clients require. This approach provides enterprise-grade authentication features without the complexity of implementing them yourself, making it the recommended pattern for most production applications.
+Remote OAuth integration allows your FastMCP server to leverage external identity providers that **support Dynamic Client Registration (DCR)**. With DCR, MCP clients can automatically register themselves with the identity provider and obtain credentials without any manual configuration. This provides enterprise-grade authentication with fully automated flows, making it ideal for production applications with modern identity providers.
-Remote OAuth requires identity providers that support **Dynamic Client Registration (DCR)**. This enables MCP clients to automatically register and authenticate without manual configuration steps.
+**When to use RemoteAuthProvider vs OAuth Proxy:**
+- **RemoteAuthProvider**: For providers WITH Dynamic Client Registration (WorkOS AuthKit, modern OIDC providers)
+- **OAuth Proxy**: For providers WITHOUT Dynamic Client Registration (GitHub, Google, Azure, Discord, etc.)
+
+RemoteAuthProvider requires DCR support for fully automated client registration and authentication.
+## DCR-Enabled Providers
+
+RemoteAuthProvider works with identity providers that support **Dynamic Client Registration (DCR)** - a critical capability that enables automated authentication flows:
+
+| Feature | DCR Providers (RemoteAuth) | Non-DCR Providers (OAuth Proxy) |
+|---------|---------------------------|--------------------------------|
+| **Client Registration** | Automatic via API | Manual in provider console |
+| **Credentials** | Dynamic per client | Fixed app credentials |
+| **Configuration** | Zero client config | Pre-shared credentials |
+| **Examples** | WorkOS AuthKit, modern OIDC | GitHub, Google, Azure |
+| **FastMCP Class** | `RemoteAuthProvider` | [`OAuthProxy`](/servers/auth/oauth-proxy) |
+
+If your provider doesn't support DCR (most traditional OAuth providers), you'll need to use [`OAuth Proxy`](/servers/auth/oauth-proxy) instead, which bridges the gap between MCP's DCR expectations and fixed OAuth credentials.
+
## The Remote OAuth Challenge
Traditional OAuth flows assume human users with web browsers who can interact with login forms, consent screens, and redirects. MCP clients operate differently - they're often automated systems that need to authenticate programmatically without human intervention.
diff --git a/examples/auth/github_oauth/README.md b/examples/auth/github_oauth/README.md
new file mode 100644
index 000000000..f3838d0be
--- /dev/null
+++ b/examples/auth/github_oauth/README.md
@@ -0,0 +1,31 @@
+# GitHub OAuth Example
+
+Demonstrates FastMCP server protection with GitHub OAuth.
+
+## Setup
+
+1. Create a GitHub OAuth App:
+ - Go to GitHub Settings > Developer settings > OAuth Apps
+ - Set Authorization callback URL to: `http://localhost:8000/oauth/callback`
+ - Copy the Client ID and Client Secret
+
+2. Set environment variables:
+
+ ```bash
+ export FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID="your-client-id"
+ export FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET="your-client-secret"
+ ```
+
+3. Run the server:
+
+ ```bash
+ python server.py
+ ```
+
+4. In another terminal, run the client:
+
+ ```bash
+ python client.py
+ ```
+
+The client will open your browser for GitHub authentication.
diff --git a/examples/auth/github_oauth/client.py b/examples/auth/github_oauth/client.py
new file mode 100644
index 000000000..5f1f39bb2
--- /dev/null
+++ b/examples/auth/github_oauth/client.py
@@ -0,0 +1,32 @@
+"""OAuth client example for connecting to FastMCP servers.
+
+This example demonstrates how to connect to an OAuth-protected FastMCP server.
+
+To run:
+ python client.py
+"""
+
+import asyncio
+
+from fastmcp.client import Client
+
+SERVER_URL = "http://127.0.0.1:8000/mcp"
+
+
+async def main():
+ try:
+ async with Client(SERVER_URL, auth="oauth") as client:
+ assert await client.ping()
+ print("✅ Successfully authenticated!")
+
+ tools = await client.list_tools()
+ print(f"🔧 Available tools ({len(tools)}):")
+ for tool in tools:
+ print(f" - {tool.name}: {tool.description}")
+ except Exception as e:
+ print(f"❌ Authentication failed: {e}")
+ raise
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/auth/github_oauth/server.py b/examples/auth/github_oauth/server.py
new file mode 100644
index 000000000..1dd8051d2
--- /dev/null
+++ b/examples/auth/github_oauth/server.py
@@ -0,0 +1,35 @@
+"""GitHub OAuth server example for FastMCP.
+
+This example demonstrates how to protect a FastMCP server with GitHub OAuth.
+
+Required environment variables:
+- FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID: Your GitHub OAuth app client ID
+- FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET: Your GitHub OAuth app client secret
+
+To run:
+ python server.py
+"""
+
+import os
+
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.github import GitHubProvider
+
+auth = GitHubProvider(
+ client_id=os.getenv("FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID") or "",
+ client_secret=os.getenv("FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET") or "",
+ base_url="http://localhost:8000",
+ # redirect_path="/oauth/callback", # Default path - change if using a different callback URL
+)
+
+mcp = FastMCP("GitHub OAuth Example Server", auth=auth)
+
+
+@mcp.tool
+def echo(message: str) -> str:
+ """Echo the provided message."""
+ return message
+
+
+if __name__ == "__main__":
+ mcp.run(transport="http", port=8000)
diff --git a/examples/auth/google_oauth/README.md b/examples/auth/google_oauth/README.md
new file mode 100644
index 000000000..e23728435
--- /dev/null
+++ b/examples/auth/google_oauth/README.md
@@ -0,0 +1,34 @@
+# Google OAuth Example
+
+Demonstrates FastMCP server protection with Google OAuth.
+
+## Setup
+
+1. Create a Google OAuth 2.0 Client:
+ - Go to [Google Cloud Console](https://console.cloud.google.com/)
+ - Create or select a project
+ - Go to APIs & Services > Credentials
+ - Create OAuth 2.0 Client ID (Web application)
+ - Add Authorized redirect URI: `http://localhost:8000/oauth/callback`
+ - Copy the Client ID and Client Secret
+
+2. Set environment variables:
+
+ ```bash
+ export FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID="your-client-id.apps.googleusercontent.com"
+ export FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET="your-client-secret"
+ ```
+
+3. Run the server:
+
+ ```bash
+ python server.py
+ ```
+
+4. In another terminal, run the client:
+
+ ```bash
+ python client.py
+ ```
+
+The client will open your browser for Google authentication.
diff --git a/examples/auth/google_oauth/client.py b/examples/auth/google_oauth/client.py
new file mode 100644
index 000000000..5f1f39bb2
--- /dev/null
+++ b/examples/auth/google_oauth/client.py
@@ -0,0 +1,32 @@
+"""OAuth client example for connecting to FastMCP servers.
+
+This example demonstrates how to connect to an OAuth-protected FastMCP server.
+
+To run:
+ python client.py
+"""
+
+import asyncio
+
+from fastmcp.client import Client
+
+SERVER_URL = "http://127.0.0.1:8000/mcp"
+
+
+async def main():
+ try:
+ async with Client(SERVER_URL, auth="oauth") as client:
+ assert await client.ping()
+ print("✅ Successfully authenticated!")
+
+ tools = await client.list_tools()
+ print(f"🔧 Available tools ({len(tools)}):")
+ for tool in tools:
+ print(f" - {tool.name}: {tool.description}")
+ except Exception as e:
+ print(f"❌ Authentication failed: {e}")
+ raise
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/auth/google_oauth/server.py b/examples/auth/google_oauth/server.py
new file mode 100644
index 000000000..65f96f1f0
--- /dev/null
+++ b/examples/auth/google_oauth/server.py
@@ -0,0 +1,37 @@
+"""Google OAuth server example for FastMCP.
+
+This example demonstrates how to protect a FastMCP server with Google OAuth.
+
+Required environment variables:
+- FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID: Your Google OAuth client ID
+- FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET: Your Google OAuth client secret
+
+To run:
+ python server.py
+"""
+
+import os
+
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.google import GoogleProvider
+
+auth = GoogleProvider(
+ client_id=os.getenv("FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID") or "",
+ client_secret=os.getenv("FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET") or "",
+ base_url="http://localhost:8000",
+ # redirect_path="/oauth/callback", # Default path - change if using a different callback URL
+ # Optional: specify required scopes
+ # required_scopes=["openid", "https://www.googleapis.com/auth/userinfo.email"],
+)
+
+mcp = FastMCP("Google OAuth Example Server", auth=auth)
+
+
+@mcp.tool
+def echo(message: str) -> str:
+ """Echo the provided message."""
+ return message
+
+
+if __name__ == "__main__":
+ mcp.run(transport="http", port=8000)
diff --git a/src/fastmcp/server/auth/providers/github.py b/src/fastmcp/server/auth/providers/github.py
new file mode 100644
index 000000000..e2a27ecae
--- /dev/null
+++ b/src/fastmcp/server/auth/providers/github.py
@@ -0,0 +1,267 @@
+"""GitHub OAuth provider for FastMCP.
+
+This module provides a complete GitHub OAuth integration that's ready to use
+with just a client ID and client secret. It handles all the complexity of
+GitHub's OAuth flow, token validation, and user management.
+
+Example:
+ ```python
+ from fastmcp import FastMCP
+ from fastmcp.server.auth.providers.github import GitHubProvider
+
+ # Simple GitHub OAuth protection
+ auth = GitHubProvider(
+ client_id="your-github-client-id",
+ client_secret="your-github-client-secret"
+ )
+
+ mcp = FastMCP("My Protected Server", auth=auth)
+ ```
+"""
+
+from __future__ import annotations
+
+import httpx
+from pydantic import AnyHttpUrl, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from fastmcp.server.auth import TokenVerifier
+from fastmcp.server.auth.auth import AccessToken
+from fastmcp.server.auth.proxy import OAuthProxy
+from fastmcp.server.auth.registry import register_provider
+from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.types import NotSet, NotSetT
+
+logger = get_logger(__name__)
+
+
+class GitHubProviderSettings(BaseSettings):
+ """Settings for GitHub OAuth provider."""
+
+ model_config = SettingsConfigDict(
+ env_prefix="FASTMCP_SERVER_AUTH_GITHUB_",
+ env_file=".env",
+ extra="ignore",
+ )
+
+ client_id: str | None = None
+ client_secret: SecretStr | None = None
+ base_url: AnyHttpUrl | str | None = None
+ redirect_path: str | None = None
+ required_scopes: list[str] | None = None
+ timeout_seconds: int | None = None
+
+
+class GitHubTokenVerifier(TokenVerifier):
+ """Token verifier for GitHub OAuth tokens.
+
+ GitHub OAuth tokens are opaque (not JWTs), so we verify them
+ by calling GitHub's API to check if they're valid and get user info.
+ """
+
+ def __init__(
+ self,
+ *,
+ required_scopes: list[str] | None = None,
+ timeout_seconds: int = 10,
+ ):
+ """Initialize the GitHub token verifier.
+
+ Args:
+ required_scopes: Required OAuth scopes (e.g., ['user:email'])
+ timeout_seconds: HTTP request timeout
+ """
+ super().__init__(required_scopes=required_scopes)
+ self.timeout_seconds = timeout_seconds
+
+ async def verify_token(self, token: str) -> AccessToken | None:
+ """Verify GitHub OAuth token by calling GitHub API."""
+ try:
+ async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+ # Get token info from GitHub API
+ response = await client.get(
+ "https://api.github.com/user",
+ headers={
+ "Authorization": f"Bearer {token}",
+ "Accept": "application/vnd.github.v3+json",
+ "User-Agent": "FastMCP-GitHub-OAuth",
+ },
+ )
+
+ if response.status_code != 200:
+ logger.debug(
+ "GitHub token verification failed: %d - %s",
+ response.status_code,
+ response.text[:200],
+ )
+ return None
+
+ user_data = response.json()
+
+ # Get token scopes from GitHub API
+ # GitHub includes scopes in the X-OAuth-Scopes header
+ scopes_response = await client.get(
+ "https://api.github.com/user/repos", # Any authenticated endpoint
+ headers={
+ "Authorization": f"Bearer {token}",
+ "Accept": "application/vnd.github.v3+json",
+ "User-Agent": "FastMCP-GitHub-OAuth",
+ },
+ )
+
+ # Extract scopes from X-OAuth-Scopes header if available
+ oauth_scopes_header = scopes_response.headers.get("x-oauth-scopes", "")
+ token_scopes = [
+ scope.strip()
+ for scope in oauth_scopes_header.split(",")
+ if scope.strip()
+ ]
+
+ # If no scopes in header, assume basic scopes based on successful user API call
+ if not token_scopes:
+ token_scopes = ["user"] # Basic scope if we can access user info
+
+ # Check required scopes
+ if self.required_scopes:
+ token_scopes_set = set(token_scopes)
+ required_scopes_set = set(self.required_scopes)
+ if not required_scopes_set.issubset(token_scopes_set):
+ logger.debug(
+ "GitHub token missing required scopes. Has %d, needs %d",
+ len(token_scopes_set),
+ len(required_scopes_set),
+ )
+ return None
+
+ # Create AccessToken with GitHub user info
+ return AccessToken(
+ token=token,
+ client_id=str(user_data.get("id", "unknown")), # Use GitHub user ID
+ scopes=token_scopes,
+ expires_at=None, # GitHub tokens don't typically expire
+ claims={
+ "sub": str(user_data["id"]),
+ "login": user_data.get("login"),
+ "name": user_data.get("name"),
+ "email": user_data.get("email"),
+ "avatar_url": user_data.get("avatar_url"),
+ "github_user_data": user_data,
+ },
+ )
+
+ except httpx.RequestError as e:
+ logger.debug("Failed to verify GitHub token: %s", e)
+ return None
+ except Exception as e:
+ logger.debug("GitHub token verification error: %s", e)
+ return None
+
+
+@register_provider("GitHub")
+class GitHubProvider(OAuthProxy):
+ """Complete GitHub OAuth provider for FastMCP.
+
+ This provider makes it trivial to add GitHub OAuth protection to any
+ FastMCP server. Just provide your GitHub OAuth app credentials and
+ a base URL, and you're ready to go.
+
+ Features:
+ - Transparent OAuth proxy to GitHub
+ - Automatic token validation via GitHub API
+ - User information extraction
+ - Minimal configuration required
+
+ Example:
+ ```python
+ from fastmcp import FastMCP
+ from fastmcp.server.auth.providers.github import GitHubProvider
+
+ auth = GitHubProvider(
+ client_id="Ov23li...",
+ client_secret="abc123...",
+ base_url="https://my-server.com" # Optional, defaults to http://localhost:8000
+ )
+
+ mcp = FastMCP("My App", auth=auth)
+ ```
+ """
+
+ def __init__(
+ self,
+ *,
+ client_id: str | NotSetT = NotSet,
+ client_secret: str | NotSetT = NotSet,
+ base_url: AnyHttpUrl | str | NotSetT = NotSet,
+ redirect_path: str | NotSetT = NotSet,
+ required_scopes: list[str] | None | NotSetT = NotSet,
+ timeout_seconds: int | NotSetT = NotSet,
+ ):
+ """Initialize GitHub OAuth provider.
+
+ Args:
+ client_id: GitHub OAuth app client ID (e.g., "Ov23li...")
+ client_secret: GitHub OAuth app client secret
+ base_url: Public URL of your FastMCP server (for OAuth callbacks)
+ redirect_path: Redirect path configured in GitHub OAuth app (defaults to "/oauth/callback")
+ required_scopes: Required GitHub scopes (defaults to ["user"])
+ timeout_seconds: HTTP request timeout for GitHub API calls
+ """
+ settings = GitHubProviderSettings.model_validate(
+ {
+ k: v
+ for k, v in {
+ "client_id": client_id,
+ "client_secret": client_secret,
+ "base_url": base_url,
+ "redirect_path": redirect_path,
+ "required_scopes": required_scopes,
+ "timeout_seconds": timeout_seconds,
+ }.items()
+ if v is not NotSet
+ }
+ )
+
+ # Validate required settings
+ if not settings.client_id:
+ raise ValueError(
+ "client_id is required - set via parameter or FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID"
+ )
+ if not settings.client_secret:
+ raise ValueError(
+ "client_secret is required - set via parameter or FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET"
+ )
+
+ # Apply defaults
+ base_url_final = settings.base_url or "http://localhost:8000"
+ redirect_path_final = settings.redirect_path or "/oauth/callback"
+ timeout_seconds_final = settings.timeout_seconds or 10
+ required_scopes_final = settings.required_scopes or ["user"]
+
+ # Create GitHub token verifier
+ token_verifier = GitHubTokenVerifier(
+ required_scopes=required_scopes_final,
+ timeout_seconds=timeout_seconds_final,
+ )
+
+ # Extract secret string from SecretStr
+ client_secret_str = (
+ settings.client_secret.get_secret_value() if settings.client_secret else ""
+ )
+
+ # Initialize OAuth proxy with GitHub endpoints
+ super().__init__(
+ upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+ upstream_token_endpoint="https://github.com/login/oauth/access_token",
+ upstream_client_id=settings.client_id,
+ upstream_client_secret=client_secret_str,
+ token_verifier=token_verifier,
+ base_url=base_url_final,
+ redirect_path=redirect_path_final,
+ issuer_url=base_url_final, # We act as the issuer for client registration
+ )
+
+ logger.info(
+ "Initialized GitHub OAuth provider for client %s with scopes: %s",
+ settings.client_id,
+ required_scopes_final,
+ )
diff --git a/src/fastmcp/server/auth/providers/google.py b/src/fastmcp/server/auth/providers/google.py
new file mode 100644
index 000000000..1c3c46e47
--- /dev/null
+++ b/src/fastmcp/server/auth/providers/google.py
@@ -0,0 +1,286 @@
+"""Google OAuth provider for FastMCP.
+
+This module provides a complete Google OAuth integration that's ready to use
+with just a client ID and client secret. It handles all the complexity of
+Google's OAuth flow, token validation, and user management.
+
+Example:
+ ```python
+ from fastmcp import FastMCP
+ from fastmcp.server.auth.providers.google import GoogleProvider
+
+ # Simple Google OAuth protection
+ auth = GoogleProvider(
+ client_id="your-google-client-id.apps.googleusercontent.com",
+ client_secret="your-google-client-secret"
+ )
+
+ mcp = FastMCP("My Protected Server", auth=auth)
+ ```
+"""
+
+from __future__ import annotations
+
+import time
+
+import httpx
+from pydantic import AnyHttpUrl, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from fastmcp.server.auth import TokenVerifier
+from fastmcp.server.auth.auth import AccessToken
+from fastmcp.server.auth.proxy import OAuthProxy
+from fastmcp.server.auth.registry import register_provider
+from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.types import NotSet, NotSetT
+
+logger = get_logger(__name__)
+
+
+class GoogleProviderSettings(BaseSettings):
+ """Settings for Google OAuth provider."""
+
+ model_config = SettingsConfigDict(
+ env_prefix="FASTMCP_SERVER_AUTH_GOOGLE_",
+ env_file=".env",
+ extra="ignore",
+ )
+
+ client_id: str | None = None
+ client_secret: SecretStr | None = None
+ base_url: AnyHttpUrl | str | None = None
+ redirect_path: str | None = None
+ required_scopes: list[str] | None = None
+ timeout_seconds: int | None = None
+
+
+class GoogleTokenVerifier(TokenVerifier):
+ """Token verifier for Google OAuth tokens.
+
+ Google OAuth tokens are opaque (not JWTs), so we verify them
+ by calling Google's tokeninfo API to check if they're valid and get user info.
+ """
+
+ def __init__(
+ self,
+ *,
+ required_scopes: list[str] | None = None,
+ timeout_seconds: int = 10,
+ ):
+ """Initialize the Google token verifier.
+
+ Args:
+ required_scopes: Required OAuth scopes (e.g., ['openid', 'email'])
+ timeout_seconds: HTTP request timeout
+ """
+ super().__init__(required_scopes=required_scopes)
+ self.timeout_seconds = timeout_seconds
+
+ async def verify_token(self, token: str) -> AccessToken | None:
+ """Verify Google OAuth token by calling Google's tokeninfo API."""
+ try:
+ async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+ # Use Google's tokeninfo endpoint to validate the token
+ response = await client.get(
+ "https://www.googleapis.com/oauth2/v1/tokeninfo",
+ params={"access_token": token},
+ headers={"User-Agent": "FastMCP-Google-OAuth"},
+ )
+
+ if response.status_code != 200:
+ logger.debug(
+ "Google token verification failed: %d",
+ response.status_code,
+ )
+ return None
+
+ token_info = response.json()
+
+ # Check if token is expired
+ expires_in = token_info.get("expires_in")
+ if expires_in and int(expires_in) <= 0:
+ logger.debug("Google token has expired")
+ return None
+
+ # Extract scopes from token info
+ scope_string = token_info.get("scope", "")
+ token_scopes = [
+ scope.strip() for scope in scope_string.split(" ") if scope.strip()
+ ]
+
+ # Check required scopes
+ if self.required_scopes:
+ token_scopes_set = set(token_scopes)
+ required_scopes_set = set(self.required_scopes)
+ if not required_scopes_set.issubset(token_scopes_set):
+ logger.debug(
+ "Google token missing required scopes. Has %d, needs %d",
+ len(token_scopes_set),
+ len(required_scopes_set),
+ )
+ return None
+
+ # Get additional user info if we have the right scopes
+ user_data = {}
+ if "openid" in token_scopes or "profile" in token_scopes:
+ try:
+ userinfo_response = await client.get(
+ "https://www.googleapis.com/oauth2/v2/userinfo",
+ headers={
+ "Authorization": f"Bearer {token}",
+ "User-Agent": "FastMCP-Google-OAuth",
+ },
+ )
+ if userinfo_response.status_code == 200:
+ user_data = userinfo_response.json()
+ except Exception as e:
+ logger.debug("Failed to fetch Google user info: %s", e)
+
+ # Calculate expiration time
+ expires_at = None
+ if expires_in:
+ expires_at = int(time.time() + int(expires_in))
+
+ # Create AccessToken with Google user info
+ access_token = AccessToken(
+ token=token,
+ client_id=token_info.get(
+ "audience", "unknown"
+ ), # Use audience as client_id
+ scopes=token_scopes,
+ expires_at=expires_at,
+ claims={
+ "sub": user_data.get("id")
+ or token_info.get("user_id", "unknown"),
+ "email": user_data.get("email"),
+ "name": user_data.get("name"),
+ "picture": user_data.get("picture"),
+ "given_name": user_data.get("given_name"),
+ "family_name": user_data.get("family_name"),
+ "locale": user_data.get("locale"),
+ "google_user_data": user_data,
+ "google_token_info": token_info,
+ },
+ )
+ logger.debug("Google token verified successfully")
+ return access_token
+
+ except httpx.RequestError as e:
+ logger.debug("Failed to verify Google token: %s", e)
+ return None
+ except Exception as e:
+ logger.debug("Google token verification error: %s", e)
+ return None
+
+
+@register_provider("Google")
+class GoogleProvider(OAuthProxy):
+ """Complete Google OAuth provider for FastMCP.
+
+ This provider makes it trivial to add Google OAuth protection to any
+ FastMCP server. Just provide your Google OAuth app credentials and
+ a base URL, and you're ready to go.
+
+ Features:
+ - Transparent OAuth proxy to Google
+ - Automatic token validation via Google's tokeninfo API
+ - User information extraction from Google APIs
+ - Minimal configuration required
+
+ Example:
+ ```python
+ from fastmcp import FastMCP
+ from fastmcp.server.auth.providers.google import GoogleProvider
+
+ auth = GoogleProvider(
+ client_id="123456789.apps.googleusercontent.com",
+ client_secret="GOCSPX-abc123...",
+ base_url="https://my-server.com" # Optional, defaults to http://localhost:8000
+ )
+
+ mcp = FastMCP("My App", auth=auth)
+ ```
+ """
+
+ def __init__(
+ self,
+ *,
+ client_id: str | NotSetT = NotSet,
+ client_secret: str | NotSetT = NotSet,
+ base_url: AnyHttpUrl | str | NotSetT = NotSet,
+ redirect_path: str | NotSetT = NotSet,
+ required_scopes: list[str] | None | NotSetT = NotSet,
+ timeout_seconds: int | NotSetT = NotSet,
+ ):
+ """Initialize Google OAuth provider.
+
+ Args:
+ client_id: Google OAuth client ID (e.g., "123456789.apps.googleusercontent.com")
+ client_secret: Google OAuth client secret (e.g., "GOCSPX-abc123...")
+ base_url: Public URL of your FastMCP server (for OAuth callbacks)
+ redirect_path: Redirect path configured in Google OAuth app (defaults to "/oauth/callback")
+ required_scopes: Required Google scopes (defaults to []). Common scopes include:
+ - "openid" for OpenID Connect
+ - "https://www.googleapis.com/auth/userinfo.email" for email access
+ - "https://www.googleapis.com/auth/userinfo.profile" for profile info
+ timeout_seconds: HTTP request timeout for Google API calls
+ """
+ settings = GoogleProviderSettings.model_validate(
+ {
+ k: v
+ for k, v in {
+ "client_id": client_id,
+ "client_secret": client_secret,
+ "base_url": base_url,
+ "redirect_path": redirect_path,
+ "required_scopes": required_scopes,
+ "timeout_seconds": timeout_seconds,
+ }.items()
+ if v is not NotSet
+ }
+ )
+
+ # Validate required settings
+ if not settings.client_id:
+ raise ValueError(
+ "client_id is required - set via parameter or FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID"
+ )
+ if not settings.client_secret:
+ raise ValueError(
+ "client_secret is required - set via parameter or FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET"
+ )
+
+ # Apply defaults
+ base_url_final = settings.base_url or "http://localhost:8000"
+ redirect_path_final = settings.redirect_path or "/oauth/callback"
+ timeout_seconds_final = settings.timeout_seconds or 10
+ required_scopes_final = settings.required_scopes or []
+
+ # Create Google token verifier
+ token_verifier = GoogleTokenVerifier(
+ required_scopes=required_scopes_final,
+ timeout_seconds=timeout_seconds_final,
+ )
+
+ # Extract secret string from SecretStr
+ client_secret_str = (
+ settings.client_secret.get_secret_value() if settings.client_secret else ""
+ )
+
+ # Initialize OAuth proxy with Google endpoints
+ super().__init__(
+ upstream_authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth",
+ upstream_token_endpoint="https://oauth2.googleapis.com/token",
+ upstream_client_id=settings.client_id,
+ upstream_client_secret=client_secret_str,
+ token_verifier=token_verifier,
+ base_url=base_url_final,
+ redirect_path=redirect_path_final,
+ issuer_url=base_url_final, # We act as the issuer for client registration
+ )
+
+ logger.info(
+ "Initialized Google OAuth provider for client %s with scopes: %s",
+ settings.client_id,
+ required_scopes_final,
+ )
diff --git a/src/fastmcp/server/auth/proxy.py b/src/fastmcp/server/auth/proxy.py
new file mode 100644
index 000000000..2c4a7de32
--- /dev/null
+++ b/src/fastmcp/server/auth/proxy.py
@@ -0,0 +1,1059 @@
+"""OAuth Proxy Provider for FastMCP.
+
+This provider acts as a transparent proxy to an upstream OAuth Authorization Server,
+handling Dynamic Client Registration locally while forwarding all other OAuth flows.
+This enables authentication with upstream providers that don't support DCR or have
+restricted client registration policies.
+
+Key features:
+- Proxies authorization and token endpoints to upstream server
+- Implements local Dynamic Client Registration with fixed upstream credentials
+- Validates tokens using upstream JWKS
+- Maintains minimal local state for bookkeeping
+- Enhanced logging with request correlation
+
+This implementation is based on the OAuth 2.1 specification and is designed for
+production use with enterprise identity providers.
+"""
+
+from __future__ import annotations
+
+import secrets
+import time
+from typing import TYPE_CHECKING, Any, Final
+from urllib.parse import urlencode
+
+import httpx
+from authlib.integrations.httpx_client import AsyncOAuth2Client
+from mcp.server.auth.provider import (
+ AccessToken,
+ AuthorizationCode,
+ AuthorizationParams,
+ RefreshToken,
+ TokenError,
+)
+from mcp.server.auth.settings import (
+ ClientRegistrationOptions,
+ RevocationOptions,
+)
+from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
+from pydantic import AnyHttpUrl, AnyUrl, SecretStr
+from starlette.requests import Request
+from starlette.responses import JSONResponse, RedirectResponse
+from starlette.routing import Route
+
+from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
+from fastmcp.utilities.logging import get_logger
+
+if TYPE_CHECKING:
+ pass
+
+logger = get_logger(__name__)
+
+
+class ProxyDCRClient(OAuthClientInformationFull):
+ """Client for DCR proxy that accepts any localhost redirect URI.
+
+ This special client class is critical for the OAuth proxy to work correctly
+ with Dynamic Client Registration (DCR). Here's why it exists:
+
+ Problem:
+ --------
+ When MCP clients use OAuth, they dynamically register with random localhost
+ ports (e.g., http://localhost:55454/callback). The OAuth proxy needs to:
+ 1. Accept these dynamic redirect URIs from clients
+ 2. Use its own fixed redirect URI with the upstream provider (Google, GitHub, etc.)
+ 3. Forward the authorization code back to the client's dynamic URI
+
+ Solution:
+ ---------
+ This class overrides redirect_uri validation to accept ANY localhost URI,
+ while the proxy internally uses its own fixed redirect URI with the upstream
+ provider. This allows the flow to work even when clients reconnect with
+ different ports or when tokens are cached.
+
+ Without this class, clients would get "Redirect URI not registered" errors
+ when trying to authenticate with cached tokens, because the stored client
+ would have fixed redirect URIs that don't match the new dynamic port.
+ """
+
+ def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
+ """Accept any localhost redirect URI for DCR clients.
+
+ Since we're acting as a proxy and clients register dynamically,
+ we need to accept their localhost redirect URIs even though they're
+ not pre-registered with us. This is essential for cached token
+ scenarios where the client may reconnect with a different port.
+ """
+ if redirect_uri is not None:
+ # Accept any localhost redirect URI for DCR clients
+ uri_str = str(redirect_uri)
+ if uri_str.startswith(("http://localhost", "http://127.0.0.1")):
+ return redirect_uri
+ # Fall back to normal validation for non-localhost URIs
+ return super().validate_redirect_uri(redirect_uri)
+ # If no redirect_uri provided, use default behavior
+ 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
+
+# HTTP client timeout
+HTTP_TIMEOUT_SECONDS: Final[int] = 30
+
+
+class OAuthProxy(OAuthProvider):
+ """OAuth provider that presents a DCR-compliant interface while proxying to non-DCR IDPs.
+
+ Purpose
+ -------
+ MCP clients expect OAuth providers to support Dynamic Client Registration (DCR),
+ where clients can register themselves dynamically and receive unique credentials.
+ Most enterprise IDPs (Google, GitHub, Azure AD, etc.) don't support DCR and require
+ pre-registered OAuth applications with fixed credentials.
+
+ This proxy bridges that gap by:
+ - Presenting a full DCR-compliant OAuth interface to MCP clients
+ - Translating DCR registration requests to use pre-configured upstream credentials
+ - Proxying all OAuth flows to the upstream IDP with appropriate translations
+ - Managing the state and security requirements of both protocols
+
+ Architecture Overview
+ --------------------
+ The proxy maintains a single OAuth app registration with the upstream provider
+ while allowing unlimited MCP clients to register and authenticate dynamically.
+ It implements the complete OAuth 2.1 + DCR specification for clients while
+ translating to whatever OAuth variant the upstream provider requires.
+
+ Key Translation Challenges Solved
+ ---------------------------------
+ 1. Dynamic Client Registration:
+ - MCP clients expect to register dynamically and get unique credentials
+ - Upstream IDPs require pre-registered apps with fixed credentials
+ - Solution: Accept DCR requests, return shared upstream credentials
+
+ 2. Dynamic Redirect URIs:
+ - MCP clients use random localhost ports that change between sessions
+ - Upstream IDPs require fixed, pre-registered redirect URIs
+ - Solution: Use proxy's fixed callback URL with upstream, forward to client's dynamic URI
+
+ 3. Authorization Code Mapping:
+ - Upstream returns codes for the proxy's redirect URI
+ - Clients expect codes for their own redirect URIs
+ - Solution: Exchange upstream code server-side, issue new code to client
+
+ 4. State Parameter Collision:
+ - Both client and proxy need to maintain state through the flow
+ - Only one state parameter available in OAuth
+ - Solution: Use transaction ID as state with upstream, preserve client's state
+
+ 5. Token Management:
+ - Clients may expect different token formats/claims than upstream provides
+ - Need to track tokens for revocation and refresh
+ - Solution: Store token relationships, forward upstream tokens transparently
+
+ OAuth Flow Implementation
+ ------------------------
+ 1. Client Registration (DCR):
+ - Accept any client registration request
+ - Store ProxyDCRClient that accepts dynamic redirect URIs
+ - Return shared upstream credentials to all clients
+
+ 2. Authorization:
+ - Store transaction mapping client details to proxy flow
+ - Redirect to upstream with proxy's fixed redirect URI
+ - Use transaction ID as state parameter with upstream
+
+ 3. Upstream Callback:
+ - Exchange upstream authorization code for tokens (server-side)
+ - Generate new authorization code bound to client's PKCE challenge
+ - Redirect to client's original dynamic redirect URI
+
+ 4. Token Exchange:
+ - Validate client's code and PKCE verifier
+ - Return previously obtained upstream tokens
+ - Clean up one-time use authorization code
+
+ 5. Token Refresh:
+ - Forward refresh requests to upstream using authlib
+ - Handle token rotation if upstream issues new refresh token
+ - Update local token mappings
+
+ State Management
+ ---------------
+ The proxy maintains minimal but crucial state:
+ - _clients: DCR registrations (all use ProxyDCRClient for flexibility)
+ - _oauth_transactions: Active authorization flows with client context
+ - _client_codes: Authorization codes with PKCE challenges and upstream tokens
+ - _access_tokens, _refresh_tokens: Token storage for revocation
+ - Token relationship mappings for cleanup and rotation
+
+ Security Considerations
+ ----------------------
+ - PKCE enforced end-to-end (client to proxy, proxy to upstream)
+ - Authorization codes are single-use with short expiry
+ - Transaction IDs are cryptographically random
+ - All state is cleaned up after use to prevent replay
+ - Token validation delegates to upstream provider
+
+ Provider Compatibility
+ ---------------------
+ Works with any OAuth 2.0 provider that supports:
+ - Authorization code flow
+ - Fixed redirect URI (configured in provider's app settings)
+ - Standard token endpoint
+
+ Handles provider-specific requirements:
+ - Google: Ensures minimum scope requirements
+ - GitHub: Compatible with OAuth Apps and GitHub Apps
+ - Azure AD: Handles tenant-specific endpoints
+ - Generic: Works with any spec-compliant provider
+ """
+
+ def __init__(
+ self,
+ *,
+ # Upstream server configuration
+ upstream_authorization_endpoint: str,
+ upstream_token_endpoint: str,
+ upstream_client_id: str,
+ upstream_client_secret: str,
+ upstream_revocation_endpoint: str | None = None,
+ # Token validation
+ token_verifier: TokenVerifier,
+ # FastMCP server configuration
+ base_url: AnyHttpUrl | str,
+ redirect_path: str = "/oauth/callback",
+ issuer_url: AnyHttpUrl | str | None = None,
+ service_documentation_url: AnyHttpUrl | str | None = None,
+ resource_server_url: AnyHttpUrl | str | None = None,
+ ):
+ """Initialize the OAuth proxy provider.
+
+ Args:
+ upstream_authorization_endpoint: URL of upstream authorization endpoint
+ upstream_token_endpoint: URL of upstream token endpoint
+ upstream_client_id: Client ID registered with upstream server
+ upstream_client_secret: Client secret for upstream server
+ upstream_revocation_endpoint: Optional upstream revocation endpoint
+ token_verifier: Token verifier for validating access tokens
+ base_url: Public URL of this FastMCP server
+ redirect_path: Redirect path configured in upstream OAuth app (defaults to "/oauth/callback")
+ issuer_url: Issuer URL for OAuth metadata (defaults to base_url)
+ service_documentation_url: Optional service documentation URL
+ resource_server_url: Resource server URL (defaults to base_url)
+ """
+ # Convert string URLs to AnyHttpUrl for parent class
+ base_url_parsed = (
+ AnyHttpUrl(base_url) if isinstance(base_url, str) else base_url
+ )
+ issuer_url_parsed = (
+ (AnyHttpUrl(issuer_url) if isinstance(issuer_url, str) else issuer_url)
+ if issuer_url
+ else None
+ )
+ service_documentation_url_parsed = (
+ (
+ AnyHttpUrl(service_documentation_url)
+ if isinstance(service_documentation_url, str)
+ else service_documentation_url
+ )
+ if service_documentation_url
+ else None
+ )
+ resource_server_url_parsed = (
+ (
+ AnyHttpUrl(resource_server_url)
+ if isinstance(resource_server_url, str)
+ else resource_server_url
+ )
+ if resource_server_url
+ else None
+ )
+
+ # Always enable DCR since we implement it locally for MCP clients
+ client_registration_options = ClientRegistrationOptions(enabled=True)
+
+ # Enable revocation only if upstream endpoint provided
+ revocation_options = (
+ RevocationOptions(enabled=True) if upstream_revocation_endpoint else None
+ )
+
+ super().__init__(
+ base_url=base_url_parsed,
+ issuer_url=issuer_url_parsed,
+ service_documentation_url=service_documentation_url_parsed,
+ client_registration_options=client_registration_options,
+ revocation_options=revocation_options,
+ required_scopes=token_verifier.required_scopes,
+ resource_server_url=resource_server_url_parsed,
+ )
+
+ # Store upstream configuration
+ self._upstream_authorization_endpoint = upstream_authorization_endpoint
+ self._upstream_token_endpoint = upstream_token_endpoint
+ self._upstream_client_id = upstream_client_id
+ self._upstream_client_secret = SecretStr(upstream_client_secret)
+ self._upstream_revocation_endpoint = upstream_revocation_endpoint
+
+ # Store redirect configuration
+ self._redirect_path = (
+ redirect_path if redirect_path.startswith("/") else f"/{redirect_path}"
+ )
+
+ # Local state for DCR and token bookkeeping
+ self._clients: dict[str, OAuthClientInformationFull] = {}
+ self._access_tokens: dict[str, AccessToken] = {}
+ self._refresh_tokens: dict[str, RefreshToken] = {}
+
+ # Token relation mappings for cleanup
+ self._access_to_refresh: dict[str, str] = {}
+ self._refresh_to_access: dict[str, str] = {}
+
+ # OAuth transaction storage for IdP callback forwarding
+ self._oauth_transactions: dict[
+ str, dict[str, Any]
+ ] = {} # txn_id -> transaction_data
+ self._client_codes: dict[str, dict[str, Any]] = {} # client_code -> code_data
+
+ # Use the provided token validator
+ self._token_validator = token_verifier
+
+ logger.info(
+ "Initialized OAuth proxy provider with upstream server %s",
+ self._upstream_authorization_endpoint,
+ )
+
+ # -------------------------------------------------------------------------
+ # Client Registration (Local Implementation)
+ # -------------------------------------------------------------------------
+
+ async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
+ """Get client information by ID.
+
+ For unregistered clients, returns a ProxyDCRClient that accepts
+ any localhost redirect URI for DCR clients.
+
+ Even registered clients use ProxyDCRClient to ensure they can
+ authenticate with different dynamic ports on reconnection. This
+ handles the case where a client with cached tokens reconnects
+ on a different port.
+ """
+ client = self._clients.get(client_id)
+
+ if client is None:
+ # For unregistered DCR clients, create a permissive client
+ # that will accept any localhost redirect URI
+ # We need at least one URI for Pydantic validation, but our custom
+ # validate_redirect_uri will accept any localhost URI
+ client = ProxyDCRClient(
+ client_id=client_id,
+ client_secret=None,
+ redirect_uris=[
+ AnyUrl("http://localhost")
+ ], # Placeholder - we accept any localhost URI
+ grant_types=["authorization_code", "refresh_token"],
+ token_endpoint_auth_method="none",
+ )
+ logger.debug("Created ProxyDCRClient for unregistered client %s", client_id)
+
+ return client
+
+ async def register_client(self, client_info: OAuthClientInformationFull) -> None:
+ """Register a client locally using fixed upstream credentials.
+
+ This implementation always uses the upstream client_id and client_secret
+ regardless of what the client requests. It modifies the client_info object
+ in place since the MCP framework ignores return values.
+
+ This ensures all clients use the same credentials that are registered
+ with the upstream server.
+
+ Implementation Detail:
+ We store a ProxyDCRClient (not the original client_info) to ensure
+ the client can reconnect with different dynamic redirect URIs. This is
+ essential for cached token scenarios where the client port changes.
+
+ The flow:
+ 1. Client provides its desired redirect URIs (dynamic localhost ports)
+ 2. We create a ProxyDCRClient that will accept ANY localhost URI
+ 3. We store this flexible client for future authentications
+ 4. When client reconnects with a different port, ProxyDCRClient accepts it
+ """
+ # Always use the upstream credentials
+ upstream_id = self._upstream_client_id
+ upstream_secret = self._upstream_client_secret.get_secret_value()
+
+ # Create a ProxyDCRClient that accepts any localhost redirect URI
+ proxy_client = ProxyDCRClient(
+ client_id=upstream_id,
+ client_secret=upstream_secret,
+ redirect_uris=client_info.redirect_uris or [AnyUrl("http://localhost")],
+ grant_types=client_info.grant_types
+ or ["authorization_code", "refresh_token"],
+ token_endpoint_auth_method="none",
+ )
+
+ # Modify the client_info object in place (framework ignores return values)
+ client_info.client_id = upstream_id
+ client_info.client_secret = upstream_secret
+ client_info.token_endpoint_auth_method = "none"
+
+ # Ensure correct grant types
+ if not client_info.grant_types:
+ client_info.grant_types = ["authorization_code", "refresh_token"]
+
+ # Store the ProxyDCRClient using the upstream ID
+ self._clients[upstream_id] = proxy_client
+
+ logger.info(
+ "Registered client %s with %d redirect URIs",
+ upstream_id,
+ len(proxy_client.redirect_uris),
+ )
+
+ # -------------------------------------------------------------------------
+ # Authorization Flow (Proxy to Upstream)
+ # -------------------------------------------------------------------------
+
+ async def authorize(
+ self,
+ client: OAuthClientInformationFull,
+ params: AuthorizationParams,
+ ) -> str:
+ """Start OAuth transaction and redirect to upstream IdP.
+
+ This implements the DCR-compliant proxy pattern:
+ 1. Store transaction with client details and PKCE challenge
+ 2. Use transaction ID as state for IdP
+ 3. Redirect to IdP with our fixed callback URL
+ """
+ # Generate transaction ID for this authorization request
+ txn_id = secrets.token_urlsafe(32)
+
+ # Store transaction data for IdP callback processing
+ self._oauth_transactions[txn_id] = {
+ "client_id": client.client_id,
+ "client_redirect_uri": str(params.redirect_uri),
+ "client_state": params.state,
+ "code_challenge": params.code_challenge,
+ "code_challenge_method": getattr(params, "code_challenge_method", "S256"),
+ "scopes": params.scopes or [],
+ "created_at": time.time(),
+ }
+
+ # Build query parameters for upstream IdP authorization request
+ # Use our fixed IdP callback and transaction ID as state
+ query_params: dict[str, Any] = {
+ "response_type": "code",
+ "client_id": self._upstream_client_id,
+ "redirect_uri": f"{str(self.base_url).rstrip('/')}{self._redirect_path}",
+ "state": txn_id, # Use txn_id as IdP state
+ }
+
+ # Add scopes - use client scopes or fallback to required scopes
+ scopes_to_use = params.scopes or self.required_scopes or []
+ # Google requires at least some scope parameter, so provide a minimal one if none specified
+ if (
+ not scopes_to_use
+ and "google" in self._upstream_authorization_endpoint.lower()
+ ):
+ scopes_to_use = ["openid"] # Minimal scope for Google
+
+ if scopes_to_use:
+ query_params["scope"] = " ".join(scopes_to_use)
+
+ # Build the upstream authorization URL
+ upstream_url = (
+ f"{self._upstream_authorization_endpoint}?{urlencode(query_params)}"
+ )
+
+ logger.info(
+ "Starting OAuth transaction %s for client %s, redirecting to IdP",
+ txn_id,
+ client.client_id,
+ )
+
+ return upstream_url
+
+ # -------------------------------------------------------------------------
+ # Authorization Code Handling
+ # -------------------------------------------------------------------------
+
+ async def load_authorization_code(
+ self,
+ client: OAuthClientInformationFull,
+ authorization_code: str,
+ ) -> AuthorizationCode | None:
+ """Load authorization code for validation.
+
+ Look up our client code and return authorization code object
+ with PKCE challenge for validation.
+ """
+ # Look up client code data
+ code_data = self._client_codes.get(authorization_code)
+ if not code_data:
+ logger.debug("Authorization code not found: %s", authorization_code)
+ return None
+
+ # Check if code expired
+ if time.time() > code_data["expires_at"]:
+ logger.debug("Authorization code expired: %s", authorization_code)
+ self._client_codes.pop(authorization_code, None)
+ return None
+
+ # Verify client ID matches
+ if code_data["client_id"] != client.client_id:
+ logger.debug(
+ "Authorization code client ID mismatch: %s vs %s",
+ code_data["client_id"],
+ client.client_id,
+ )
+ return None
+
+ # Create authorization code object with PKCE challenge
+ return AuthorizationCode(
+ code=authorization_code,
+ client_id=client.client_id,
+ redirect_uri=code_data["redirect_uri"],
+ redirect_uri_provided_explicitly=True,
+ scopes=code_data["scopes"],
+ expires_at=code_data["expires_at"],
+ code_challenge=code_data.get("code_challenge", ""),
+ )
+
+ async def exchange_authorization_code(
+ self,
+ client: OAuthClientInformationFull,
+ authorization_code: AuthorizationCode,
+ ) -> OAuthToken:
+ """Exchange authorization code for stored IdP tokens.
+
+ For the DCR-compliant proxy flow, we return the IdP tokens that were obtained
+ during the IdP callback exchange. PKCE validation is handled by the MCP framework.
+ """
+ # Look up stored code data
+ code_data = self._client_codes.get(authorization_code.code)
+ if not code_data:
+ logger.error(
+ "Authorization code not found in client codes: %s",
+ authorization_code.code,
+ )
+ raise TokenError("invalid_grant", "Authorization code not found")
+
+ # Get stored IdP tokens
+ idp_tokens = code_data["idp_tokens"]
+
+ # Clean up client code (one-time use)
+ self._client_codes.pop(authorization_code.code, None)
+
+ # Extract token information for local tracking
+ access_token_value = idp_tokens["access_token"]
+ refresh_token_value = idp_tokens.get("refresh_token")
+ expires_in = int(
+ idp_tokens.get("expires_in", DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
+ )
+ expires_at = int(time.time() + expires_in)
+
+ # Store access token locally for tracking
+ access_token = AccessToken(
+ token=access_token_value,
+ client_id=client.client_id,
+ scopes=authorization_code.scopes,
+ expires_at=expires_at,
+ )
+ self._access_tokens[access_token_value] = access_token
+
+ # Store refresh token if provided
+ if refresh_token_value:
+ refresh_token = RefreshToken(
+ token=refresh_token_value,
+ client_id=client.client_id,
+ scopes=authorization_code.scopes,
+ expires_at=None, # Refresh tokens typically don't expire
+ )
+ self._refresh_tokens[refresh_token_value] = refresh_token
+
+ # Maintain token relationships for cleanup
+ self._access_to_refresh[access_token_value] = refresh_token_value
+ self._refresh_to_access[refresh_token_value] = access_token_value
+
+ logger.info(
+ "Successfully exchanged client code for stored IdP tokens (client: %s)",
+ client.client_id,
+ )
+
+ return OAuthToken(**idp_tokens) # type: ignore[arg-type]
+
+ # -------------------------------------------------------------------------
+ # Refresh Token Flow
+ # -------------------------------------------------------------------------
+
+ async def load_refresh_token(
+ self,
+ client: OAuthClientInformationFull,
+ refresh_token: str,
+ ) -> RefreshToken | None:
+ """Load refresh token from local storage."""
+ return self._refresh_tokens.get(refresh_token)
+
+ async def exchange_refresh_token(
+ self,
+ client: OAuthClientInformationFull,
+ refresh_token: RefreshToken,
+ scopes: list[str],
+ ) -> OAuthToken:
+ """Exchange refresh token for new access token using authlib."""
+
+ # Use authlib's AsyncOAuth2Client for refresh token exchange
+ oauth_client = AsyncOAuth2Client(
+ client_id=self._upstream_client_id,
+ client_secret=self._upstream_client_secret.get_secret_value(),
+ timeout=HTTP_TIMEOUT_SECONDS,
+ )
+
+ try:
+ logger.debug("Using authlib to refresh token from upstream")
+
+ # Let authlib handle the refresh token exchange
+ token_response: dict[str, Any] = await oauth_client.refresh_token( # type: ignore[misc]
+ url=self._upstream_token_endpoint,
+ refresh_token=refresh_token.token,
+ scope=" ".join(scopes) if scopes else None,
+ )
+
+ logger.info(
+ "Successfully refreshed access token via authlib (client: %s)",
+ client.client_id,
+ )
+
+ except Exception as e:
+ logger.error("Authlib refresh token exchange failed: %s", e)
+ raise TokenError(
+ "invalid_grant", f"Upstream refresh token exchange failed: {e}"
+ ) from e
+
+ # Update local token storage
+ new_access_token = token_response["access_token"]
+ expires_in = int(
+ token_response.get("expires_in", DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
+ )
+
+ self._access_tokens[new_access_token] = AccessToken(
+ token=new_access_token,
+ client_id=client.client_id,
+ scopes=scopes,
+ expires_at=int(time.time() + expires_in),
+ )
+
+ # Handle refresh token rotation if new one provided
+ if "refresh_token" in token_response:
+ new_refresh_token = token_response["refresh_token"]
+ if new_refresh_token != refresh_token.token:
+ # Remove old refresh token
+ self._refresh_tokens.pop(refresh_token.token, None)
+ old_access = self._refresh_to_access.pop(refresh_token.token, None)
+ if old_access:
+ self._access_to_refresh.pop(old_access, None)
+
+ # Store new refresh token
+ self._refresh_tokens[new_refresh_token] = RefreshToken(
+ token=new_refresh_token,
+ client_id=client.client_id,
+ scopes=scopes,
+ expires_at=None,
+ )
+ self._access_to_refresh[new_access_token] = new_refresh_token
+ self._refresh_to_access[new_refresh_token] = new_access_token
+
+ return OAuthToken(**token_response) # type: ignore[arg-type]
+
+ # -------------------------------------------------------------------------
+ # Token Validation
+ # -------------------------------------------------------------------------
+
+ async def load_access_token(self, token: str) -> AccessToken | None:
+ """Validate access token using upstream JWKS.
+
+ Delegates to the JWT verifier which handles signature validation,
+ expiration checking, and claims validation using the upstream JWKS.
+ """
+ result = await self._token_validator.verify_token(token)
+ if result:
+ logger.debug("Token validated successfully")
+ else:
+ logger.debug("Token validation failed")
+ return result
+
+ # -------------------------------------------------------------------------
+ # Token Revocation
+ # -------------------------------------------------------------------------
+
+ async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
+ """Revoke token locally and with upstream server if supported.
+
+ Removes tokens from local storage and attempts to revoke them with
+ the upstream server if a revocation endpoint is configured.
+ """
+ # Clean up local token storage
+ if isinstance(token, AccessToken):
+ self._access_tokens.pop(token.token, None)
+ # Also remove associated refresh token
+ paired_refresh = self._access_to_refresh.pop(token.token, None)
+ if paired_refresh:
+ self._refresh_tokens.pop(paired_refresh, None)
+ self._refresh_to_access.pop(paired_refresh, None)
+ else: # RefreshToken
+ self._refresh_tokens.pop(token.token, None)
+ # Also remove associated access token
+ paired_access = self._refresh_to_access.pop(token.token, None)
+ if paired_access:
+ self._access_tokens.pop(paired_access, None)
+ self._access_to_refresh.pop(paired_access, None)
+
+ # Attempt upstream revocation if endpoint is configured
+ if self._upstream_revocation_endpoint:
+ try:
+ async with httpx.AsyncClient(
+ timeout=HTTP_TIMEOUT_SECONDS
+ ) as http_client:
+ await http_client.post(
+ self._upstream_revocation_endpoint,
+ data={"token": token.token},
+ auth=(
+ self._upstream_client_id,
+ self._upstream_client_secret.get_secret_value(),
+ ),
+ )
+ logger.info("Successfully revoked token with upstream server")
+ except Exception as e:
+ logger.warning("Failed to revoke token with upstream server: %s", e)
+ else:
+ logger.debug("No upstream revocation endpoint configured")
+
+ logger.info("Token revoked successfully")
+
+ # -------------------------------------------------------------------------
+ # Custom Route Handling
+ # -------------------------------------------------------------------------
+
+ async def _handle_proxy_token_request(self, request: Request) -> JSONResponse:
+ """Custom token endpoint using authlib for upstream requests.
+
+ This handler uses authlib's OAuth2Client to forward token requests to the
+ upstream OAuth server, automatically handling response format differences.
+ """
+ try:
+ # Parse the incoming request form data
+ form_data = await request.form()
+
+ # Log the incoming request (with sensitive data redacted)
+ redacted_form = {
+ k: (
+ str(v)[:8] + "..."
+ if k in {"code", "code_verifier", "client_secret", "refresh_token"}
+ and v
+ else str(v)
+ )
+ for k, v in form_data.items()
+ }
+ logger.debug("Proxy token request form data: %s", redacted_form)
+
+ # Create authlib OAuth2 client
+ oauth_client = AsyncOAuth2Client(
+ client_id=self._upstream_client_id,
+ client_secret=self._upstream_client_secret.get_secret_value(),
+ timeout=HTTP_TIMEOUT_SECONDS,
+ )
+
+ grant_type = str(form_data.get("grant_type", ""))
+
+ if grant_type == "authorization_code":
+ # Authorization code grant
+ try:
+ token_data: dict[str, Any] = await oauth_client.fetch_token( # type: ignore[misc]
+ url=self._upstream_token_endpoint,
+ code=str(form_data.get("code", "")),
+ redirect_uri=str(form_data.get("redirect_uri", "")),
+ code_verifier=str(form_data.get("code_verifier"))
+ if "code_verifier" in form_data
+ else None,
+ )
+
+ # Store tokens locally for tracking
+ if "access_token" in token_data:
+ self._store_tokens_from_response(token_data)
+
+ logger.info(
+ "Successfully proxied authorization code exchange via authlib"
+ )
+
+ except Exception as e:
+ logger.error("Authlib authorization code exchange failed: %s", e)
+ return JSONResponse(
+ content={
+ "error": "invalid_grant",
+ "error_description": f"Authorization code exchange failed: {e}",
+ },
+ status_code=400,
+ )
+
+ elif grant_type == "refresh_token":
+ # Refresh token grant
+ try:
+ token_data: dict[str, Any] = await oauth_client.refresh_token( # type: ignore[misc]
+ url=self._upstream_token_endpoint,
+ refresh_token=str(form_data.get("refresh_token", "")),
+ scope=str(form_data.get("scope"))
+ if "scope" in form_data
+ else None,
+ )
+
+ logger.info(
+ "Successfully proxied refresh token exchange via authlib"
+ )
+
+ except Exception as e:
+ logger.error("Authlib refresh token exchange failed: %s", e)
+ return JSONResponse(
+ content={
+ "error": "invalid_grant",
+ "error_description": f"Refresh token exchange failed: {e}",
+ },
+ status_code=400,
+ )
+ else:
+ # Unsupported grant type
+ logger.error("Unsupported grant type: %s", grant_type)
+ return JSONResponse(
+ content={
+ "error": "unsupported_grant_type",
+ "error_description": f"Grant type '{grant_type}' not supported by proxy",
+ },
+ status_code=400,
+ )
+
+ return JSONResponse(content=token_data)
+
+ except Exception as e:
+ logger.error("Error in proxy token handler: %s", e, exc_info=True)
+ return JSONResponse(
+ content={
+ "error": "server_error",
+ "error_description": "Internal server error",
+ },
+ status_code=500,
+ )
+
+ def _store_tokens_from_response(self, token_data: dict[str, Any]) -> None:
+ """Store tokens from upstream response for local tracking."""
+ try:
+ access_token_value = token_data.get("access_token")
+ refresh_token_value = token_data.get("refresh_token")
+ expires_in = int(
+ token_data.get("expires_in", DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
+ )
+ expires_at = int(time.time() + expires_in)
+
+ if access_token_value:
+ access_token = AccessToken(
+ token=access_token_value,
+ client_id=self._upstream_client_id,
+ scopes=[], # Will be determined by token validation
+ expires_at=expires_at,
+ )
+ self._access_tokens[access_token_value] = access_token
+
+ if refresh_token_value:
+ refresh_token = RefreshToken(
+ token=refresh_token_value,
+ client_id=self._upstream_client_id,
+ scopes=[],
+ expires_at=None,
+ )
+ self._refresh_tokens[refresh_token_value] = refresh_token
+
+ # Maintain token relationships
+ self._access_to_refresh[access_token_value] = refresh_token_value
+ self._refresh_to_access[refresh_token_value] = access_token_value
+
+ logger.debug("Stored tokens from upstream response for tracking")
+
+ except Exception as e:
+ logger.warning("Failed to store tokens from upstream response: %s", e)
+
+ def get_routes(self) -> list[Route]:
+ """Get OAuth routes with custom proxy token handler.
+
+ This method creates standard OAuth routes and replaces the token endpoint
+ with our proxy handler that forwards requests to the upstream OAuth server.
+ """
+ # Get standard OAuth routes from parent class
+ routes = super().get_routes()
+ custom_routes = []
+ token_route_found = False
+
+ logger.info(
+ f"get_routes called - configuring OAuth routes in {len(routes)} routes"
+ )
+
+ for i, route in enumerate(routes):
+ logger.debug(
+ f"Route {i}: {route} - path: {getattr(route, 'path', 'N/A')}, methods: {getattr(route, 'methods', 'N/A')}"
+ )
+
+ # Keep all standard OAuth routes unchanged - our DCR-compliant flow handles everything
+ custom_routes.append(route)
+
+ if (
+ isinstance(route, Route)
+ and route.path == "/token"
+ and route.methods is not None
+ and "POST" in route.methods
+ ):
+ token_route_found = True
+ logger.info("✅ KEEPING standard token endpoint for DCR-compliant flow")
+
+ if not token_route_found:
+ logger.warning("⚠️ No /token POST route found!")
+ # This shouldn't happen with standard OAuth provider
+
+ # Add OAuth callback endpoint for forwarding to client callbacks
+ custom_routes.append(
+ Route(
+ path=self._redirect_path,
+ endpoint=self._handle_idp_callback,
+ methods=["GET"],
+ )
+ )
+
+ logger.info(
+ f"✅ OAuth routes configured: token_endpoint={token_route_found}, total routes={len(custom_routes)} (includes OAuth callback)"
+ )
+ return custom_routes
+
+ # -------------------------------------------------------------------------
+ # IdP Callback Forwarding
+ # -------------------------------------------------------------------------
+
+ async def _handle_idp_callback(self, request: Request) -> RedirectResponse:
+ """Handle callback from upstream IdP and forward to client.
+
+ This implements the DCR-compliant callback forwarding:
+ 1. Receive IdP callback with code and txn_id as state
+ 2. Exchange IdP code for tokens (server-side)
+ 3. Generate our own client code bound to PKCE challenge
+ 4. Redirect to client's callback with client code and original state
+ """
+ try:
+ idp_code = request.query_params.get("code")
+ txn_id = request.query_params.get("state")
+ error = request.query_params.get("error")
+
+ if error:
+ logger.error(
+ "IdP callback error: %s - %s",
+ error,
+ request.query_params.get("error_description"),
+ )
+ # TODO: Forward error to client callback
+ return RedirectResponse(
+ url=f"data:text/html,OAuth Error
{error}: {request.query_params.get('error_description', 'Unknown error')}
",
+ status_code=302,
+ )
+
+ if not idp_code or not txn_id:
+ logger.error("IdP callback missing code or transaction ID")
+ return RedirectResponse(
+ url="data:text/html,OAuth Error
Missing authorization code or transaction ID
",
+ status_code=302,
+ )
+
+ # Look up transaction data
+ transaction = self._oauth_transactions.get(txn_id)
+ if not transaction:
+ logger.error("IdP callback with invalid transaction ID: %s", txn_id)
+ return RedirectResponse(
+ url="data:text/html,OAuth Error
Invalid or expired transaction
",
+ status_code=302,
+ )
+
+ # Exchange IdP code for tokens (server-side)
+ oauth_client = AsyncOAuth2Client(
+ client_id=self._upstream_client_id,
+ client_secret=self._upstream_client_secret.get_secret_value(),
+ timeout=HTTP_TIMEOUT_SECONDS,
+ )
+
+ try:
+ idp_redirect_uri = (
+ f"{str(self.base_url).rstrip('/')}{self._redirect_path}"
+ )
+ logger.debug(
+ f"Exchanging IdP code for tokens with redirect_uri: {idp_redirect_uri}"
+ )
+
+ 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,
+ )
+
+ logger.info(
+ f"Successfully exchanged IdP code for tokens (transaction: {txn_id})"
+ )
+
+ except Exception as e:
+ logger.error("IdP token exchange failed: %s", e)
+ # TODO: Forward error to client callback
+ return RedirectResponse(
+ url=f"data:text/html,OAuth Error
Token exchange failed: {e}
",
+ status_code=302,
+ )
+
+ # Generate our own authorization code for the client
+ client_code = secrets.token_urlsafe(32)
+ code_expires_at = int(time.time() + DEFAULT_AUTH_CODE_EXPIRY_SECONDS)
+
+ # Store client code with PKCE challenge and IdP tokens
+ self._client_codes[client_code] = {
+ "client_id": transaction["client_id"],
+ "redirect_uri": transaction["client_redirect_uri"],
+ "code_challenge": transaction["code_challenge"],
+ "code_challenge_method": transaction["code_challenge_method"],
+ "scopes": transaction["scopes"],
+ "idp_tokens": idp_tokens,
+ "expires_at": code_expires_at,
+ "created_at": time.time(),
+ }
+
+ # Clean up transaction
+ self._oauth_transactions.pop(txn_id, None)
+
+ # Build client callback URL with our code and original state
+ client_redirect_uri = transaction["client_redirect_uri"]
+ client_state = transaction["client_state"]
+
+ callback_params = {
+ "code": client_code,
+ "state": client_state,
+ }
+
+ # Add query parameters to client redirect URI
+ separator = "&" if "?" in client_redirect_uri else "?"
+ client_callback_url = (
+ f"{client_redirect_uri}{separator}{urlencode(callback_params)}"
+ )
+
+ logger.debug(f"Forwarding to client callback for transaction {txn_id}")
+
+ return RedirectResponse(url=client_callback_url, status_code=302)
+
+ except Exception as e:
+ logger.error("Error in IdP callback handler: %s", e, exc_info=True)
+ return RedirectResponse(
+ url="data:text/html,OAuth Error
Internal server error during IdP callback
",
+ status_code=302,
+ )
diff --git a/src/fastmcp/utilities/tests.py b/src/fastmcp/utilities/tests.py
index 4a2e6061e..705e2b22f 100644
--- a/src/fastmcp/utilities/tests.py
+++ b/src/fastmcp/utilities/tests.py
@@ -76,6 +76,8 @@ def run_server_in_process(
server_fn: Callable[..., None],
*args,
provide_host_and_port: bool = True,
+ host: str = "127.0.0.1",
+ port: int | None = None,
**kwargs,
) -> Generator[str, None, None]:
"""
@@ -87,13 +89,16 @@ def run_server_in_process(
not pickleable, so we need a function that creates and runs one.
*args: Arguments to pass to the server function.
provide_host_and_port: Whether to provide the host and port to the server function as kwargs.
+ host: Host to bind the server to (default: "127.0.0.1").
+ port: Port to bind the server to (default: find available port).
**kwargs: Keyword arguments to pass to the server function.
Returns:
The server URL.
"""
- host = "127.0.0.1"
- port = find_available_port()
+ # Use provided port or find an available one
+ if port is None:
+ port = find_available_port()
if provide_host_and_port:
kwargs |= {"host": host, "port": port}
diff --git a/test_github_oauth.py b/test_github_oauth.py
new file mode 100644
index 000000000..c008d82a1
--- /dev/null
+++ b/test_github_oauth.py
@@ -0,0 +1,21 @@
+import os
+
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.github import GitHubProvider
+
+auth = GitHubProvider(
+ client_id=os.getenv("FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID") or "",
+ client_secret=os.getenv("FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET") or "",
+ base_url="http://localhost:8000",
+)
+
+mcp = FastMCP("GitHub OAuth Test Server", auth=auth)
+
+
+@mcp.tool
+def echo(message: str) -> str:
+ return message
+
+
+if __name__ == "__main__":
+ mcp.run(transport="http", port=8000)
diff --git a/test_google_oauth.py b/test_google_oauth.py
new file mode 100644
index 000000000..5e9270038
--- /dev/null
+++ b/test_google_oauth.py
@@ -0,0 +1,22 @@
+import os
+
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.google import GoogleProvider
+
+auth = GoogleProvider(
+ client_id=os.getenv("FASTMCP_TEST_AUTH_GOOGLE_CLIENT_ID") or "",
+ client_secret=os.getenv("FASTMCP_TEST_AUTH_GOOGLE_CLIENT_SECRET") or "",
+ base_url="http://localhost:8000",
+ required_scopes=["openid"],
+)
+
+mcp = FastMCP("Google OAuth Test Server", auth=auth)
+
+
+@mcp.tool
+def echo(message: str) -> str:
+ return message
+
+
+if __name__ == "__main__":
+ mcp.run(transport="http", port=8000)
diff --git a/tests/integration_tests/auth/__init__.py b/tests/integration_tests/auth/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/integration_tests/auth/test_github_provider_integration.py b/tests/integration_tests/auth/test_github_provider_integration.py
new file mode 100644
index 000000000..ad49d5e7c
--- /dev/null
+++ b/tests/integration_tests/auth/test_github_provider_integration.py
@@ -0,0 +1,356 @@
+"""Integration tests for GitHub OAuth Provider.
+
+Tests the complete GitHub OAuth flow using HeadlessOAuth to bypass browser interaction.
+
+This test requires a GitHub OAuth app to be created at https://github.com/settings/developers
+with the following configuration:
+- Redirect URL: http://127.0.0.1:9100/oauth/callback
+- Client ID and Client Secret should be set as environment variables:
+ - FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID
+ - FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET
+"""
+
+import os
+from collections.abc import Generator
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+import pytest
+
+from fastmcp import FastMCP
+from fastmcp.client import Client
+from fastmcp.server.auth.providers.github import GitHubProvider
+from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
+
+FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID = os.getenv("FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID")
+FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET = os.getenv(
+ "FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET"
+)
+
+# Skip tests if no GitHub OAuth credentials are available
+pytestmark = pytest.mark.xfail(
+ not FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID
+ or not FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET,
+ reason="FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID and FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET environment variables are not set or empty",
+)
+
+
+def create_github_server(host: str = "127.0.0.1", port: int = 9100, **kwargs) -> None:
+ """Create FastMCP server with GitHub OAuth protection."""
+ assert FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID is not None
+ assert FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET is not None
+
+ # Create GitHub OAuth provider
+ auth = GitHubProvider(
+ client_id=FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID,
+ client_secret=FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET,
+ base_url=f"http://{host}:{port}",
+ )
+
+ # Create FastMCP server with GitHub authentication
+ server = FastMCP("GitHub OAuth Integration Test Server", auth=auth)
+
+ @server.tool
+ def get_protected_data() -> str:
+ """Returns protected data - requires GitHub OAuth."""
+ return "🔐 This data requires GitHub OAuth authentication!"
+
+ @server.tool
+ def get_user_info() -> str:
+ """Returns user info from OAuth context."""
+ return "📝 GitHub OAuth user authenticated successfully"
+
+ # Run the server
+ server.run(host=host, port=port, **kwargs)
+
+
+def create_github_server_with_mock_callback(
+ host: str = "127.0.0.1", port: int = 9100, **kwargs
+) -> None:
+ """Create FastMCP server with GitHub OAuth that mocks the callback for testing."""
+ assert FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID is not None
+ assert FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET is not None
+
+ # Create GitHub OAuth provider
+ auth = GitHubProvider(
+ client_id=FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID,
+ client_secret=FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET,
+ base_url=f"http://{host}:{port}",
+ )
+
+ # Mock the authorize method to return a fake code instead of redirecting to GitHub
+ async def mock_authorize(client, params):
+ # Instead of redirecting to GitHub, simulate an immediate callback
+ import secrets
+ import time
+
+ # Generate a fake authorization code
+ fake_code = secrets.token_urlsafe(32)
+
+ # Create mock token response (simulating what GitHub would return)
+ mock_tokens = {
+ "access_token": f"gho_mock_token_{secrets.token_hex(16)}",
+ "token_type": "bearer",
+ "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(),
+ }
+
+ # Return the redirect to the client's callback with the fake code
+ callback_params = {
+ "code": fake_code,
+ "state": params.state,
+ }
+ from urllib.parse import urlencode
+
+ separator = "&" if "?" in str(params.redirect_uri) else "?"
+ return f"{params.redirect_uri}{separator}{urlencode(callback_params)}"
+
+ auth.authorize = mock_authorize
+
+ # Mock the token verifier to accept our fake tokens
+ original_verify_token = auth._token_validator.verify_token
+
+ async def mock_verify_token(token: str):
+ if token.startswith("gho_mock_token_"):
+ # Return a mock AccessToken for our fake tokens
+ import time
+
+ from fastmcp.server.auth.auth import AccessToken
+
+ return AccessToken(
+ token=token,
+ client_id=FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID or "test-client",
+ scopes=["user"],
+ expires_at=int(time.time() + 3600),
+ )
+ # Fall back to original verification for other tokens
+ return await original_verify_token(token)
+
+ auth._token_validator.verify_token = mock_verify_token
+
+ # Create FastMCP server with mocked GitHub authentication
+ server = FastMCP("GitHub OAuth Integration Test Server (Mock)", auth=auth)
+
+ @server.tool
+ def get_protected_data() -> str:
+ """Returns protected data - requires GitHub OAuth."""
+ return "🔐 This data requires GitHub OAuth authentication!"
+
+ @server.tool
+ def get_user_info() -> str:
+ """Returns user info from OAuth context."""
+ return "📝 GitHub OAuth user authenticated successfully"
+
+ # Run the server
+ server.run(host=host, port=port, **kwargs)
+
+
+@pytest.fixture(scope="module")
+def github_server() -> Generator[str, None, None]:
+ """Start GitHub OAuth server in background process on fixed port 9100."""
+ with run_server_in_process(
+ create_github_server, transport="http", host="127.0.0.1", port=9100
+ ) as url:
+ yield f"{url}/mcp"
+
+
+@pytest.fixture(scope="module")
+def github_server_with_mock() -> Generator[str, None, None]:
+ """Start GitHub OAuth server with mocked callback in background process on port 9101."""
+ with run_server_in_process(
+ create_github_server_with_mock_callback,
+ transport="http",
+ host="127.0.0.1",
+ port=9101,
+ ) as url:
+ yield f"{url}/mcp"
+
+
+@pytest.fixture
+def github_client(github_server: str) -> Client:
+ """Create FastMCP client with HeadlessOAuth for GitHub server."""
+ return Client(
+ github_server,
+ auth=HeadlessOAuth(mcp_url=github_server),
+ )
+
+
+@pytest.fixture
+def github_client_with_mock(github_server_with_mock: str) -> Client:
+ """Create FastMCP client with HeadlessOAuth for mocked GitHub server."""
+ return Client(
+ github_server_with_mock,
+ auth=HeadlessOAuth(mcp_url=github_server_with_mock),
+ )
+
+
+async def test_github_oauth_credentials_available():
+ """Test that GitHub OAuth credentials are available for testing."""
+ assert FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID is not None
+ assert FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET is not None
+ assert len(FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID) > 0
+ assert len(FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET) > 0
+
+
+async def test_github_oauth_authorization_redirect(github_server: str):
+ """Test that GitHub OAuth authorization redirects to GitHub correctly.
+
+ 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
+ """
+ # Extract base URL
+ parsed = urlparse(github_server)
+ base_url = f"{parsed.scheme}://{parsed.netloc}"
+
+ async with httpx.AsyncClient() as http_client:
+ # Step 1: Register OAuth client (DCR)
+ register_response = await http_client.post(
+ f"{base_url}/register",
+ json={
+ "client_name": "Integration Test Client",
+ "redirect_uris": ["http://localhost:12345/callback"],
+ "grant_types": ["authorization_code", "refresh_token"],
+ "response_types": ["code"],
+ "token_endpoint_auth_method": "client_secret_post",
+ },
+ )
+ if register_response.status_code != 201:
+ print(f"Registration failed: {register_response.status_code}")
+ print(f"Response: {register_response.text}")
+ assert register_response.status_code == 201
+
+ client_info = register_response.json()
+ client_id = client_info["client_id"]
+ assert client_id is not None
+
+ # Step 2: Test authorization endpoint redirects to GitHub
+ auth_url = f"{base_url}/authorize"
+ auth_params = {
+ "response_type": "code",
+ "client_id": client_id,
+ "redirect_uri": "http://localhost:12345/callback",
+ "state": "test-state-123",
+ "code_challenge": "test-challenge",
+ "code_challenge_method": "S256",
+ }
+
+ auth_response = await http_client.get(
+ auth_url, params=auth_params, follow_redirects=False
+ )
+
+ # Should redirect to GitHub
+ assert auth_response.status_code == 302
+ redirect_location = auth_response.headers["location"]
+
+ # Parse redirect URL - should be GitHub
+ redirect_parsed = urlparse(redirect_location)
+ assert redirect_parsed.hostname == "github.com"
+ assert redirect_parsed.path == "/login/oauth/authorize"
+
+ # Check that GitHub gets the right parameters
+ github_params = parse_qs(redirect_parsed.query)
+ assert "client_id" in github_params
+ assert github_params["client_id"][0] == FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID
+ assert "redirect_uri" in github_params
+ # The redirect_uri should be our proxy's callback, not the client's
+ proxy_callback = github_params["redirect_uri"][0]
+ assert proxy_callback.startswith(base_url)
+ assert proxy_callback.endswith("/oauth/callback")
+
+
+async def test_github_oauth_server_metadata(github_server: str):
+ """Test OAuth server metadata discovery."""
+ from urllib.parse import urlparse
+
+ import httpx
+
+ # Extract base URL from server URL
+ parsed = urlparse(github_server)
+ base_url = f"{parsed.scheme}://{parsed.netloc}"
+
+ async with httpx.AsyncClient() as http_client:
+ # Test OAuth authorization server metadata
+ metadata_response = await http_client.get(
+ f"{base_url}/.well-known/oauth-authorization-server"
+ )
+ assert metadata_response.status_code == 200
+
+ metadata = metadata_response.json()
+ assert "authorization_endpoint" in metadata
+ assert "token_endpoint" in metadata
+ assert "registration_endpoint" in metadata
+ assert "issuer" in metadata
+
+ # Verify endpoints are properly formed
+ assert metadata["authorization_endpoint"].startswith(base_url)
+ assert metadata["token_endpoint"].startswith(base_url)
+ assert metadata["registration_endpoint"].startswith(base_url)
+
+
+async def test_github_oauth_unauthorized_access(github_server: str):
+ """Test that unauthenticated requests are rejected."""
+ import httpx
+
+ from fastmcp.client.transports import StreamableHttpTransport
+
+ # Create client without OAuth authentication
+ unauthorized_client = Client(transport=StreamableHttpTransport(github_server))
+
+ # Attempt to connect without authentication should fail
+ with pytest.raises(httpx.HTTPStatusError, match="401 Unauthorized"):
+ async with unauthorized_client:
+ pass
+
+
+async def test_github_oauth_with_mock(github_client_with_mock: Client):
+ """Test complete GitHub OAuth flow with mocked callback."""
+ async with github_client_with_mock:
+ # Test that we can ping the server (requires successful OAuth)
+ assert await github_client_with_mock.ping()
+
+ # Test that we can call protected tools
+ result = await github_client_with_mock.call_tool("get_protected_data", {})
+ assert "🔐 This data requires GitHub OAuth authentication!" in str(result.data)
+
+ # Test that we can call user info tool
+ result = await github_client_with_mock.call_tool("get_user_info", {})
+ assert "📝 GitHub OAuth user authenticated successfully" in str(result.data)
+
+
+async def test_github_oauth_mock_only_accepts_mock_tokens(github_server_with_mock: str):
+ """Test that the mock token verifier only accepts mock tokens, not real ones."""
+ from urllib.parse import urlparse
+
+ import httpx
+
+ # Extract base URL
+ parsed = urlparse(github_server_with_mock)
+ base_url = f"{parsed.scheme}://{parsed.netloc}"
+
+ async with httpx.AsyncClient() as http_client:
+ # Test that a fake "real" GitHub token is rejected
+ fake_real_token = "gho_real_token_should_be_rejected"
+
+ auth_response = await http_client.post(
+ f"{base_url}/mcp",
+ headers={
+ "Authorization": f"Bearer {fake_real_token}",
+ "Content-Type": "application/json",
+ },
+ json={"jsonrpc": "2.0", "id": 1, "method": "ping"},
+ )
+
+ # Should be unauthorized because it's not a mock token
+ assert auth_response.status_code == 401
diff --git a/tests/server/auth/providers/test_github.py b/tests/server/auth/providers/test_github.py
new file mode 100644
index 000000000..31505679a
--- /dev/null
+++ b/tests/server/auth/providers/test_github.py
@@ -0,0 +1,227 @@
+"""Unit tests for GitHub OAuth provider."""
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from fastmcp.server.auth.providers.github import (
+ GitHubProvider,
+ GitHubProviderSettings,
+ GitHubTokenVerifier,
+)
+
+
+class TestGitHubProviderSettings:
+ """Test settings for GitHub OAuth provider."""
+
+ def test_settings_from_env_vars(self):
+ """Test that settings can be loaded from environment variables."""
+ with patch.dict(
+ os.environ,
+ {
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID": "env_client_id",
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET": "env_secret",
+ "FASTMCP_SERVER_AUTH_GITHUB_BASE_URL": "https://example.com",
+ "FASTMCP_SERVER_AUTH_GITHUB_REDIRECT_PATH": "/custom/callback",
+ "FASTMCP_SERVER_AUTH_GITHUB_TIMEOUT_SECONDS": "30",
+ },
+ ):
+ settings = GitHubProviderSettings()
+
+ assert settings.client_id == "env_client_id"
+ assert (
+ settings.client_secret
+ and settings.client_secret.get_secret_value() == "env_secret"
+ )
+ assert settings.base_url == "https://example.com"
+ assert settings.redirect_path == "/custom/callback"
+ assert settings.timeout_seconds == 30
+
+ def test_settings_explicit_override_env(self):
+ """Test that explicit settings override environment variables."""
+ with patch.dict(
+ os.environ,
+ {
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID": "env_client_id",
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET": "env_secret",
+ },
+ ):
+ settings = GitHubProviderSettings.model_validate(
+ {
+ "client_id": "explicit_client_id",
+ "client_secret": "explicit_secret",
+ }
+ )
+
+ assert settings.client_id == "explicit_client_id"
+ assert (
+ settings.client_secret
+ and settings.client_secret.get_secret_value() == "explicit_secret"
+ )
+
+
+class TestGitHubProvider:
+ """Test GitHubProvider initialization."""
+
+ def test_init_with_explicit_params(self):
+ """Test initialization with explicit parameters."""
+ provider = GitHubProvider(
+ client_id="test_client",
+ client_secret="test_secret",
+ base_url="https://example.com",
+ redirect_path="/custom/callback",
+ required_scopes=["user", "repo"],
+ timeout_seconds=30,
+ )
+
+ # Check that the provider was initialized correctly
+ assert provider._upstream_client_id == "test_client"
+ assert provider._upstream_client_secret.get_secret_value() == "test_secret"
+ assert (
+ str(provider.base_url) == "https://example.com/"
+ ) # URLs get normalized with trailing slash
+ assert provider._redirect_path == "/custom/callback"
+
+ def test_init_with_env_vars(self):
+ """Test initialization with environment variables."""
+ with patch.dict(
+ os.environ,
+ {
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID": "env_client_id",
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET": "env_secret",
+ "FASTMCP_SERVER_AUTH_GITHUB_BASE_URL": "https://env-example.com",
+ },
+ ):
+ provider = GitHubProvider()
+
+ assert provider._upstream_client_id == "env_client_id"
+ assert provider._upstream_client_secret.get_secret_value() == "env_secret"
+ assert str(provider.base_url) == "https://env-example.com/"
+
+ def test_init_explicit_overrides_env(self):
+ """Test that explicit parameters override environment variables."""
+ with patch.dict(
+ os.environ,
+ {
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID": "env_client_id",
+ "FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET": "env_secret",
+ },
+ ):
+ provider = GitHubProvider(
+ client_id="explicit_client",
+ client_secret="explicit_secret",
+ )
+
+ assert provider._upstream_client_id == "explicit_client"
+ assert (
+ provider._upstream_client_secret.get_secret_value() == "explicit_secret"
+ )
+
+ def test_init_missing_client_id_raises_error(self):
+ """Test that missing client_id raises ValueError."""
+ with pytest.raises(ValueError, match="client_id is required"):
+ GitHubProvider(client_secret="test_secret")
+
+ def test_init_missing_client_secret_raises_error(self):
+ """Test that missing client_secret raises ValueError."""
+ with pytest.raises(ValueError, match="client_secret is required"):
+ GitHubProvider(client_id="test_client")
+
+ def test_init_defaults(self):
+ """Test that default values are applied correctly."""
+ provider = GitHubProvider(
+ client_id="test_client",
+ client_secret="test_secret",
+ )
+
+ # Check defaults
+ assert str(provider.base_url) == "http://localhost:8000/"
+ assert provider._redirect_path == "/oauth/callback"
+ # The required_scopes should be passed to the token verifier
+ assert provider._token_validator.required_scopes == ["user"]
+
+
+class TestGitHubTokenVerifier:
+ """Test GitHubTokenVerifier."""
+
+ def test_init_with_custom_scopes(self):
+ """Test initialization with custom required scopes."""
+ verifier = GitHubTokenVerifier(
+ required_scopes=["user", "repo"],
+ timeout_seconds=30,
+ )
+
+ assert verifier.required_scopes == ["user", "repo"]
+ assert verifier.timeout_seconds == 30
+
+ def test_init_defaults(self):
+ """Test initialization with defaults."""
+ verifier = GitHubTokenVerifier()
+
+ assert (
+ verifier.required_scopes == []
+ ) # Parent TokenVerifier sets empty list as default
+ assert verifier.timeout_seconds == 10
+
+ @pytest.mark.asyncio
+ async def test_verify_token_github_api_failure(self):
+ """Test token verification when GitHub API returns error."""
+ verifier = GitHubTokenVerifier()
+
+ # Mock httpx.AsyncClient to simulate GitHub API failure
+ with patch("httpx.AsyncClient") as mock_client_class:
+ mock_client = MagicMock()
+ mock_client_class.return_value.__aenter__.return_value = mock_client
+
+ # Simulate 401 response from GitHub
+ mock_response = MagicMock()
+ mock_response.status_code = 401
+ mock_response.text = "Bad credentials"
+ mock_client.get.return_value = mock_response
+
+ result = await verifier.verify_token("invalid_token")
+ assert result is None
+
+ @pytest.mark.asyncio
+ async def test_verify_token_success(self):
+ """Test successful token verification."""
+ from unittest.mock import AsyncMock
+
+ verifier = GitHubTokenVerifier(required_scopes=["user"])
+
+ # Mock the httpx.AsyncClient directly
+ mock_client = AsyncMock()
+
+ # Mock successful user API response
+ user_response = MagicMock()
+ user_response.status_code = 200
+ user_response.json.return_value = {
+ "id": 12345,
+ "login": "testuser",
+ "name": "Test User",
+ "email": "test@example.com",
+ "avatar_url": "https://github.com/testuser.png",
+ }
+
+ # Mock successful scopes API response
+ scopes_response = MagicMock()
+ scopes_response.headers = {"x-oauth-scopes": "user,repo"}
+
+ # Set up the mock client to return our responses
+ mock_client.get.side_effect = [user_response, scopes_response]
+
+ # Patch the AsyncClient context manager
+ with patch(
+ "fastmcp.server.auth.providers.github.httpx.AsyncClient"
+ ) as mock_client_class:
+ mock_client_class.return_value.__aenter__.return_value = mock_client
+
+ result = await verifier.verify_token("valid_token")
+
+ assert result is not None
+ assert result.token == "valid_token"
+ assert result.client_id == "12345"
+ assert result.scopes == ["user", "repo"]
+ assert result.claims["login"] == "testuser"
+ assert result.claims["name"] == "Test User"
diff --git a/tests/server/auth/test_oauth_proxy.py b/tests/server/auth/test_oauth_proxy.py
new file mode 100644
index 000000000..eded58da1
--- /dev/null
+++ b/tests/server/auth/test_oauth_proxy.py
@@ -0,0 +1,548 @@
+"""Comprehensive tests for OAuth Proxy Provider functionality."""
+
+import time
+from unittest.mock import Mock
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+from mcp.server.auth.provider import AuthorizationParams
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
+
+from fastmcp.server.auth.auth import AccessToken
+from fastmcp.server.auth.providers.jwt import JWTVerifier
+from fastmcp.server.auth.proxy import OAuthProxy
+
+
+class TestOAuthProxyComprehensive:
+ """Comprehensive test suite for OAuthProxy provider functionality."""
+
+ @pytest.fixture
+ def jwt_verifier(self):
+ """Create a mock JWT verifier for testing."""
+ verifier = Mock(spec=JWTVerifier)
+ verifier.required_scopes = ["read", "write"]
+ verifier.verify_token = Mock(return_value=None)
+ return verifier
+
+ @pytest.fixture
+ def oauth_proxy(self, jwt_verifier):
+ """Create an OAuthProxy instance for testing."""
+ 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-client-id",
+ upstream_client_secret="test-client-secret",
+ token_verifier=jwt_verifier,
+ base_url="https://myserver.com",
+ redirect_path="/oauth/callback",
+ )
+
+ def test_initialization_with_string_urls(self, jwt_verifier):
+ """Test OAuthProxy initialization with string URLs (not AnyHttpUrl objects)."""
+ proxy = OAuthProxy(
+ upstream_authorization_endpoint="https://auth.example.com/authorize",
+ upstream_token_endpoint="https://auth.example.com/token",
+ upstream_client_id="client-123",
+ upstream_client_secret="secret-456",
+ token_verifier=jwt_verifier,
+ base_url="https://api.example.com", # String instead of AnyHttpUrl
+ issuer_url="https://issuer.example.com", # String
+ service_documentation_url="https://docs.example.com", # String
+ resource_server_url="https://resources.example.com", # String
+ )
+
+ # Should work fine and convert internally to AnyHttpUrl
+ assert str(proxy.base_url) == "https://api.example.com/"
+ assert str(proxy.issuer_url) == "https://issuer.example.com/"
+ assert str(proxy.service_documentation_url) == "https://docs.example.com/"
+ assert str(proxy.resource_server_url) == "https://resources.example.com/"
+
+ def test_initialization_with_all_parameters(self, jwt_verifier):
+ """Test OAuthProxy initialization with all optional parameters."""
+ proxy = OAuthProxy(
+ upstream_authorization_endpoint="https://auth.example.com/authorize",
+ upstream_token_endpoint="https://auth.example.com/token",
+ upstream_client_id="client-123",
+ upstream_client_secret="secret-456",
+ upstream_revocation_endpoint="https://auth.example.com/revoke",
+ token_verifier=jwt_verifier,
+ base_url="https://api.example.com",
+ redirect_path="/auth/callback",
+ issuer_url="https://issuer.example.com",
+ service_documentation_url="https://docs.example.com",
+ resource_server_url="https://resources.example.com",
+ )
+
+ # Verify all parameters are set correctly
+ assert (
+ proxy._upstream_authorization_endpoint
+ == "https://auth.example.com/authorize"
+ )
+ assert proxy._upstream_token_endpoint == "https://auth.example.com/token"
+ assert proxy._upstream_client_id == "client-123"
+ assert proxy._upstream_client_secret.get_secret_value() == "secret-456"
+ assert proxy._upstream_revocation_endpoint == "https://auth.example.com/revoke"
+ assert proxy._redirect_path == "/auth/callback"
+ assert str(proxy.issuer_url) == "https://issuer.example.com/"
+ assert str(proxy.service_documentation_url) == "https://docs.example.com/"
+ assert str(proxy.resource_server_url) == "https://resources.example.com/"
+
+ def test_redirect_path_normalization(self, jwt_verifier):
+ """Test that redirect_path is normalized to start with /."""
+ # Without leading slash
+ proxy1 = OAuthProxy(
+ upstream_authorization_endpoint="https://auth.com/authorize",
+ upstream_token_endpoint="https://auth.com/token",
+ upstream_client_id="client",
+ upstream_client_secret="secret",
+ token_verifier=jwt_verifier,
+ base_url="https://server.com",
+ redirect_path="oauth/callback",
+ )
+ assert proxy1._redirect_path == "/oauth/callback"
+
+ # With leading slash
+ proxy2 = OAuthProxy(
+ upstream_authorization_endpoint="https://auth.com/authorize",
+ upstream_token_endpoint="https://auth.com/token",
+ upstream_client_id="client",
+ upstream_client_secret="secret",
+ token_verifier=jwt_verifier,
+ base_url="https://server.com",
+ redirect_path="/oauth/callback",
+ )
+ assert proxy2._redirect_path == "/oauth/callback"
+
+ def test_dcr_always_enabled(self, jwt_verifier):
+ """Test that DCR is always enabled for OAuth Proxy."""
+ proxy = OAuthProxy(
+ upstream_authorization_endpoint="https://auth.com/authorize",
+ upstream_token_endpoint="https://auth.com/token",
+ upstream_client_id="client",
+ upstream_client_secret="secret",
+ token_verifier=jwt_verifier,
+ base_url="https://server.com",
+ )
+
+ assert proxy.client_registration_options is not None
+ assert proxy.client_registration_options.enabled is True
+
+ def test_revocation_enabled_with_endpoint(self, jwt_verifier):
+ """Test that revocation is enabled when upstream endpoint is provided."""
+ proxy = OAuthProxy(
+ upstream_authorization_endpoint="https://auth.com/authorize",
+ upstream_token_endpoint="https://auth.com/token",
+ upstream_client_id="client",
+ upstream_client_secret="secret",
+ upstream_revocation_endpoint="https://auth.com/revoke",
+ token_verifier=jwt_verifier,
+ base_url="https://server.com",
+ )
+
+ assert proxy.revocation_options is not None
+ assert proxy.revocation_options.enabled is True
+ assert proxy._upstream_revocation_endpoint == "https://auth.com/revoke"
+
+ def test_revocation_disabled_without_endpoint(self, jwt_verifier):
+ """Test that revocation is disabled when no upstream endpoint is provided."""
+ proxy = OAuthProxy(
+ upstream_authorization_endpoint="https://auth.com/authorize",
+ upstream_token_endpoint="https://auth.com/token",
+ upstream_client_id="client",
+ upstream_client_secret="secret",
+ token_verifier=jwt_verifier,
+ base_url="https://server.com",
+ )
+
+ assert proxy.revocation_options is None
+ assert proxy._upstream_revocation_endpoint is None
+
+ @pytest.mark.asyncio
+ async def test_register_client(self, oauth_proxy):
+ """Test client registration always uses upstream credentials."""
+ client_info = OAuthClientInformationFull(
+ client_id="original-client-id",
+ client_secret="original-secret",
+ redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+ grant_types=["authorization_code"],
+ token_endpoint_auth_method="client_secret_post",
+ )
+
+ await oauth_proxy.register_client(client_info)
+
+ # Verify client was modified to use upstream credentials
+ assert client_info.client_id == "test-client-id"
+ assert client_info.client_secret == "test-client-secret"
+ assert client_info.token_endpoint_auth_method == "none"
+ assert "authorization_code" in client_info.grant_types
+ # refresh_token is only added if grant_types was empty
+
+ # Verify client was stored
+ stored_client = oauth_proxy._clients.get("test-client-id")
+ assert stored_client is not None
+ assert stored_client.client_id == "test-client-id"
+
+ @pytest.mark.asyncio
+ async def test_register_client_empty_grant_types(self, oauth_proxy):
+ """Test client registration adds grant types when empty."""
+ client_info = OAuthClientInformationFull(
+ client_id="original-client-id",
+ client_secret="original-secret",
+ redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+ grant_types=[], # Empty grant types list
+ )
+
+ await oauth_proxy.register_client(client_info)
+
+ # Should add both authorization_code and refresh_token
+ assert client_info.grant_types == ["authorization_code", "refresh_token"]
+
+ @pytest.mark.asyncio
+ async def test_get_client_existing(self, oauth_proxy):
+ """Test getting an existing registered client."""
+ # Register a client first
+ client_info = OAuthClientInformationFull(
+ client_id="test-id",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+ )
+ await oauth_proxy.register_client(client_info)
+
+ # Get the client
+ retrieved = await oauth_proxy.get_client("test-client-id")
+ assert retrieved is not None
+ assert retrieved.client_id == "test-client-id"
+
+ @pytest.mark.asyncio
+ async def test_get_client_temporary(self, oauth_proxy):
+ """Test getting a temporary client for unregistered client ID."""
+ # Get a client that hasn't been registered
+ temp_client = await oauth_proxy.get_client("unknown-client-id")
+
+ assert temp_client is not None
+ assert temp_client.client_id == "unknown-client-id"
+ assert temp_client.client_secret is None
+ assert temp_client.token_endpoint_auth_method == "none"
+ assert len(temp_client.redirect_uris) >= 1
+ # ProxyDCRClient uses a placeholder URL but accepts any localhost URI
+ assert str(temp_client.redirect_uris[0]) == "http://localhost/"
+
+ # Test that it accepts any localhost redirect URI
+ from pydantic import AnyUrl
+
+ test_uri = temp_client.validate_redirect_uri(
+ AnyUrl("http://localhost:55454/callback")
+ )
+ assert str(test_uri) == "http://localhost:55454/callback"
+
+ @pytest.mark.asyncio
+ async def test_authorize_creates_transaction(self, oauth_proxy):
+ """Test that authorize creates a transaction and returns upstream URL."""
+ client = OAuthClientInformationFull(
+ client_id="test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:54321/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="client-state-123",
+ code_challenge="challenge-abc",
+ scopes=["read", "write"],
+ )
+
+ # Call authorize
+ redirect_url = await oauth_proxy.authorize(client, params)
+
+ # Parse the redirect URL
+ parsed = urlparse(redirect_url)
+ query_params = parse_qs(parsed.query)
+
+ # Verify it's redirecting to upstream
+ assert parsed.scheme == "https"
+ assert parsed.netloc == "github.com"
+ assert parsed.path == "/login/oauth/authorize"
+
+ # Verify query parameters
+ assert query_params["response_type"] == ["code"]
+ assert query_params["client_id"] == ["test-client-id"]
+ assert query_params["redirect_uri"] == ["https://myserver.com/oauth/callback"]
+ assert "state" in query_params # This should be the transaction ID
+ assert query_params["scope"] == ["read write"]
+
+ # Verify transaction was stored
+ txn_id = query_params["state"][0]
+ transaction = oauth_proxy._oauth_transactions.get(txn_id)
+ assert transaction is not None
+ 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["code_challenge_method"] == "S256"
+ assert transaction["scopes"] == ["read", "write"]
+
+ @pytest.mark.asyncio
+ async def test_authorize_without_scopes(self, oauth_proxy):
+ """Test authorize without scopes uses required scopes from verifier."""
+ client = OAuthClientInformationFull(
+ client_id="test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:54321/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="client-state",
+ code_challenge="challenge",
+ scopes=[], # Empty scopes to test fallback
+ )
+
+ redirect_url = await oauth_proxy.authorize(client, params)
+
+ parsed = urlparse(redirect_url)
+ query_params = parse_qs(parsed.query)
+
+ # Should use required_scopes from token_verifier
+ assert query_params["scope"] == ["read write"]
+
+ @pytest.mark.asyncio
+ async def test_authorize_google_minimal_scope(self, jwt_verifier):
+ """Test that Google OAuth gets minimal scope when none specified."""
+ # Create proxy with Google endpoints
+ proxy = OAuthProxy(
+ upstream_authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth",
+ upstream_token_endpoint="https://oauth2.googleapis.com/token",
+ upstream_client_id="google-client",
+ upstream_client_secret="google-secret",
+ token_verifier=Mock(required_scopes=None), # No required scopes
+ base_url="https://myserver.com",
+ )
+
+ client = OAuthClientInformationFull(
+ client_id="test-client",
+ client_secret="test-secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+
+ params = AuthorizationParams(
+ redirect_uri=AnyUrl("http://localhost:54321/callback"),
+ redirect_uri_provided_explicitly=True,
+ state="state",
+ code_challenge="challenge",
+ scopes=[], # Empty scopes to test Google fallback
+ )
+
+ redirect_url = await proxy.authorize(client, params)
+
+ parsed = urlparse(redirect_url)
+ query_params = parse_qs(parsed.query)
+
+ # Should add minimal scope for Google
+ assert query_params["scope"] == ["openid"]
+
+ @pytest.mark.asyncio
+ async def test_load_authorization_code_valid(self, oauth_proxy):
+ """Test loading a valid authorization code."""
+ # Store a client code
+ code = "test-auth-code"
+ oauth_proxy._client_codes[code] = {
+ "client_id": "test-client-id",
+ "redirect_uri": "http://localhost:54321/callback",
+ "code_challenge": "challenge-123",
+ "scopes": ["read", "write"],
+ "expires_at": time.time() + 300, # 5 minutes from now
+ "idp_tokens": {"access_token": "token-123"},
+ }
+
+ client = OAuthClientInformationFull(
+ client_id="test-client-id",
+ client_secret="secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+
+ # Load the code
+ auth_code = await oauth_proxy.load_authorization_code(client, code)
+
+ assert auth_code is not None
+ assert auth_code.code == code
+ assert auth_code.client_id == "test-client-id"
+ assert str(auth_code.redirect_uri) == "http://localhost:54321/callback"
+ assert auth_code.code_challenge == "challenge-123"
+ assert auth_code.scopes == ["read", "write"]
+
+ @pytest.mark.asyncio
+ async def test_load_authorization_code_expired(self, oauth_proxy):
+ """Test loading an expired authorization code returns None."""
+ code = "expired-code"
+ oauth_proxy._client_codes[code] = {
+ "client_id": "test-client-id",
+ "redirect_uri": "http://localhost:54321/callback",
+ "expires_at": time.time() - 60, # Expired 1 minute ago
+ }
+
+ client = OAuthClientInformationFull(
+ client_id="test-client-id",
+ client_secret="secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+
+ auth_code = await oauth_proxy.load_authorization_code(client, code)
+ assert auth_code is None
+ # Code should be cleaned up
+ assert code not in oauth_proxy._client_codes
+
+ @pytest.mark.asyncio
+ async def test_load_authorization_code_wrong_client(self, oauth_proxy):
+ """Test loading authorization code with wrong client ID returns None."""
+ code = "test-code"
+ oauth_proxy._client_codes[code] = {
+ "client_id": "correct-client-id",
+ "redirect_uri": "http://localhost:54321/callback",
+ "expires_at": time.time() + 300,
+ }
+
+ wrong_client = OAuthClientInformationFull(
+ client_id="wrong-client-id",
+ client_secret="secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+
+ auth_code = await oauth_proxy.load_authorization_code(wrong_client, code)
+ assert auth_code is None
+
+ @pytest.mark.asyncio
+ async def test_load_access_token_delegates_to_verifier(
+ self, oauth_proxy, jwt_verifier
+ ):
+ """Test that load_access_token delegates to the token verifier."""
+ token = "test-access-token"
+ expected_result = AccessToken(
+ token=token,
+ client_id="test-client",
+ scopes=["read"],
+ expires_at=int(time.time() + 3600),
+ )
+
+ # Mock the async method properly
+ async def mock_verify(token):
+ return expected_result
+
+ jwt_verifier.verify_token = mock_verify
+
+ result = await oauth_proxy.load_access_token(token)
+
+ assert result == expected_result
+ # Can't assert on the mock function call in this case
+
+ def test_get_routes_includes_callback(self, oauth_proxy):
+ """Test that get_routes includes the OAuth callback route."""
+ routes = oauth_proxy.get_routes()
+
+ # Find the callback route
+ callback_routes = [
+ r for r in routes if hasattr(r, "path") and r.path == "/oauth/callback"
+ ]
+
+ assert len(callback_routes) == 1
+ callback_route = callback_routes[0]
+ assert "GET" in callback_route.methods
+ assert callback_route.endpoint == oauth_proxy._handle_idp_callback
+
+ def test_get_routes_preserves_standard_routes(self, oauth_proxy):
+ """Test that get_routes preserves standard OAuth routes."""
+ routes = oauth_proxy.get_routes()
+
+ # Should have standard OAuth routes
+ paths = [r.path for r in routes if hasattr(r, "path")]
+
+ # Standard OAuth endpoints should be present
+ assert "/authorize" in paths
+ assert "/token" in paths
+ assert "/.well-known/oauth-authorization-server" in paths
+
+ # Plus our custom callback
+ assert "/oauth/callback" in paths
+
+ @pytest.mark.asyncio
+ async def test_revoke_token_access_token(self, oauth_proxy):
+ """Test revoking an access token cleans up local storage."""
+ # Store tokens
+ access_token = "access-123"
+ refresh_token = "refresh-456"
+
+ oauth_proxy._access_tokens[access_token] = AccessToken(
+ token=access_token,
+ client_id="client",
+ scopes=[],
+ expires_at=int(time.time() + 3600),
+ )
+ oauth_proxy._refresh_tokens[refresh_token] = Mock(token=refresh_token)
+ oauth_proxy._access_to_refresh[access_token] = refresh_token
+ oauth_proxy._refresh_to_access[refresh_token] = access_token
+
+ # Revoke access token
+ await oauth_proxy.revoke_token(oauth_proxy._access_tokens[access_token])
+
+ # Verify cleanup
+ assert access_token not in oauth_proxy._access_tokens
+ assert refresh_token not in oauth_proxy._refresh_tokens
+ assert access_token not in oauth_proxy._access_to_refresh
+ assert refresh_token not in oauth_proxy._refresh_to_access
+
+ @pytest.mark.asyncio
+ async def test_exchange_authorization_code_stores_tokens(self, oauth_proxy):
+ """Test that exchange_authorization_code stores tokens locally."""
+ from mcp.server.auth.provider import AuthorizationCode
+
+ # Set up client code with IdP tokens
+ code = "client-code-123"
+ idp_tokens = {
+ "access_token": "idp-access-token",
+ "refresh_token": "idp-refresh-token",
+ "expires_in": 3600,
+ "token_type": "Bearer",
+ }
+
+ oauth_proxy._client_codes[code] = {
+ "client_id": "test-client",
+ "redirect_uri": "http://localhost:54321/callback",
+ "scopes": ["read", "write"],
+ "idp_tokens": idp_tokens,
+ "expires_at": time.time() + 300,
+ }
+
+ client = OAuthClientInformationFull(
+ client_id="test-client",
+ client_secret="secret",
+ redirect_uris=[AnyUrl("http://localhost:54321/callback")],
+ )
+
+ auth_code = AuthorizationCode(
+ code=code,
+ client_id="test-client",
+ redirect_uri=AnyUrl("http://localhost:54321/callback"),
+ redirect_uri_provided_explicitly=True,
+ scopes=["read", "write"],
+ expires_at=time.time() + 300,
+ code_challenge="test-challenge",
+ )
+
+ # Exchange the code
+ result = await oauth_proxy.exchange_authorization_code(client, auth_code)
+
+ # Verify result
+ assert result.access_token == "idp-access-token"
+ assert result.refresh_token == "idp-refresh-token"
+ assert result.expires_in == 3600
+
+ # Verify tokens were stored locally
+ assert "idp-access-token" in oauth_proxy._access_tokens
+ assert "idp-refresh-token" in oauth_proxy._refresh_tokens
+ assert oauth_proxy._access_to_refresh["idp-access-token"] == "idp-refresh-token"
+ assert oauth_proxy._refresh_to_access["idp-refresh-token"] == "idp-access-token"
+
+ # Verify code was cleaned up
+ assert code not in oauth_proxy._client_codes