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

Add OAuth proxy that allows authentication with social IDPs without DCR support (#1434)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-08-18 13:39:58 -04:00 • committed by GitHub
commit ec015de3b0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 3818 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/run-tests.yml vendored
View file

@ -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 }}

24
docs/docs.json
View file

@ -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"]
}
]
},

205
docs/integrations/github.mdx Normal file
View file

@ -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"
<VersionBadge version="2.12.0" />
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:
<Steps>
<Step title="Navigate to OAuth Apps">
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.
</Step>
<Step title="Configure Your OAuth App">
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`)
<Warning>
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.
</Warning>
<Tip>
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.
</Tip>
</Step>
<Step title="Save Your Credentials">
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
<Tip>
Store these credentials securely. Never commit them to version control. Use environment variables or a secrets manager in production.
</Tip>
</Step>
</Steps>
### 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
<Info>
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.
</Info>
## Environment Variables
For production deployments, use environment variables instead of hardcoding credentials.
<Info>
To use the registered GitHub provider, you must set `FASTMCP_SERVER_AUTH=GITHUB`. Learn more about [registered providers](/servers/auth/authentication#registered-providers).
</Info>
### Provider Selection
<ParamField path="FASTMCP_SERVER_AUTH" default="Not set" required>
Set to `GITHUB` to use the registered GitHubProvider with default configuration.
</ParamField>
### GitHub-Specific Configuration
<Card>
<ParamField path="FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID" required>
Your GitHub OAuth App Client ID (e.g., `Ov23liAbcDefGhiJkLmN`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET" required>
Your GitHub OAuth App Client Secret
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GITHUB_BASE_URL" default="http://localhost:8000">
Public URL of your FastMCP server for OAuth callbacks
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GITHUB_REDIRECT_PATH" default="/oauth/callback">
Redirect path configured in your GitHub OAuth App
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GITHUB_REQUIRED_SCOPES" default='["user"]'>
Comma-separated list of required GitHub scopes (e.g., `user,repo`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GITHUB_TIMEOUT_SECONDS" default="10">
HTTP request timeout for GitHub API calls
</ParamField>
</Card>
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
```

215
docs/integrations/google.mdx Normal file
View file

@ -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"
<VersionBadge version="2.12.0" />
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:
<Steps>
<Step title="Navigate to OAuth Consent Screen">
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.
</Step>
<Step title="Create OAuth 2.0 Client ID">
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`)
<Warning>
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.
</Warning>
<Tip>
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.
</Tip>
</Step>
<Step title="Save Your Credentials">
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.
<Tip>
Store these credentials securely. Never commit them to version control. Use environment variables or a secrets manager in production.
</Tip>
</Step>
</Steps>
### 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
<Info>
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.
</Info>
## Environment Variables
For production deployments, use environment variables instead of hardcoding credentials.
<Info>
To use the registered Google provider, you must set `FASTMCP_SERVER_AUTH=GOOGLE`. Learn more about [registered providers](/servers/auth/authentication#registered-providers).
</Info>
### Provider Selection
<ParamField path="FASTMCP_SERVER_AUTH" default="Not set" required>
Set to `GOOGLE` to use the registered GoogleProvider with default configuration.
</ParamField>
### Google-Specific Configuration
<Card>
<ParamField path="FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID" required>
Your Google OAuth 2.0 Client ID (e.g., `123456789.apps.googleusercontent.com`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET" required>
Your Google OAuth 2.0 Client Secret (e.g., `GOCSPX-abc123...`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GOOGLE_BASE_URL" default="http://localhost:8000">
Public URL of your FastMCP server for OAuth callbacks
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GOOGLE_REDIRECT_PATH" default="/oauth/callback">
Redirect path configured in your Google OAuth Client
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GOOGLE_REQUIRED_SCOPES" default="[]">
Comma-separated list of required Google scopes (e.g., `openid`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_GOOGLE_TIMEOUT_SECONDS" default="10">
HTTP request timeout for Google API calls
</ParamField>
</Card>
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}"
```

82
docs/servers/auth/authentication.mdx
View file

@ -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
<VersionBadge version="2.12.0" />
`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:
<ParamField path="FASTMCP_SERVER_AUTH" type="string">
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
</ParamField>
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.

300
docs/servers/auth/oauth-proxy.mdx Normal file
View file

@ -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"
<VersionBadge version="2.12.0" />
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.
<Tip>
**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.
</Tip>
## 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<br/>(localhost:random)
participant Proxy as FastMCP OAuth Proxy<br/>(server:8000)
participant Provider as OAuth Provider<br/>(GitHub, etc.)
Note over Client, Proxy: Dynamic Registration (Local)
Client->>Proxy: 1. POST /register<br/>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<br/>redirect_uri=localhost:54321/callback
Note over Proxy: Store transaction with client callback
Proxy->>Provider: 4. Redirect to provider<br/>redirect_uri=server:8000/oauth/callback
Note over Provider, Proxy: Provider Callback
Provider->>Proxy: 5. GET /oauth/callback<br/>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<br/>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:
<Card icon="code" title="OAuthProxy Constructor Parameters">
<ParamField body="upstream_authorization_endpoint" type="str" required>
URL of your OAuth provider's authorization endpoint (e.g., `https://github.com/login/oauth/authorize`)
</ParamField>
<ParamField body="upstream_token_endpoint" type="str" required>
URL of your OAuth provider's token endpoint (e.g., `https://github.com/login/oauth/access_token`)
</ParamField>
<ParamField body="upstream_client_id" type="str" required>
Client ID from your registered OAuth application
</ParamField>
<ParamField body="upstream_client_secret" type="str" required>
Client secret from your registered OAuth application
</ParamField>
<ParamField body="token_verifier" type="TokenVerifier" required>
A [`TokenVerifier`](/servers/auth/token-verification) instance to validate the provider's tokens
</ParamField>
<ParamField body="base_url" type="AnyHttpUrl | str" required>
Public URL of your FastMCP server (e.g., `https://your-server.com`)
</ParamField>
<ParamField body="redirect_path" type="str" default="/oauth/callback">
Path for OAuth callbacks. Must match the redirect URI configured in your OAuth application
</ParamField>
<ParamField body="upstream_revocation_endpoint" type="str | None">
Optional URL of provider's token revocation endpoint
</ParamField>
<ParamField body="issuer_url" type="AnyHttpUrl | str | None">
Issuer URL for OAuth metadata (defaults to base_url)
</ParamField>
<ParamField body="service_documentation_url" type="AnyHttpUrl | str | None">
Optional URL to your service documentation
</ParamField>
<ParamField body="resource_server_url" type="AnyHttpUrl | str | None">
Resource server URL (defaults to base_url)
</ParamField>
</Card>
```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
<Tip>
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.
</Tip>
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")
```

22
docs/servers/auth/remote-oauth.mdx
View file

@ -10,12 +10,30 @@ import { VersionBadge } from "/snippets/version-badge.mdx"
<VersionBadge version="2.11.0" />
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.
<Tip>
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.
</Tip>
## 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.

31
examples/auth/github_oauth/README.md Normal file
View file

@ -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.

32
examples/auth/github_oauth/client.py Normal file
View file

@ -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())

35
examples/auth/github_oauth/server.py Normal file
View file

@ -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)

34
examples/auth/google_oauth/README.md Normal file
View file

@ -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.

32
examples/auth/google_oauth/client.py Normal file
View file

@ -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())

37
examples/auth/google_oauth/server.py Normal file
View file

@ -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)

267
src/fastmcp/server/auth/providers/github.py Normal file
View file

@ -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,
)

286
src/fastmcp/server/auth/providers/google.py Normal file
View file

@ -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,
)

1059
src/fastmcp/server/auth/proxy.py Normal file
View file

File diff suppressed because it is too large Load diff

9
src/fastmcp/utilities/tests.py
View file

@ -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}

21
test_github_oauth.py Normal file
View file

@ -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)

22
test_google_oauth.py Normal file
View file

@ -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)

0
tests/integration_tests/auth/__init__.py Normal file
View file

356
tests/integration_tests/auth/test_github_provider_integration.py Normal file
View file

@ -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

227
tests/server/auth/providers/test_github.py Normal file
View file

@ -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"

548
tests/server/auth/test_oauth_proxy.py Normal file
View file

@ -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