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

Adds PropelAuth as an AuthProvider

Browse source 
Adds the PropelAuthProvider which delegates to the IntrospectionTokenVerifier
and optionally does an additional resource check.
Adds an example server and client which makes an authenticated request
and gets information from the token.
Updates the documentation (but only for v3 as this isn't in v2).
This commit is contained in:
Andrew Israel 2026-02-26 23:50:45 -08:00 • committed by Jeremiah Lowin
commit 5fb72c7200
7 changed files with 888 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/docs.json
View file

@ -277,6 +277,7 @@
"integrations/github",
"integrations/google",
"integrations/oci",
"integrations/propelauth",
"integrations/scalekit",
"integrations/supabase",
"integrations/workos",

164
docs/integrations/propelauth.mdx Normal file
View file

@ -0,0 +1,164 @@
---
title: PropelAuth 🤝 FastMCP
sidebarTitle: PropelAuth
description: Secure your FastMCP server with PropelAuth
icon: shield-check
---
import { VersionBadge } from "/snippets/version-badge.mdx";
<VersionBadge version="3.0.3" />
This guide shows you how to secure your FastMCP server using [**PropelAuth**](https://www.propelauth.com), a complete authentication and user management solution. This integration uses the [**Remote OAuth**](/servers/auth/remote-oauth) pattern, where PropelAuth handles user login, consent management, and your FastMCP server validates the tokens.
## Configuration
### Prerequisites
Before you begin, you will need:
1. A [PropelAuth](https://www.propelauth.com) account
2. Your FastMCP server's base URL (can be localhost for development, e.g., `http://localhost:8000`)
### Step 1: Configure PropelAuth
<Steps>
<Step title="Enable MCP Authentication">
Navigate to the **MCP** section in your PropelAuth dashboard, click **Enable MCP**, and choose which environments to enable it for (Test, Staging, Prod).
</Step>
<Step title="Configure Allowed MCP Clients">
Under **MCP > Allowed MCP Clients**, add redirect URIs for each MCP client you want to allow. PropelAuth provides templates for popular clients like Claude, Cursor, and ChatGPT.
</Step>
<Step title="Configure Scopes">
Under **MCP > Scopes**, define the permissions available to MCP clients (e.g., `read:user_data`).
</Step>
<Step title="Choose How Users Create OAuth Clients">
Under **MCP > Settings > How Do Users Create OAuth Clients?**, you can optionally enable:
- **Dynamic Client Registration** — clients self-register automatically via the DCR protocol
- **Manually via Hosted Pages** — PropelAuth creates a UI for your users to register OAuth clients
You can enable neither, one, or both. If you enable neither, you'll manage OAuth client creation yourself.
</Step>
<Step title="Generate Introspection Credentials">
Go to **MCP > Request Validation** and click **Create Credentials**. Note the **Client ID** and **Client Secret** - you'll need these to validate tokens.
</Step>
<Step title="Note Your Auth URL">
Find your Auth URL in the **Backend Integration** section of the dashboard (e.g., `https://auth.yourdomain.com`).
</Step>
</Steps>
For more details, see the [PropelAuth MCP documentation](https://docs.propelauth.com/mcp-authentication/overview).
### Step 2: Environment Setup
Create a `.env` file with your PropelAuth configuration:
```bash
PROPELAUTH_AUTH_URL=https://auth.yourdomain.com # From Backend Integration page
PROPELAUTH_INTROSPECTION_CLIENT_ID=your-client-id # From MCP > Request Validation
PROPELAUTH_INTROSPECTION_CLIENT_SECRET=your-client-secret # From MCP > Request Validation
SERVER_URL=http://localhost:8000 # Your server's base URL
```
### Step 3: FastMCP Configuration
Create your FastMCP server file and use the PropelAuthProvider to handle all the OAuth integration automatically:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.propelauth import PropelAuthProvider
auth_provider = PropelAuthProvider(
auth_url=os.environ["PROPELAUTH_AUTH_URL"],
introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"],
introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"],
base_url=os.environ["SERVER_URL"],
required_scopes=["read:user_data"], # Optional scope enforcement
)
mcp = FastMCP(name="My PropelAuth Protected Server", auth=auth_provider)
```
## Testing
With your `.env` loaded, start the server:
```bash
fastmcp run server.py --transport http --port 8000
```
Then use a FastMCP client to verify authentication works:
```python
from fastmcp import Client
import asyncio
async def main():
async with Client("http://localhost:8000/mcp", auth="oauth") as client:
assert await client.ping()
if __name__ == "__main__":
asyncio.run(main())
```
## Accessing User Information
You can use `get_access_token()` inside your tools to identify the authenticated user:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.propelauth import PropelAuthProvider
from fastmcp.server.dependencies import get_access_token
auth = PropelAuthProvider(
auth_url=os.environ["PROPELAUTH_AUTH_URL"],
introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"],
introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"],
base_url=os.environ["SERVER_URL"],
required_scopes=["read:user_data"],
)
mcp = FastMCP(name="My PropelAuth Protected Server", auth=auth)
@mcp.tool
def whoami() -> dict:
"""Return the authenticated user's ID."""
token = get_access_token()
if token is None:
return {"error": "Not authenticated"}
user_id = token.claims.get("sub")
return {"user_id": user_id}
```
## Advanced Configuration
The `PropelAuthProvider` supports optional overrides for token introspection behavior, including caching and request timeouts:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.propelauth import PropelAuthProvider
auth = PropelAuthProvider(
auth_url=os.environ["PROPELAUTH_AUTH_URL"],
introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"],
introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"],
base_url=os.environ.get("BASE_URL", "https://your-server.com"),
required_scopes=["read:user_data"],
resource="https://your-server.com/mcp", # Restrict to tokens intended for this server (RFC 8707)
token_introspection_overrides={
"cache_ttl_seconds": 300, # Cache introspection results for 5 minutes
"max_cache_size": 1000, # Maximum cached tokens
"timeout_seconds": 15, # HTTP request timeout
},
)
mcp = FastMCP(name="My PropelAuth Protected Server", auth=auth)
```

67
examples/auth/propelauth_oauth/README.md Normal file
View file

@ -0,0 +1,67 @@
# PropelAuth OAuth Example
Demonstrates FastMCP server protection with PropelAuth OAuth.
## Setup
### 1. Configure MCP Authentication in PropelAuth
**Create a PropelAuth Account**:
- Go to [PropelAuth Dashboard](https://www.propelauth.com)
- Navigate to the **MCP** section and click **Enable MCP**
**Configure Allowed MCP Clients**:
- Under **MCP > Allowed MCP Clients**, add redirect URIs for each MCP client you want to allow
- PropelAuth provides templates for popular clients like Claude, Cursor, and ChatGPT
**Configure Scopes**:
- Under **MCP > Scopes**, define the permissions available to MCP clients (e.g., `read:user_data`)
**Generate Introspection Credentials**:
- Go to **MCP > Request Validation** and click **Create Credentials**
- Note the **Client ID** and **Client Secret**
**Note Your Auth URL**:
- Find your Auth URL in the **Backend Integration** section (e.g., `https://auth.yourdomain.com`)
Create a `.env` file:
```bash
# Required PropelAuth credentials
PROPELAUTH_AUTH_URL=https://auth.yourdomain.com
PROPELAUTH_INTROSPECTION_CLIENT_ID=your-client-id
PROPELAUTH_INTROSPECTION_CLIENT_SECRET=your-client-secret
BASE_URL=http://localhost:8000/
# Optional: additional scopes tokens must include (comma-separated)
# PROPELAUTH_REQUIRED_SCOPES=read:user_data
```
### 2. Run the Example
Start the server:
```bash
# From this directory
uv run python server.py
```
The server will start on `http://localhost:8000/mcp` with PropelAuth OAuth authentication enabled.
Test with client:
```bash
uv run python client.py
```
The `client.py` will:
1. Attempt to connect to the server
2. Detect that OAuth authentication is required
3. Open a browser for PropelAuth authentication
4. Complete the OAuth flow and connect to the server
5. Demonstrate calling authenticated tools (echo and whoami)

43
examples/auth/propelauth_oauth/client.py Normal file
View file

@ -0,0 +1,43 @@
"""OAuth client example for connecting to PropelAuth-protected FastMCP servers.
This example demonstrates how to connect to a PropelAuth 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 with PropelAuth!")
tools = await client.list_tools()
print(f"🔧 Available tools ({len(tools)}):")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
# Test calling a tool
result = await client.call_tool(
"echo", {"message": "Hello from PropelAuth!"}
)
print(f"🎯 Echo result: {result}")
# Test calling whoami tool
whoami = await client.call_tool("whoami", {})
print(f"👤 Who am I: {whoami}")
except Exception as e:
print(f"❌ Authentication failed: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())

54
examples/auth/propelauth_oauth/server.py Normal file
View file

@ -0,0 +1,54 @@
"""PropelAuth OAuth server example for FastMCP.
This example demonstrates how to protect a FastMCP server with PropelAuth OAuth.
Required environment variables:
- PROPELAUTH_AUTH_URL: Your PropelAuth Auth URL (from Backend Integration page)
- PROPELAUTH_INTROSPECTION_CLIENT_ID: Introspection Client ID (from MCP > Request Validation)
- PROPELAUTH_INTROSPECTION_CLIENT_SECRET: Introspection Client Secret (from MCP > Request Validation)
Optional:
- PROPELAUTH_REQUIRED_SCOPES: Comma-separated scopes tokens must include
- BASE_URL: Public URL where the FastMCP server is exposed (defaults to `http://localhost:8000/`)
To run:
python server.py
"""
import os
from dotenv import load_dotenv
from fastmcp import FastMCP
from fastmcp.server.auth.providers.propelauth import PropelAuthProvider
from fastmcp.server.dependencies import get_access_token
load_dotenv()
auth = PropelAuthProvider(
auth_url=os.environ["PROPELAUTH_AUTH_URL"],
introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"],
introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"],
base_url=os.getenv("BASE_URL", "http://localhost:8000/"),
)
mcp = FastMCP("PropelAuth OAuth Example Server", auth=auth)
@mcp.tool
def echo(message: str) -> str:
"""Echo the provided message."""
return message
@mcp.tool
def whoami() -> dict:
"""Return the authenticated user's ID."""
token = get_access_token()
if token is None:
return {"error": "Not authenticated"}
return {"user_id": token.claims.get("sub")}
if __name__ == "__main__":
mcp.run(transport="http", port=8000)

223
src/fastmcp/server/auth/providers/propelauth.py Normal file
View file

@ -0,0 +1,223 @@
"""PropelAuth authentication provider for FastMCP.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.auth.providers.propelauth import PropelAuthProvider
auth = PropelAuthProvider(
auth_url="https://auth.yourdomain.com",
introspection_client_id="your-client-id",
introspection_client_secret="your-client-secret",
base_url="https://your-fastmcp-server.com",
required_scopes=["read:user_data"],
)
mcp = FastMCP("My App", auth=auth)
```
"""
from __future__ import annotations
from typing import TypedDict
import httpx
from pydantic import AnyHttpUrl, SecretStr
from starlette.responses import JSONResponse
from starlette.routing import Route
from fastmcp.server.auth import AccessToken, RemoteAuthProvider
from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class PropelAuthTokenIntrospectionOverrides(TypedDict, total=False):
timeout_seconds: int
cache_ttl_seconds: int | None
max_cache_size: int | None
http_client: httpx.AsyncClient | None
class PropelAuthProvider(RemoteAuthProvider):
"""PropelAuth resource server provider using OAuth 2.1 token introspection.
This provider validates access tokens via PropelAuth's introspection endpoint
and forwards authorization server metadata for OAuth discovery.
Setup:
1. Enable MCP authentication in the PropelAuth Dashboard
2. Configure scopes on the MCP page
3. Select which redirect URIs to enable by picking which clients you support
4. Generate introspection credentials (Client ID + Client Secret)
For detailed setup instructions, see:
https://docs.propelauth.com/mcp-authentication/overview
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.auth.providers.propelauth import PropelAuthProvider
auth = PropelAuthProvider(
auth_url="https://auth.yourdomain.com",
introspection_client_id="your-client-id",
introspection_client_secret="your-client-secret",
base_url="https://your-fastmcp-server.com",
required_scopes=["read:user_data"],
)
mcp = FastMCP("My App", auth=auth)
```
"""
def __init__(
self,
*,
auth_url: AnyHttpUrl | str,
introspection_client_id: str,
introspection_client_secret: str | SecretStr,
base_url: AnyHttpUrl | str,
required_scopes: list[str] | None = None,
resource: AnyHttpUrl | str | None = None,
token_introspection_overrides: (
PropelAuthTokenIntrospectionOverrides | None
) = None,
):
"""Initialize PropelAuth provider.
Args:
auth_url: Your PropelAuth Auth URL (from the Backend Integration page)
introspection_client_id: Introspection Client ID from the PropelAuth Dashboard
introspection_client_secret: Introspection Client Secret from the PropelAuth Dashboard
base_url: Public URL of this FastMCP server
required_scopes: Optional list of scopes that must be present in tokens
resource: Optional resource URI (RFC 8707) identifying this MCP server.
Use this when multiple MCP servers share the same PropelAuth
authorization server (e.g. ``resource="https://api.example.com/mcp"``),
so only tokens intended for this MCP server are accepted.
token_introspection_overrides: Optional overrides for the underlying
IntrospectionTokenVerifier (timeout, caching, http_client)
"""
normalized_auth_url = str(auth_url).rstrip("/")
introspection_url = f"{normalized_auth_url}/oauth/2.1/introspect"
authorization_server_url = AnyHttpUrl(f"{normalized_auth_url}/oauth/2.1")
if resource is None:
self._resource = None
logger.debug(
"PropelAuthProvider: no resource configured, audience checking disabled"
)
else:
self._resource = str(resource)
token_verifier = self._create_token_verifier(
introspection_url=introspection_url,
client_id=introspection_client_id,
client_secret=introspection_client_secret,
required_scopes=required_scopes,
introspection_overrides=token_introspection_overrides,
)
self._normalized_auth_url = normalized_auth_url
super().__init__(
token_verifier=token_verifier,
authorization_servers=[authorization_server_url],
base_url=base_url,
)
def get_routes(
self,
mcp_path: str | None = None,
) -> list[Route]:
"""Get routes for this provider.
Includes the standard routes from the RemoteAuthProvider (protected resource metadata routes (RFC 9728)),
and creates an authorization server metadata route that forwards to PropelAuth's route
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
This is used to advertise the resource URL in metadata.
"""
routes = super().get_routes(mcp_path)
async def oauth_authorization_server_metadata(request):
"""Forward PropelAuth OAuth authorization server metadata"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self._normalized_auth_url}/.well-known/oauth-authorization-server/oauth/2.1"
)
response.raise_for_status()
metadata = response.json()
return JSONResponse(metadata)
except Exception as e:
return JSONResponse(
{
"error": "server_error",
"error_description": f"Failed to fetch PropelAuth metadata: {e}",
},
status_code=500,
)
routes.append(
Route(
"/.well-known/oauth-authorization-server",
endpoint=oauth_authorization_server_metadata,
methods=["GET"],
)
)
return routes
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify token and check the ``aud`` claim against the configured resource."""
result = await super().verify_token(token)
if result is None or self._resource is None:
return result
aud = result.claims.get("aud")
if aud != self._resource:
logger.debug(
"PropelAuthProvider: token audience %r does not match resource %s",
aud,
self._resource,
)
return None
return result
def _create_token_verifier(
self,
introspection_url: str,
client_id: str,
client_secret: str | SecretStr,
required_scopes: list[str] | None,
introspection_overrides: PropelAuthTokenIntrospectionOverrides | None,
) -> IntrospectionTokenVerifier:
# Being defensive here, check for only the fields we are expecting
safe_overrides: PropelAuthTokenIntrospectionOverrides = {}
if introspection_overrides is not None:
if "timeout_seconds" in introspection_overrides:
safe_overrides["timeout_seconds"] = introspection_overrides[
"timeout_seconds"
]
if "cache_ttl_seconds" in introspection_overrides:
safe_overrides["cache_ttl_seconds"] = introspection_overrides[
"cache_ttl_seconds"
]
if "max_cache_size" in introspection_overrides:
safe_overrides["max_cache_size"] = introspection_overrides[
"max_cache_size"
]
if "http_client" in introspection_overrides:
safe_overrides["http_client"] = introspection_overrides["http_client"]
return IntrospectionTokenVerifier(
introspection_url=introspection_url,
client_id=client_id,
client_secret=client_secret,
required_scopes=required_scopes,
**safe_overrides,
)

336
tests/server/auth/providers/test_propelauth.py Normal file
View file

@ -0,0 +1,336 @@
"""Tests for PropelAuthProvider."""
from typing import cast
from unittest.mock import AsyncMock
import httpx
import pytest
from pydantic import SecretStr
from fastmcp import Client, FastMCP
from fastmcp.server.auth import AccessToken
from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
from fastmcp.server.auth.providers.propelauth import (
PropelAuthProvider,
PropelAuthTokenIntrospectionOverrides,
)
from fastmcp.utilities.tests import run_server_async
class TestPropelAuthProvider:
"""Test PropelAuth's auth provider."""
def test_init_with_only_required_params(self):
"""Test PropelAuthProvider initialization with only required params."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
)
# Verify the provider is configured correctly
assert len(provider.authorization_servers) == 1
assert (
str(provider.authorization_servers[0])
== "https://auth.example.com/oauth/2.1"
)
assert str(provider.base_url) == "https://example.com/"
# Verify token verifier is configured correctly
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert (
provider.token_verifier.introspection_url
== "https://auth.example.com/oauth/2.1/introspect"
)
assert provider.token_verifier.client_id == "client_id_123"
assert provider.token_verifier.client_secret == "client_secret_123"
def test_auth_url_trailing_slash_normalization(self):
"""Test that trailing slash on auth_url is stripped before building URLs."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com/",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert len(provider.authorization_servers) == 1
assert (
str(provider.authorization_servers[0])
== "https://auth.example.com/oauth/2.1"
)
assert (
provider.token_verifier.introspection_url
== "https://auth.example.com/oauth/2.1/introspect"
)
def test_required_scopes_passed_to_verifier(self):
"""Test that required_scopes are passed through to the token verifier."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
required_scopes=["read", "write"],
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier.required_scopes == ["read", "write"]
def test_introspection_client_secret_as_secret_str(self):
"""Test that SecretStr client_secret is unwrapped correctly."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret=SecretStr("my_secret"),
base_url="https://example.com",
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier.client_secret == "my_secret"
def test_authorization_servers_configuration(self):
"""Test that authorization_servers contains the correct PropelAuth URL."""
provider = PropelAuthProvider(
auth_url="https://auth.propelauth.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
)
assert len(provider.authorization_servers) == 1
assert (
str(provider.authorization_servers[0])
== "https://auth.propelauth.com/oauth/2.1"
)
def test_token_introspection_overrides_timeout(self):
"""Test that timeout_seconds override is passed to the verifier."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
token_introspection_overrides={"timeout_seconds": 30},
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier.timeout_seconds == 30
def test_token_introspection_overrides_cache(self):
"""Test that cache overrides are passed to the verifier."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
token_introspection_overrides={
"cache_ttl_seconds": 300,
"max_cache_size": 500,
},
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier._cache_ttl == 300
assert provider.token_verifier._max_cache_size == 500
def test_token_introspection_overrides_http_client(self):
"""Test that http_client override is passed to the verifier."""
client = httpx.AsyncClient()
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
token_introspection_overrides={"http_client": client},
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier._http_client is client
def test_token_introspection_overrides_ignores_unknown_keys(self):
"""Test that unknown override keys are silently ignored."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
# This won't typecheck without casting, since it shouldn't be allowed
token_introspection_overrides=cast(
PropelAuthTokenIntrospectionOverrides, {"unknown_key": "value"}
),
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier.timeout_seconds == 10
def test_token_introspection_overrides_ignores_disallowed_known_keys(self):
"""Test that known IntrospectionTokenVerifier keys not in the allow list are ignored."""
provider = PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
# This won't typecheck without casting, since it shouldn't be allowed
token_introspection_overrides=cast(
PropelAuthTokenIntrospectionOverrides, {"client_id": "sneaky_override"}
),
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier.client_id == "client_id_123"
class TestPropelAuthResourceChecking:
"""Test audience (aud) checking when resource is configured."""
def _make_provider(self, resource: str | None = None) -> PropelAuthProvider:
return PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="https://example.com",
resource=resource,
)
def _make_access_token(self, aud: str) -> AccessToken:
return AccessToken(
token="test-token",
client_id="client_id_123",
scopes=[],
claims={"active": True, "sub": "user-1", "aud": aud},
)
async def test_no_resource_skips_aud_check(self, monkeypatch: pytest.MonkeyPatch):
"""When resource is not configured, tokens are accepted without aud checking."""
provider = self._make_provider(resource=None)
token = self._make_access_token(aud="https://anything.example.com")
monkeypatch.setattr(
provider.token_verifier, "verify_token", AsyncMock(return_value=token)
)
result = await provider.verify_token("test-token")
assert result is token
async def test_aud_matches_resource(self, monkeypatch: pytest.MonkeyPatch):
"""Token is accepted when aud matches the configured resource."""
provider = self._make_provider(resource="https://api.example.com/mcp")
token = self._make_access_token(aud="https://api.example.com/mcp")
monkeypatch.setattr(
provider.token_verifier, "verify_token", AsyncMock(return_value=token)
)
result = await provider.verify_token("test-token")
assert result is token
async def test_aud_does_not_match_resource(self, monkeypatch: pytest.MonkeyPatch):
"""Token is rejected when aud doesn't match the configured resource."""
provider = self._make_provider(resource="https://api.example.com/mcp")
token = self._make_access_token(aud="https://other-server.example.com/mcp")
monkeypatch.setattr(
provider.token_verifier, "verify_token", AsyncMock(return_value=token)
)
result = await provider.verify_token("test-token")
assert result is None
async def test_inner_verifier_returns_none(self, monkeypatch: pytest.MonkeyPatch):
"""When the inner verifier rejects the token, None is returned without aud checking."""
provider = self._make_provider(resource="https://api.example.com/mcp")
monkeypatch.setattr(
provider.token_verifier, "verify_token", AsyncMock(return_value=None)
)
result = await provider.verify_token("test-token")
assert result is None
@pytest.fixture
async def mcp_server_url():
"""Start MCP server with PropelAuth authentication."""
mcp = FastMCP(
auth=PropelAuthProvider(
auth_url="https://auth.example.com",
introspection_client_id="client_id_123",
introspection_client_secret="client_secret_123",
base_url="http://localhost:4321",
)
)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
async with run_server_async(mcp, transport="http") as url:
yield url
class TestPropelAuthProviderIntegration:
async def test_unauthorized_access(self, mcp_server_url: str):
with pytest.raises(httpx.HTTPStatusError) as exc_info:
async with Client(mcp_server_url) as client:
tools = await client.list_tools() # noqa: F841
assert isinstance(exc_info.value, httpx.HTTPStatusError)
assert exc_info.value.response.status_code == 401
assert "tools" not in locals()
async def test_metadata_route_forwards_propelauth_response(
self,
monkeypatch: pytest.MonkeyPatch,
mcp_server_url: str,
) -> None:
"""Ensure PropelAuth metadata route proxies upstream JSON."""
metadata_payload = {
"issuer": "https://auth.example.com",
"token_endpoint": "https://auth.example.com/oauth/2.1/token",
"authorization_endpoint": "https://auth.example.com/oauth/2.1/authorize",
}
class DummyResponse:
status_code = 200
def __init__(self, data: dict[str, str]):
self._data = data
def json(self):
return self._data
def raise_for_status(self):
return None
class DummyAsyncClient:
last_url: str | None = None
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
async def get(self, url: str):
DummyAsyncClient.last_url = url
return DummyResponse(metadata_payload)
real_httpx_client = httpx.AsyncClient
monkeypatch.setattr(
"fastmcp.server.auth.providers.propelauth.httpx.AsyncClient",
DummyAsyncClient,
)
base_url = mcp_server_url.rsplit("/mcp", 1)[0]
async with real_httpx_client() as client:
response = await client.get(
f"{base_url}/.well-known/oauth-authorization-server"
)
assert response.status_code == 200
assert response.json() == metadata_payload
assert (
DummyAsyncClient.last_url
== "https://auth.example.com/.well-known/oauth-authorization-server/oauth/2.1"
)