diff --git a/docs/.ccignore b/docs/.ccignore new file mode 100644 index 000000000..ad117c3ee --- /dev/null +++ b/docs/.ccignore @@ -0,0 +1,2 @@ +changelog.mdx +python-sdk/ diff --git a/docs/clients/auth/oauth.mdx b/docs/clients/auth/oauth.mdx index 31b7fbdf4..56e02e0b4 100644 --- a/docs/clients/auth/oauth.mdx +++ b/docs/clients/auth/oauth.mdx @@ -55,6 +55,7 @@ async with Client("https://fastmcp.cloud/mcp", auth=oauth) as client: - **`client_name`** (`str`, optional): Client name for dynamic registration. Defaults to `"FastMCP Client"` - **`token_storage_cache_dir`** (`Path`, optional): Token cache directory. Defaults to `~/.fastmcp/oauth-mcp-client-cache/` - **`additional_client_metadata`** (`dict[str, Any]`, optional): Extra metadata for client registration +- **`callback_port`** (`int`, optional): Fixed port for OAuth callback server. If not specified, uses a random available port ## OAuth Flow @@ -72,7 +73,7 @@ If no valid tokens exist, the client attempts to discover the OAuth server's end If the OAuth server supports it and the client isn't already registered (or credentials aren't cached), the client performs dynamic client registration according to RFC 7591. -A temporary local HTTP server is started on an available port. This server's address (e.g., `http://127.0.0.1:/callback`) acts as the `redirect_uri` for the OAuth flow. +A temporary local HTTP server is started on an available port (or the port specified via `callback_port`). This server's address (e.g., `http://127.0.0.1:/callback`) acts as the `redirect_uri` for the OAuth flow. The user's default web browser is automatically opened, directing them to the OAuth server's authorization endpoint. The user logs in and grants (or denies) the requested `scopes`. diff --git a/docs/docs.json b/docs/docs.json index 263be3e14..8947aef0b 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -93,7 +93,12 @@ { "group": "Authentication", "icon": "shield-check", - "pages": ["servers/auth/verifiers"] + "pages": [ + "servers/auth/authentication", + "servers/auth/token-verification", + "servers/auth/remote-authentication", + "servers/auth/full-oauth-server" + ] } ] }, @@ -148,7 +153,8 @@ "integrations/openai", "integrations/openapi", "integrations/permit", - "integrations/starlette" + "integrations/starlette", + "integrations/authkit" ] }, { diff --git a/docs/integrations/anthropic.mdx b/docs/integrations/anthropic.mdx index 4736c122c..3d9a784bd 100644 --- a/docs/integrations/anthropic.mdx +++ b/docs/integrations/anthropic.mdx @@ -120,12 +120,12 @@ The MCP connector supports OAuth authentication through authorization tokens, wh The simplest way to add authentication to the server is to use a bearer token scheme. -For this example, we'll quickly generate our own tokens with FastMCP's `RSAKeyPair` utility, but this may not be appropriate for production use. For more details, see the complete server-side [Bearer Auth](/servers/auth/bearer) documentation. +For this example, we'll quickly generate our own tokens with FastMCP's `RSAKeyPair` utility, but this may not be appropriate for production use. For more details, see the complete server-side [Token Verification](/servers/auth/token-verification) documentation. We'll start by creating an RSA key pair to sign and verify tokens. ```python -from fastmcp.server.auth.verifiers import RSAKeyPair +from fastmcp.server.auth.providers.jwt import RSAKeyPair key_pair = RSAKeyPair.generate() access_token = key_pair.create_token(audience="dice-server") @@ -154,7 +154,7 @@ Here is a complete example that you can copy/paste. For simplicity and the purpo ```python server.py [expandable] from fastmcp import FastMCP from fastmcp.server.auth import JWTVerifier -from fastmcp.server.auth.verifiers import RSAKeyPair +from fastmcp.server.auth.providers.jwt import RSAKeyPair import random key_pair = RSAKeyPair.generate() diff --git a/docs/integrations/authkit.mdx b/docs/integrations/authkit.mdx new file mode 100644 index 000000000..b61dfae14 --- /dev/null +++ b/docs/integrations/authkit.mdx @@ -0,0 +1,103 @@ +--- +title: WorkOS AuthKit 🤝 FastMCP +sidebarTitle: WorkOS AuthKit +description: Secure your FastMCP server with WorkOS AuthKit +icon: shield-check +tag: NEW +--- + +import { VersionBadge } from "/snippets/version-badge.mdx" + + + +This guide shows you how to secure your FastMCP server using **WorkOS AuthKit**, a complete authentication and user management solution. This integration uses the [**Remote Authentication**](/servers/auth/remote-authentication) pattern, where WorkOS handles user login and your FastMCP server validates the tokens. + + +## Configuration +### Prerequisites + +Before you begin, you will need: +1. A **WorkOS Account** and a new **Project**. +2. An **AuthKit** instance configured within your WorkOS project. +3. Your FastMCP server's URL (can be localhost for development, e.g., `http://localhost:8000`). + +### Step 1: AuthKit Configuration + +In your WorkOS Dashboard, navigate to your AuthKit instance and configure the following settings: + + + + Go to **Applications → Configuration** and enable **Dynamic Client Registration**. This allows MCP clients register with your application automatically. + + ![Enable Dynamic Client Registration](./images/authkit/enable_dcr.png) + + + + Find your **AuthKit Domain** on the configuration page. It will look like `https://your-project-12345.authkit.app`. You'll need this for your FastMCP server configuration. + + + +### Step 2: FastMCP Configuration + +Create your FastMCP server file and use the `AuthKitProvider` to handle all the OAuth integration automatically: + +```python server.py +from fastmcp import FastMCP +from fastmcp.server.auth.providers.workos import AuthKitProvider + +# The AuthKitProvider automatically discovers WorkOS endpoints +# and configures JWT token validation +auth_provider = AuthKitProvider( + authkit_domain="https://your-project-12345.authkit.app", + base_url="http://localhost:8000" # Use your actual server URL +) + +mcp = FastMCP(name="AuthKit Secured App", auth=auth_provider) +``` + +## Testing + +To test your server, you can use the `fastmcp` CLI to run it locally. Assuming you've saved the above code to `server.py` (after replacing the `authkit_domain` and `base_url` with your actual values!), you can run the following command: + +```bash +fastmcp run server.py --transport http --port 8000 +``` + +Now, you can use a FastMCP client to test that you can reach your server after authenticating: + +```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()) +``` + + +## Environment Variables + +You can use environment variables to configure an AuthKit provider without instantiating the provider in your code. + +To do so, set the following environment variables: + +```bash +# instruct FastMCP to use the AuthKit provider +FASTMCP_SERVER_AUTH=AUTHKIT + +# configure the AuthKit provider +FASTMCP_SERVER_AUTH_AUTHKITPROVIDER_AUTHKIT_DOMAIN="https://your-project-12345.authkit.app" +FASTMCP_SERVER_AUTH_AUTHKITPROVIDER_BASE_URL="http://localhost:8000" +``` + +For clarity, you do **not** need to instantiate an auth provider when using environment variables: + +```python server.py +from fastmcp import FastMCP + +# FastMCP automatically creates the AuthKitProvider from environment variables +mcp = FastMCP(name="WorkOS Secured App") +``` \ No newline at end of file diff --git a/docs/integrations/images/authkit/enable_dcr.png b/docs/integrations/images/authkit/enable_dcr.png new file mode 100644 index 000000000..e5942f60e Binary files /dev/null and b/docs/integrations/images/authkit/enable_dcr.png differ diff --git a/docs/integrations/images/abac_condition_example.png b/docs/integrations/images/permit/abac_condition_example.png similarity index 100% rename from docs/integrations/images/abac_condition_example.png rename to docs/integrations/images/permit/abac_condition_example.png diff --git a/docs/integrations/images/abac_policy_example.png b/docs/integrations/images/permit/abac_policy_example.png similarity index 100% rename from docs/integrations/images/abac_policy_example.png rename to docs/integrations/images/permit/abac_policy_example.png diff --git a/docs/integrations/images/policy_mapping.png b/docs/integrations/images/permit/policy_mapping.png similarity index 100% rename from docs/integrations/images/policy_mapping.png rename to docs/integrations/images/permit/policy_mapping.png diff --git a/docs/integrations/images/role_assignement.png b/docs/integrations/images/permit/role_assignement.png similarity index 100% rename from docs/integrations/images/role_assignement.png rename to docs/integrations/images/permit/role_assignement.png diff --git a/docs/integrations/openai.mdx b/docs/integrations/openai.mdx index 28e3de841..0e9f3c585 100644 --- a/docs/integrations/openai.mdx +++ b/docs/integrations/openai.mdx @@ -118,12 +118,12 @@ The Responses API can include headers to authenticate the request, which means y The simplest way to add authentication to the server is to use a bearer token scheme. -For this example, we'll quickly generate our own tokens with FastMCP's `RSAKeyPair` utility, but this may not be appropriate for production use. For more details, see the complete server-side [Bearer Auth](/servers/auth/bearer) documentation. +For this example, we'll quickly generate our own tokens with FastMCP's `RSAKeyPair` utility, but this may not be appropriate for production use. For more details, see the complete server-side [Token Verification](/servers/auth/token-verification) documentation. We'll start by creating an RSA key pair to sign and verify tokens. ```python -from fastmcp.server.auth.verifiers import RSAKeyPair +from fastmcp.server.auth.providers.jwt import RSAKeyPair key_pair = RSAKeyPair.generate() access_token = key_pair.create_token(audience="dice-server") @@ -152,7 +152,7 @@ Here is a complete example that you can copy/paste. For simplicity and the purpo ```python server.py [expandable] from fastmcp import FastMCP from fastmcp.server.auth import JWTVerifier -from fastmcp.server.auth.verifiers import RSAKeyPair +from fastmcp.server.auth.providers.jwt import RSAKeyPair import random key_pair = RSAKeyPair.generate() diff --git a/docs/integrations/permit.mdx b/docs/integrations/permit.mdx index b91214503..9b1babe35 100644 --- a/docs/integrations/permit.mdx +++ b/docs/integrations/permit.mdx @@ -25,14 +25,14 @@ The middleware automatically maps MCP methods to Permit.io resources and actions - **Resource**: `{server_name}` (e.g., `myserver`) - **Action**: The tool name (e.g., `greet`) -![Permit.io Policy Mapping Example](./images/policy_mapping.png) +![Permit.io Policy Mapping Example](./images/permit/policy_mapping.png) *Example: In Permit.io, the 'Admin' role is granted permissions on resources and actions as mapped by the middleware. For example, 'greet', 'greet-jwt', and 'login' are actions on the 'mcp_server' resource, and 'list' is an action on the 'mcp_server_tools' resource.* > **Note:** > Don't forget to assign the relevant role (e.g., Admin, User) to the user authenticating to your MCP server (such as the user in the JWT) in the Permit.io Directory. Without the correct role assignment, users will not have access to the resources and actions you've configured in your policies. > -> ![Permit.io Directory Role Assignment Example](./images/role_assignement.png) +> ![Permit.io Directory Role Assignment Example](./images/permit/role_assignement.png) > > *Example: In Permit.io Directory, both 'client' and 'admin' users are assigned the 'Admin' role, granting them the permissions defined in your policy mapping.* @@ -223,7 +223,7 @@ mcp.add_middleware(PermitMcpMiddleware( The middleware supports Attribute-Based Access Control (ABAC) policies that can evaluate tool arguments as attributes. Tool arguments are automatically flattened as individual attributes (e.g., `arg_name`, `arg_number`) for granular policy conditions. -![ABAC Condition Example](./images/abac_condition_example.png) +![ABAC Condition Example](./images/permit/abac_condition_example.png) *Example: Create dynamic resources with conditions like `resource.arg_number greater-than 10` to allow the `conditional-greet` tool only when the number argument exceeds 10.* @@ -238,7 +238,7 @@ def conditional_greet(name: str, number: int) -> str: return f"Hello, {name}! Your number is {number}" ``` -![ABAC Policy Example](./images/abac_policy_example.png) +![ABAC Policy Example](./images/permit/abac_policy_example.png) *Example: The Admin role is granted access to the "conditional-greet" action on the "Big-greets" dynamic resource, while other tools like "greet", "greet-jwt", and "login" are granted on the base "mcp_server" resource.* diff --git a/docs/patterns/tool-transformation.mdx b/docs/patterns/tool-transformation.mdx index 3d7c5bd43..5f274fd75 100644 --- a/docs/patterns/tool-transformation.mdx +++ b/docs/patterns/tool-transformation.mdx @@ -567,7 +567,7 @@ Use a transform function returning `ToolResult` for complete control over both c Tool transformation is a flexible feature that supports many powerful patterns. Here are a few common use cases to give you ideas. ### Adapting Remote or Generated Tools -This is one of the most common reasons to use tool transformation. Tools from remote servers (via a [proxy](/servers/proxy)) or generated from an [OpenAPI spec](/servers/openapi) are often too generic for direct use by an LLM. You can use transformation to create a simpler, more intuitive version for your specific needs. +This is one of the most common reasons to use tool transformation. Tools from remote servers (via a [proxy](/servers/proxy)) or generated from an [OpenAPI spec](/integrations/openapi) are often too generic for direct use by an LLM. You can use transformation to create a simpler, more intuitive version for your specific needs. ### Chaining Transformations You can chain transformations by using an already transformed tool as the parent for a new transformation. This lets you build up complex behaviors in layers, for example, first renaming arguments, and then adding validation logic to the renamed tool. diff --git a/docs/servers/auth/authentication.mdx b/docs/servers/auth/authentication.mdx new file mode 100644 index 000000000..fe41caea4 --- /dev/null +++ b/docs/servers/auth/authentication.mdx @@ -0,0 +1,167 @@ +--- +title: Authentication +sidebarTitle: Overview +description: Secure your FastMCP server with flexible authentication patterns, from simple API keys to full OAuth 2.1 integration with external identity providers. +icon: user-shield +--- + +import { VersionBadge } from "/snippets/version-badge.mdx" + + + + +Authentication is only relevant for FastMCP's HTTP-based transports (`http` and `sse`). STDIO transport relies on the security of the local environment where it runs. + + +FastMCP provides a powerful and flexible authentication system designed to fit modern application needs. Authentication is a fast-moving and often confusing part of the MCP specification, so FastMCP endeavors to make it as straightforward as possible while adhering to industry best practices as the MCP community evolves new standards. + + +**Authentication is rapidly evolving in MCP.** The specification and best practices are changing quickly. FastMCP aims to provide stable, secure patterns that adapt to these changes while keeping your code simple and maintainable. + + +## Authentication Patterns + +MCP supports a variety of authentication options depending on how much of the authentication complexity you want to pull into your server itself. This can be thought of as a trade-off between whether your MCP server acts as a **Resource Server (RS)** that protects resources, an **Authorization Server (AS)** that handles user authentication and issues tokens, or neither. + +Think of it as a spectrum: + +- **No responsibility:** Your server has no authentication *(none)* +- **Minimal responsibility:** Your server only validates tokens issued elsewhere *(RS)* +- **Moderate responsibility:** Your server coordinates with external identity providers *(RS + remote AS)* — **recommended for most users** +- **Full responsibility:** Your server handles the entire authentication lifecycle *(RS + AS)* + +### Unauthenticated + +Unauthenticated FastMCP servers run without any mechanisms to protect their components. All tools and resources are publicly accessible to any client that can connect to your server. + +**Use this when:** +- Building development or testing environments +- Creating internal tools where network access controls provide sufficient security +- Prototyping before implementing proper authentication + + + +**Security considerations:** +- Only suitable for trusted environments or carefully designed public APIs +- Consider network-level security (VPNs, firewalls, private networks) +- Exercise extreme caution when exposing unauthenticated servers to the public internet +- Ensure any public endpoints only expose non-sensitive data or operations + + +### Token Verification + +Token verification is the conceptually simplest approach to authentication, where your FastMCP server acts as a pure **Resource Server**. Your server validates `Bearer` tokens on incoming requests but has no knowledge of how those tokens were obtained. This is analogous to how a web server validates API keys on incoming requests. Read more in the [token verification documentation](/servers/auth/token-verification). + + +**Protocol Note:** While simple to implement, this pattern operates somewhat outside the formal MCP authentication flow, which expects OAuth-style interactions. It's best suited for internal systems or when you have full control over token generation. + + +**Use this when:** +- You just need to validate tokens issued by another system +- Building internal microservices that trust a central auth service +- Working with static, long-lived API keys +- You control both the token issuer and your FastMCP server + +**Responsibilities you're taking on:** +- Token validation logic +- Ensuring tokens are securely transmitted to your server +- Managing token lifecycle in your issuing system + +### Remote Authentication + +This is the **recommended pattern for most FastMCP users** and follows the 2025-6-18 MCP protocol update. Your FastMCP server acts as a **Resource Server** and integrates with an external, trusted **Authorization Server** like WorkOS, Auth0, or Okta. You can learn more about this pattern in the [remote authentication documentation](/servers/auth/remote-authentication). + +**Use this when:** +- You want to integrate with external identity providers +- Building user-facing applications that need SSO +- You want enterprise-grade authentication without building it yourself +- You need features like multi-factor authentication, social logins, or directory sync + +**Responsibilities you're taking on:** +- Configuring your server to trust the external provider +- Token validation using the provider's public keys +- Mapping token claims to your application's user model + +**What the external provider handles:** +- User login and consent flows +- Token issuance and management +- User account management +- Security features like MFA and fraud detection + +### Full OAuth Server + + +**This is extremely advanced.** Most people should not build their own OAuth server. It requires deep security expertise and ongoing maintenance. Consider this only if you need complete control and have the resources to implement it securely. + + +In this pattern, your FastMCP server acts as both the **Authorization Server** and **Resource Server**, handling the entire authentication lifecycle from user login to token validation. Read more in the [full OAuth server documentation](/servers/auth/full-oauth-server). + +**Use this when:** +- Building a completely standalone application with its own user database +- You need full control over every aspect of authentication +- Prototyping OAuth flows locally without external dependencies +- You have the security expertise to implement OAuth securely + +**Responsibilities you're taking on:** +- Secure user credential storage and verification +- OAuth flow implementation +- Token lifecycle management +- Security measures like rate limiting and attack prevention +- User consent and account management interfaces +- Ongoing security updates and compliance + +## Configuring Authentication + +FastMCP provides a variety of `AuthProvider` classes that can be used to configure your server. To use one, instantiate it and pass it to your FastMCP server's `auth` parameter. + + + + +```python Token Verification +from fastmcp import FastMCP +from fastmcp.server.auth.providers.jwt import JWTVerifier + +jwt_verifier = JWTVerifier(...) + +mcp = FastMCP(name="My Server", auth=jwt_verifier) +``` + +```python Remote Authentication +from fastmcp import FastMCP +from fastmcp.server.auth.providers.workos import AuthKitProvider + +auth_provider = AuthKitProvider(...) + +mcp = FastMCP(name="My Server", auth=auth_provider) +``` + + +### Environment Variables + +For providers that support it, you can configure authentication entirely through environment variables. + +There are two steps to this process: + +1. Set `FASTMCP_SERVER_AUTH` to the registered name of your provider. For example, `JWT` for the `JWTVerifier` or `AUTHKIT` for the `AuthKitProvider`. +2. Set the appropriate environment variables for your provider in order to configure it. These are provider-specific and can be found in the provider's documentation. Not all providers will support environment variable configuration for all of their settings. + +For example, to configure a JWT verifier, you would set: + +```bash +export FASTMCP_SERVER_AUTH=JWT +export FASTMCP_SERVER_AUTH_JWT_JWKS_URI="https://your-idp.com/.well-known/jwks.json" +``` + +And now your FastMCP server will automatically be configured with the JWT verifier: + +```python +from fastmcp import FastMCP + +# Assumes the environment variables are set as above +mcp = FastMCP(name="My Protected Server") + +assert mcp.auth is not None +assert mcp.auth.jwks_uri == "https://your-idp.com/.well-known/jwks.json" +``` + +Note that if you provide an `auth` parameter to your FastMCP server, it will override the environment variable configuration. You can also set `auth=None` to disable authentication entirely and prohibit environment variable configuration. \ No newline at end of file diff --git a/docs/servers/auth/full-oauth-server.mdx b/docs/servers/auth/full-oauth-server.mdx new file mode 100644 index 000000000..c512b2bf6 --- /dev/null +++ b/docs/servers/auth/full-oauth-server.mdx @@ -0,0 +1,128 @@ +--- +title: Full OAuth Server +sidebarTitle: Full OAuth Server +description: Build a self-contained authentication system where your FastMCP server manages users, issues tokens, and validates them. +icon: users-between-lines +--- + +import { VersionBadge } from "/snippets/version-badge.mdx" + + + + +**This is an extremely advanced pattern.** Building a secure, production-ready OAuth 2.1 server is a complex undertaking that requires deep expertise in authentication protocols, cryptography, and security best practices. + +This pattern exists primarily to support the MCP protocol specification's requirements. **Most users should strongly prefer the [Remote Authentication pattern](/servers/auth/remote-authentication)** to integrate with a dedicated identity provider like WorkOS, Auth0, or Okta. + + +In the **Full OAuth Server** pattern, your FastMCP server acts as both the **Authorization Server (AS)** and the **Resource Server (RS)**. It becomes responsible for the entire authentication lifecycle: + +- **User Management**: Storing user credentials, profiles, and permissions +- **Client Registration**: Managing MCP client applications and their credentials +- **Authentication Flow**: Handling login pages, multi-factor authentication, and user consent +- **Token Lifecycle**: Issuing, refreshing, and revoking access tokens +- **Security Controls**: Rate limiting, audit logging, and threat detection + +This pattern should only be considered if you have strict requirements that prevent using external identity providers, such as air-gapped environments or highly specialized compliance needs. + +## Building an OAuth Provider + +To implement this pattern, you must subclass `fastmcp.server.auth.auth.OAuthProvider` and implement all of its abstract methods. This class extends the low-level OAuth authorization server interface and requires implementing the complete OAuth 2.1 specification. + +```python +from fastmcp.server.auth.auth import OAuthProvider +from mcp.server.auth.provider import ( + AccessToken, AuthorizationCode, RefreshToken, AuthorizationParams +) +from mcp.shared.auth import OAuthClientInformationFull, OAuthToken + +class MyOAuthProvider(OAuthProvider): + """ + A production OAuth provider implementation. + + WARNING: This is a simplified example. A real implementation + requires extensive security considerations, persistent storage, + proper error handling, and adherence to OAuth 2.1 security + best practices. + """ + + def __init__(self, base_url: str): + super().__init__(base_url=base_url) + # Initialize your database connections, cryptographic keys, etc. + + # === Client Management === + async def get_client(self, client_id: str) -> OAuthClientInformationFull | None: + """Retrieve client information by ID from your database.""" + # Query your client database and return client info or None if not found + raise NotImplementedError + + async def register_client(self, client_info: OAuthClientInformationFull) -> None: + """Store new client registration information.""" + # Validate and save client metadata to your database + # May raise RegistrationError if client data is invalid + raise NotImplementedError + + # === Authorization Flow === + async def authorize( + self, client: OAuthClientInformationFull, params: AuthorizationParams + ) -> str: + """ + Handle authorization request and return redirect URL. + + Many implementations redirect to a third-party OAuth provider, creating + a chain: Client -> MCP Server -> External IdP -> MCP Server -> Client. + You must generate an authorization code with at least 128 bits of entropy. + """ + # Authenticate user, get consent, generate auth code, return redirect URL + raise NotImplementedError + + async def load_authorization_code( + self, client: OAuthClientInformationFull, authorization_code: str + ) -> AuthorizationCode | None: + """Load authorization code from storage by code string.""" + # Look up stored authorization code, return None if not found or expired + raise NotImplementedError + + # === Token Management === + async def exchange_authorization_code( + self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode + ) -> OAuthToken: + """Exchange authorization code for access and refresh tokens.""" + # Validate code, generate new token pair, invalidate the auth code + raise NotImplementedError + + async def load_refresh_token( + self, client: OAuthClientInformationFull, refresh_token: str + ) -> RefreshToken | None: + """Load refresh token from storage by token string.""" + # Look up refresh token, return None if not found or expired + raise NotImplementedError + + async def exchange_refresh_token( + self, + client: OAuthClientInformationFull, + refresh_token: RefreshToken, + scopes: list[str] + ) -> OAuthToken: + """Exchange refresh token for new access/refresh token pair.""" + # Should rotate both tokens for security best practices + raise NotImplementedError + + async def load_access_token(self, token: str) -> AccessToken | None: + """Load and validate access token - called on every protected request.""" + # Look up token, check expiration, return None if invalid + raise NotImplementedError + + async def revoke_token(self, token: AccessToken | RefreshToken) -> None: + """Revoke access or refresh token.""" + # Should revoke both access and refresh tokens regardless of which is provided + # Do nothing if token is already invalid or revoked + raise NotImplementedError + + # === Token Verification (AuthProvider interface) === + async def verify_token(self, token: str) -> AccessToken | None: + """Verify bearer token for incoming requests.""" + # This is called on every protected MCP request + # Typically delegates to load_access_token + return await self.load_access_token(token) +``` diff --git a/docs/servers/auth/remote-authentication.mdx b/docs/servers/auth/remote-authentication.mdx new file mode 100644 index 000000000..a0d8c71e9 --- /dev/null +++ b/docs/servers/auth/remote-authentication.mdx @@ -0,0 +1,146 @@ +--- +title: Remote Authentication +sidebarTitle: Remote Authentication +description: Integrate with external identity providers like WorkOS, Auth0, or Okta by trusting them to handle user authentication. +icon: camera-cctv +--- + +import { VersionBadge } from "/snippets/version-badge.mdx" + + + +**Remote Authentication** is the recommended pattern for securing most production applications. In this model, your FastMCP server acts as a **Resource Server (RS)** and integrates with an external, trusted **Authorization Server (AS)**, such as WorkOS, Auth0, or a corporate SSO system. + +This approach lets you leverage robust, feature-rich identity platforms for user management, multi-factor authentication, and social logins, while your FastMCP server focuses on its core job: providing tools and resources. + +### How It Works + +The flow relies on the MCP client's ability to discover your server's authentication requirements. Your server doesn't handle logins itself; it tells the client where to find the real identity provider. + +The key endpoint is **`/.well-known/oauth-protected-resource`** which returns static metadata pointing to the authorization server. You can optionally also provide **`/.well-known/oauth-authorization-server`** that forwards the authorization server's metadata for convenience. + +```mermaid +sequenceDiagram + participant Client + participant FastMCPServer as FastMCP (RS) + participant ExternalIdP as External IdP (AS) + + Client->>FastMCPServer: 1. GET /.well-known/oauth-protected-resource + FastMCPServer-->>Client: 2. "Use https://my-idp.com for auth" + + note over Client, ExternalIdP: Client goes directly to the IdP + Client->>ExternalIdP: 3. GET /.well-known/oauth-authorization-server + ExternalIdP-->>Client: 4. OAuth endpoints & capabilities + + Client->>ExternalIdP: 5. User authenticates & gets token + ExternalIdP-->>Client: + + Client->>FastMCPServer: 6. MCP request with Bearer token + note right of FastMCPServer: Server verifies the token + FastMCPServer->>FastMCPServer: 7. Verify token signature
(using IdP's public keys) + FastMCPServer-->>Client: 8. MCP Response +``` + +## Building a Custom Provider + +To connect to any identity provider, you create a custom `AuthProvider` subclass. This class has two main responsibilities: + +1. **Verifying Tokens:** Validate tokens issued by the external provider. +2. **Forwarding Metadata:** Tell MCP clients where to find the external provider's login pages and token endpoints. + +### Step 1: Verifying Tokens + +Your provider must implement the `verify_token` method. For most modern identity providers that issue JWTs, you can simply delegate this task to FastMCP's built-in `JWTVerifier`. + +```python +from fastmcp.server.auth.auth import AuthProvider +from fastmcp.server.auth.providers.jwt import JWTVerifier +from mcp.server.auth.provider import AccessToken + +class MyIdPAuthProvider(AuthProvider): + def __init__(self): + super().__init__() + # The verifier validates tokens from the upstream provider. + self.token_verifier = JWTVerifier( + jwks_uri="https://my-idp.com/.well-known/jwks.json", + issuer="https://my-idp.com", + audience="my-fastmcp-api" + ) + + async def verify_token(self, token: str) -> AccessToken | None: + return await self.token_verifier.verify_token(token) +``` + +### Step 2: Adding Discovery Metadata + +Next, implement the `customize_auth_routes` method. The essential endpoint is `/.well-known/oauth-protected-resource` which tells clients where to find your authorization server. You can optionally add the authorization server forwarding endpoint for convenience. + +```python +import httpx +from starlette.responses import JSONResponse +from starlette.routing import Route + +class MyIdPAuthProvider(AuthProvider): + # ... (init and verify_token from above) ... + + def customize_auth_routes(self, routes: list[Route]) -> list[Route]: + # Essential: Tell clients which authorization server to use + async def protected_resource_metadata(request): + return JSONResponse({ + "resource": "https://my-fastmcp-server.com", + "authorization_servers": ["https://my-idp.com"], + "bearer_methods_supported": ["header"], + }) + + routes.append(Route("/.well-known/oauth-protected-resource", protected_resource_metadata)) + + # Optional: Forward the authorization server's metadata for convenience + # (Clients can also fetch this directly from the IdP) + async def authorization_server_metadata(request): + async with httpx.AsyncClient() as client: + resp = await client.get("https://my-idp.com/.well-known/oauth-authorization-server") + resp.raise_for_status() + return JSONResponse(resp.json()) + + routes.append(Route("/.well-known/oauth-authorization-server", authorization_server_metadata)) + return routes +``` + +### Step 3: Using Your Provider + +With these two methods implemented, your auth provider is now fully integrated with your identity provider. You can now use your custom provider with FastMCP by passing it to the `auth` parameter of your `FastMCP` instance: + +```python +from fastmcp import FastMCP + +mcp = FastMCP(name="My Secure Server", auth=MyIdPAuthProvider()) +``` + + +## Example: WorkOS AuthKit Provider + +FastMCP provides a built-in provider for **WorkOS AuthKit** that handles this entire pattern for you. It's a perfect example of the remote authentication pattern in action. + +**Prerequisites:** +1. A WorkOS account with an AuthKit project. +2. **Dynamic Client Registration (DCR)** must be enabled in your WorkOS application settings. +3. Your FastMCP server's URL must be added as a **Redirect URI** in your WorkOS project (can be localhost for development). + +```python +from fastmcp import FastMCP +from fastmcp.server.auth.providers.workos import AuthKitProvider + +# The AuthKitProvider implements both metadata forwarding and token validation. +auth_provider = AuthKitProvider( + # Your unique AuthKit domain from the WorkOS dashboard + authkit_domain="https://your-project.authkit.app", + # The URL of THIS FastMCP server (can be localhost for development) + base_url="https://your-fastmcp-server.com" +) + +mcp = FastMCP(name="My WorkOS-Protected Server", auth=auth_provider) +``` + + +For a complete, step-by-step tutorial on using this provider, see the [**WorkOS AuthKit Integration Guide**](/integrations/authkit). + \ No newline at end of file diff --git a/docs/servers/auth/token-verification.mdx b/docs/servers/auth/token-verification.mdx new file mode 100644 index 000000000..aedef6e38 --- /dev/null +++ b/docs/servers/auth/token-verification.mdx @@ -0,0 +1,210 @@ +--- +title: Token Verification +sidebarTitle: Token Verification +description: Protect your server by validating bearer tokens. +icon: key +--- + +import { VersionBadge } from "/snippets/version-badge.mdx" + + + +In the **Token Verification** pattern, your FastMCP server acts as a pure **Resource Server**. It validates Bearer tokens on incoming requests and uses the token claims to authorize access to protected resources (tools, resources, and prompts). It does not participate in user login, token issuance, or consent flows - it trusts tokens issued by another system. This is equivalent to how a traditional web server validates API keys on incoming requests. + +This is the right pattern if you have another system responsible for generating tokens and you simply need your FastMCP server to trust them or use the information in the token to make decisions. + +## JWT Verification + +JWT (JSON Web Token) verification is the recommended approach for production environments. The `JWTVerifier` class validates tokens using secure, asymmetric public key cryptography. This means your server only needs access to a public key to verify tokens, while the corresponding private key used for signing remains secure on your identity provider. + +### Using the JWTVerifier + +The most common and flexible approach is to point the verifier at a **JSON Web Key Set (JWKS)** endpoint. This allows your identity provider to rotate signing keys automatically without requiring you to update your server's configuration. + + +```python Using a JWKS Endpoint (Recommended) +from fastmcp import FastMCP +from fastmcp.server.auth.providers.jwt import JWTVerifier + +# The verifier will periodically fetch keys from this URL +# to validate incoming tokens. +verifier = JWTVerifier( + jwks_uri="https://my-identity-provider.com/.well-known/jwks.json", + issuer="https://my-identity-provider.com/", + audience="my-mcp-server-identifier" +) + +mcp = FastMCP(name="My Secure Server", auth=verifier) +``` + +```python Using a Static Public Key +from fastmcp import FastMCP +from fastmcp.server.auth.providers.jwt import JWTVerifier + +# This public key corresponds to the private key used by your token issuer. +# Use this only if a JWKS endpoint is not available. +public_key_pem = """-----BEGIN PUBLIC KEY----- +MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAy... +-----END PUBLIC KEY-----""" + +verifier = JWTVerifier( + public_key=public_key_pem, + issuer="https://my-identity-provider.com/", + audience="my-mcp-server-identifier" +) + +mcp = FastMCP(name="My Secure Server", auth=verifier) +``` + + + + + + + PEM-encoded public key for JWT verification. Use this for static public key configuration. Cannot be used together with `jwks_uri`. + + + + URI to fetch JSON Web Key Set (JWKS) for automatic key rotation. Cannot be used together with `public_key`. + + + + Expected JWT issuer claim (`iss`) for validation. If provided, tokens must have a matching issuer. + + + + Expected JWT audience claim (`aud`) for validation. Can be a single string or list of accepted audiences. + + + + JWT signing algorithm (e.g., RS256, HS256, ES256, PS256) + + + + List of scopes that all tokens must have. Tokens missing any required scope will be rejected. + + + + Resource server URL for OAuth protocol compliance + + + + + +### Environment Variable Configuration + +You can configure the `JWTVerifier` entirely through environment variables. Set `FASTMCP_SERVER_AUTH=JWT` to enable automatic configuration, then provide the verifier's settings. + +Example configuration: +```bash +export FASTMCP_SERVER_AUTH=JWT +export FASTMCP_SERVER_AUTH_JWT_JWKS_URI="https://your-idp.com/.well-known/jwks.json" +export FASTMCP_SERVER_AUTH_JWT_ISSUER="https://your-idp.com" +export FASTMCP_SERVER_AUTH_JWT_AUDIENCE="your-server-id" +``` + +Your FastMCP server will now be automatically configured with JWT verification: +```python +from fastmcp import FastMCP + +# This server is automatically protected with JWT verification +# based on the environment variables. +mcp = FastMCP(name="My Protected Server") +``` + + + + PEM-encoded public key for JWT verification. Use this for static public key configuration. + + + + URI to fetch JSON Web Key Set (JWKS). Use this for automatic key rotation support. + + + + Expected JWT issuer claim for validation + + + + Expected JWT audience claim for validation + + + + JWT signing algorithm (e.g., RS256, HS256, ES256) + + + + Comma-separated list of required scopes that all tokens must have + + + + Resource server URL for OAuth protocol compliance + + + +## Development Tools + +For local development and testing, managing JWTs can be cumbersome. FastMCP provides simpler tools for these scenarios. + +### Static Token Verification + +The `StaticTokenVerifier` validates tokens against a hardcoded set of tokens and claims. It's perfect for quickly getting a secure server running in your local environment for testing or prototyping without the complexity of a real identity provider. + + + +**Never use static token verification in production.** Tokens are stored as plain text strings, which is highly insecure. + + +To use the static token verifier, you need to provide a dictionary of tokens and their claims. Each key of the dictionary is a token, and the value is a dictionary of token data. Note that the tokens will be stored as plain text strings and validated as-is. For example, the following configuration would recognize a token provided in the `Authorization` header as `Bearer dev-token-for-alice` and load `alice@example.com` as the client ID: + +```python +from fastmcp import FastMCP +from fastmcp.server.auth.providers.jwt import StaticTokenVerifier + +verifier = StaticTokenVerifier( + tokens={ + "dev-token-for-alice": { + "client_id": "alice@example.com", + "scopes": ["read:data", "write:data"] + }, + "readonly-token-for-guest": { + "client_id": "guest-user", + "scopes": ["read:data"] + } + }, + required_scopes=["read:data"] # Optionally enforce a base scope for all tokens. +) + +mcp = FastMCP(name="Development Server", auth=verifier) +``` + +### Generating Test Tokens + +To help test a `JWTVerifier`-protected server, FastMCP includes a simple `RSAKeyPair` utility to generate a key pair and sign your own JWTs. This is for convenience in development and is not intended for production use. + +```python +from fastmcp import FastMCP +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair + +# 1. In a secure part of your test setup, generate a key pair. +key_pair = RSAKeyPair.generate() + +# 2. Configure your FastMCP server's verifier with the PUBLIC key. +auth_verifier = JWTVerifier( + public_key=key_pair.public_key, + issuer="https://dev.fastmcp.com", + audience="test-server" +) +mcp = FastMCP(name="Test Server", auth=auth_verifier) + +# 3. Use the PRIVATE key to create a valid token for your client tests. +test_token = key_pair.create_token( + subject="test-user-123", + issuer="https://dev.fastmcp.com", + audience="test-server", + scopes=["read", "write"] +) + +print(f"Generated Test Token:\n{test_token}") +``` + diff --git a/docs/servers/auth/verifiers.mdx b/docs/servers/auth/verifiers.mdx deleted file mode 100644 index 46f447e4b..000000000 --- a/docs/servers/auth/verifiers.mdx +++ /dev/null @@ -1,332 +0,0 @@ ---- -title: Token Verification -sidebarTitle: Token Verification -description: Secure your FastMCP server's HTTP endpoints by validating JWT tokens. -icon: key -tag: NEW ---- - -import { VersionBadge } from "/snippets/version-badge.mdx" - - - - -Authentication and authorization are only relevant for HTTP-based transports. - - -Bearer Token authentication is a common way to secure HTTP-based APIs. In this model, the client sends a token (usually a JSON Web Token or JWT) in the `Authorization` header with the "Bearer" scheme. The server then validates this token to grant or deny access. - -FastMCP supports Bearer Token authentication for its HTTP-based transports (`http` and `sse`), allowing you to protect your server from unauthorized access. - -## Authentication Strategy - -FastMCP uses **asymmetric encryption** for token validation, which provides a clean security separation between token issuers and FastMCP servers. This approach means: - -- **No shared secrets**: Your FastMCP server never needs access to private keys or client secrets -- **Public key verification**: The server only needs a public key (or JWKS endpoint) to verify token signatures -- **Secure token issuance**: Tokens are signed by an external service using a private key that never leaves the issuer -- **Scalable architecture**: Multiple FastMCP servers can validate tokens without coordinating secrets - -This design allows you to integrate FastMCP servers into existing authentication infrastructures without compromising security boundaries. - -## Token Verification Approaches - -FastMCP provides three token verification approaches: - -### JWTVerifier -Validates JWT tokens using public key cryptography. Use when you have JWT tokens issued by an external identity provider (Auth0, Okta, Keycloak, etc.) and want self-contained validation without network calls. - -### IntrospectionTokenVerifier -Validates tokens by calling a remote OAuth 2.0 authorization server's introspection endpoint (RFC 7662). Use when your authorization server is separate from your FastMCP server, you're using opaque tokens, or you need real-time token revocation. - -### StaticTokenVerifier -Validates tokens against a predefined dictionary. Use for development and testing only - never in production. - - -These verifiers validate tokens; they do **not** issue them (or implement any part of an OAuth flow). You'll need to generate tokens separately, either using FastMCP utilities or an external Identity Provider (IdP) or OAuth 2.1 Authorization Server. - - - -### Configuration Parameters - - - - - - RSA public key in PEM format for static key validation. Required if `jwks_uri` is not provided - - - - URL for JSON Web Key Set endpoint. Required if `public_key` is not provided - - - - Expected JWT `iss` claim value - - - - Algorithm for decoding JWT token. Defaults to 'RS256' - - - - Expected JWT `aud` claim value - - - - Global scopes required for all requests - - - - - - - - OAuth 2.0 Token Introspection endpoint URL (RFC 7662) - - - - Resource server client ID for introspection authentication - - - - Resource server client secret for introspection authentication - - - - Global scopes required for all requests - - - - - - - - Mapping of valid tokens to their claims. Each token maps to a dictionary containing token metadata like `sub`, `scope`, etc. - - - - Global scopes required for all requests - - - - - -## JWT Verification - -The `JWTVerifier` validates JWT tokens using public key cryptography. Use this when you have JWT tokens issued by an external identity provider and want self-contained validation without network calls. - -```python -from fastmcp import FastMCP -from fastmcp.server.auth.verifiers import JWTVerifier - -verifier = JWTVerifier( - jwks_uri="https://my-identity-provider.com/.well-known/jwks.json", - issuer="https://my-identity-provider.com/", - audience="my-mcp-server" -) - -mcp = FastMCP(name="My MCP Server", auth=verifier) -``` - -### Public Key Configuration - -#### Using a Static Public Key - -If you have a public key in PEM format, you can provide it to the `JWTVerifier` as a string. - -```python {12} -from fastmcp.server.auth.verifiers import JWTVerifier -import inspect - -public_key_pem = inspect.cleandoc( - """ - -----BEGIN PUBLIC KEY----- - MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAy... - -----END PUBLIC KEY----- - """ -) - -auth = JWTVerifier(public_key=public_key_pem) -``` - -#### Using JWKS URI - -```python -verifier = JWTVerifier( - jwks_uri="https://idp.example.com/.well-known/jwks.json" -) -``` - - -JWKS is recommended for production as it supports automatic key rotation and multiple signing keys. - - -## OAuth 2.0 Token Introspection - -The `IntrospectionTokenVerifier` validates tokens by calling an OAuth 2.0 authorization server's introspection endpoint (RFC 7662). This is useful when your authorization server is separate from your FastMCP server, you're using opaque tokens, or you need real-time token validation with immediate revocation support. - -```python -from fastmcp.server.auth.verifiers import IntrospectionTokenVerifier - -verifier = IntrospectionTokenVerifier( - introspection_endpoint="https://auth.company.com/oauth/introspect", - server_url="https://mcp.company.com", # This server's URL - client_id="mcp-resource-server", - client_secret="your-secret", - required_scopes=["mcp:access"] -) - -mcp = FastMCP(name="MCP Server", auth=verifier) -``` - -For each request, the verifier makes an HTTP call to the introspection endpoint to check if the token is valid and active. This provides real-time validation but requires network connectivity. - -## Static Token Verification - -The `StaticTokenVerifier` validates tokens against a predefined dictionary of token strings and claims. Use this for development and testing when you need predictable tokens without setting up a real OAuth server. - -```python -from fastmcp.server.auth.verifiers import StaticTokenVerifier - -verifier = StaticTokenVerifier( - tokens={ - "dev-token-123": { - "client_id": "dev-user", - "scopes": ["read", "write"], - "sub": "developer@example.com" - }, - "readonly-token": { - "client_id": "readonly-user", - "scopes": ["read"], - "expires_at": 1735689600 # Optional expiration - } - }, - required_scopes=["read"] -) - -mcp = FastMCP(name="Development Server", auth=verifier) -``` - -Token claims can include `client_id` (required), `scopes`, `sub`, `expires_at`, and any custom metadata your application needs. - - -Never use StaticTokenVerifier in production - tokens are stored in plain text. - - -## Generating Tokens - -For development and testing, FastMCP provides the `RSAKeyPair` utility class to generate tokens without needing an external OAuth provider. - - -The `RSAKeyPair` utility is intended for development and testing only. For production, use a proper OAuth 2.1 Authorization Server or Identity Provider. - -### Basic Token Generation - -```python -from fastmcp import FastMCP -from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair - -# Generate a new key pair -key_pair = RSAKeyPair.generate() - -# Configure the auth verifier with the public key -auth = JWTVerifier( - public_key=key_pair.public_key, - issuer="https://dev.example.com", - audience="my-dev-server" -) - -mcp = FastMCP(name="Development Server", auth=auth) - -# Generate a token for testing -token = key_pair.create_token( - subject="dev-user", - issuer="https://dev.example.com", - audience="my-dev-server", - scopes=["read", "write"] -) - -print(f"Test token: {token}") -``` - -### Token Creation Parameters - -The `create_token()` method accepts these parameters: - - - - JWT subject claim (usually user ID) - - - - JWT issuer claim - - - - JWT audience claim - - - - OAuth scopes to include - - - - Token expiration time in seconds - - - - Extra claims to include in the token - - - - Key ID for JWKS lookup - - - - - -## Accessing Token Claims - -Once authenticated, your tools, resources, or prompts can access token information using the `get_access_token()` dependency function: - -```python -from fastmcp import FastMCP, Context, ToolError -from fastmcp.server.dependencies import get_access_token, AccessToken - -@mcp.tool -async def get_my_data(ctx: Context) -> dict: - access_token: AccessToken = get_access_token() - - user_id = access_token.client_id # From JWT 'sub' or 'client_id' claim - user_scopes = access_token.scopes - - if "data:read_sensitive" not in user_scopes: - raise ToolError("Insufficient permissions: 'data:read_sensitive' scope required.") - - return { - "user": user_id, - "sensitive_data": f"Private data for {user_id}", - "granted_scopes": user_scopes - } -``` - -### AccessToken Properties - - - - The raw JWT string - - - - Authenticated principal identifier - - - - Granted scopes - - - - Token expiration timestamp - - - diff --git a/docs/servers/server.mdx b/docs/servers/server.mdx index 1a54c8964..971119e6d 100644 --- a/docs/servers/server.mdx +++ b/docs/servers/server.mdx @@ -41,7 +41,7 @@ The `FastMCP` constructor accepts several arguments: - Authentication provider for securing HTTP-based transports. See [Bearer Token Authentication](/servers/auth/bearer) for configuration options + Authentication provider for securing HTTP-based transports. See [Authentication](/servers/auth/authentication) for configuration options diff --git a/docs/tutorials/rest-api.mdx b/docs/tutorials/rest-api.mdx index a0efeae62..445932626 100644 --- a/docs/tutorials/rest-api.mdx +++ b/docs/tutorials/rest-api.mdx @@ -35,7 +35,7 @@ For this tutorial, we'll use the [JSONPlaceholder API](https://jsonplaceholder.t Now for the magic. We'll use `FastMCP.from_openapi`. This method takes an `httpx.AsyncClient` configured for your API and its OpenAPI specification, and automatically converts **every endpoint** into a callable MCP `Tool`. -Learn more about working with OpenAPI specs in the [OpenAPI integration docs](/servers/openapi). +Learn more about working with OpenAPI specs in the [OpenAPI integration docs](/integrations/openapi). @@ -144,7 +144,7 @@ However, for clients that support the full MCP spec, representing `GET` requests FastMCP allows users to customize this behavior using the concept of "route maps". A `RouteMap` is a mapping of an API route to an MCP type. FastMCP checks each API route against your custom maps in order. If a route matches a map, it's converted to the specified `mcp_type`. Any route that doesn't match your custom maps will fall back to the default behavior (becoming a `Tool`). -Learn more about route maps in the [OpenAPI integration docs](/servers/openapi#route-mapping). +Learn more about route maps in the [OpenAPI integration docs](/integrations/openapi#route-mapping). Here’s how you can add custom route maps to turn `GET` requests into `Resources` and `ResourceTemplates` (if they have path parameters): diff --git a/pyproject.toml b/pyproject.toml index 40b4682c2..308cabc0c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -73,7 +73,6 @@ Documentation = "https://gofastmcp.com" [project.optional-dependencies] websockets = ["websockets>=15.0.1"] - [build-system] requires = ["hatchling", "uv-dynamic-versioning>=0.7.0"] build-backend = "hatchling.build" diff --git a/src/fastmcp/client/auth/oauth.py b/src/fastmcp/client/auth/oauth.py index ca86088d4..06f71253e 100644 --- a/src/fastmcp/client/auth/oauth.py +++ b/src/fastmcp/client/auth/oauth.py @@ -5,7 +5,7 @@ import json import webbrowser from pathlib import Path from typing import Any, Literal -from urllib.parse import urljoin, urlparse +from urllib.parse import urlparse import anyio import httpx @@ -13,7 +13,6 @@ from mcp.client.auth import OAuthClientProvider, TokenStorage from mcp.shared.auth import ( OAuthClientInformationFull, OAuthClientMetadata, - OAuthMetadata, ) from mcp.shared.auth import ( OAuthToken as OAuthToken, @@ -150,41 +149,6 @@ class FileTokenStorage(TokenStorage): logger.info("Cleared all OAuth client cache data.") -async def discover_oauth_metadata( - server_base_url: str, httpx_kwargs: dict[str, Any] | None = None -) -> OAuthMetadata | None: - """ - Discover OAuth metadata from the server using RFC 8414 well-known endpoint. - - Args: - server_base_url: Base URL of the OAuth server (e.g., "https://example.com") - httpx_kwargs: Additional kwargs for httpx client - - Returns: - OAuth metadata if found, None otherwise - """ - well_known_url = urljoin(server_base_url, "/.well-known/oauth-authorization-server") - logger.debug(f"Discovering OAuth metadata from: {well_known_url}") - - async with httpx.AsyncClient(**(httpx_kwargs or {})) as client: - try: - response = await client.get(well_known_url, timeout=10.0) - if response.status_code == 200: - logger.debug("Successfully discovered OAuth metadata") - return OAuthMetadata.model_validate(response.json()) - elif response.status_code == 404: - logger.debug( - "OAuth metadata not found (404) - server may not require auth" - ) - return None - else: - logger.warning(f"OAuth metadata request failed: {response.status_code}") - return None - except (httpx.RequestError, json.JSONDecodeError, ValidationError) as e: - logger.debug(f"OAuth metadata discovery failed: {e}") - return None - - async def check_if_auth_required( mcp_url: str, httpx_kwargs: dict[str, Any] | None = None ) -> bool: @@ -215,70 +179,86 @@ async def check_if_auth_required( return True -def OAuth( - mcp_url: str, - scopes: str | list[str] | None = None, - client_name: str = "FastMCP Client", - token_storage_cache_dir: Path | None = None, - additional_client_metadata: dict[str, Any] | None = None, -) -> OAuthClientProvider: +class OAuth(OAuthClientProvider): """ - Create an OAuthClientProvider for an MCP server. + OAuth client provider for MCP servers with browser-based authentication. - This is intended to be provided to the `auth` parameter of an - httpx.AsyncClient (or appropriate FastMCP client/transport instance) - - Args: - mcp_url: Full URL to the MCP endpoint (e.g. "http://host/mcp/sse/") - scopes: OAuth scopes to request. Can be a - space-separated string or a list of strings. - client_name: Name for this client during registration - token_storage_cache_dir: Directory for FileTokenStorage - additional_client_metadata: Extra fields for OAuthClientMetadata - - Returns: - OAuthClientProvider + This class provides OAuth authentication for FastMCP clients by opening + a browser for user authorization and running a local callback server. """ - parsed_url = urlparse(mcp_url) - server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" - # Setup OAuth client - redirect_port = find_available_port() - redirect_uri = f"http://127.0.0.1:{redirect_port}/callback" + def __init__( + self, + mcp_url: str, + scopes: str | list[str] | None = None, + client_name: str = "FastMCP Client", + token_storage_cache_dir: Path | None = None, + additional_client_metadata: dict[str, Any] | None = None, + callback_port: int | None = None, + ): + """ + Initialize OAuth client provider for an MCP server. - if isinstance(scopes, list): - scopes = " ".join(scopes) + Args: + mcp_url: Full URL to the MCP endpoint (e.g. "http://host/mcp/sse/") + scopes: OAuth scopes to request. Can be a + space-separated string or a list of strings. + client_name: Name for this client during registration + token_storage_cache_dir: Directory for FileTokenStorage + additional_client_metadata: Extra fields for OAuthClientMetadata + callback_port: Fixed port for OAuth callback (default: random available port) + """ + parsed_url = urlparse(mcp_url) + server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" - client_metadata = OAuthClientMetadata( - client_name=client_name, - redirect_uris=[AnyHttpUrl(redirect_uri)], - grant_types=["authorization_code", "refresh_token"], - response_types=["code"], - token_endpoint_auth_method="client_secret_post", - scope=scopes, - **(additional_client_metadata or {}), - ) + # Setup OAuth client + self.redirect_port = callback_port or find_available_port() + redirect_uri = f"http://localhost:{self.redirect_port}/callback" - # Create server-specific token storage - storage = FileTokenStorage( - server_url=server_base_url, cache_dir=token_storage_cache_dir - ) + if isinstance(scopes, list): + scopes = " ".join(scopes) - # Define OAuth handlers - async def redirect_handler(authorization_url: str) -> None: + client_metadata = OAuthClientMetadata( + client_name=client_name, + redirect_uris=[AnyHttpUrl(redirect_uri)], + grant_types=["authorization_code", "refresh_token"], + response_types=["code"], + # token_endpoint_auth_method="client_secret_post", + scope=scopes, + **(additional_client_metadata or {}), + ) + + # Create server-specific token storage + storage = FileTokenStorage( + server_url=server_base_url, cache_dir=token_storage_cache_dir + ) + + # Store server_base_url for use in callback_handler + self.server_base_url = server_base_url + + # Initialize parent class + super().__init__( + server_url=server_base_url, + client_metadata=client_metadata, + storage=storage, + redirect_handler=self.redirect_handler, + callback_handler=self.callback_handler, + ) + + async def redirect_handler(self, authorization_url: str) -> None: """Open browser for authorization.""" logger.info(f"OAuth authorization URL: {authorization_url}") webbrowser.open(authorization_url) - async def callback_handler() -> tuple[str, str | None]: + async def callback_handler(self) -> tuple[str, str | None]: """Handle OAuth callback and return (auth_code, state).""" # Create a future to capture the OAuth response response_future = asyncio.get_running_loop().create_future() # Create server with the future server = create_oauth_callback_server( - port=redirect_port, - server_url=server_base_url, + port=self.redirect_port, + server_url=self.server_base_url, response_future=response_future, ) @@ -286,7 +266,7 @@ def OAuth( async with anyio.create_task_group() as tg: tg.start_soon(server.serve) logger.info( - f"🎧 OAuth callback server started on http://127.0.0.1:{redirect_port}" + f"🎧 OAuth callback server started on http://localhost:{self.redirect_port}" ) TIMEOUT = 300.0 # 5 minute timeout @@ -300,14 +280,3 @@ def OAuth( server.should_exit = True await asyncio.sleep(0.1) # Allow server to shutdown gracefully tg.cancel_scope.cancel() - - # Create OAuth provider - oauth_provider = OAuthClientProvider( - server_url=server_base_url, - client_metadata=client_metadata, - storage=storage, - redirect_handler=redirect_handler, - callback_handler=callback_handler, - ) - - return oauth_provider diff --git a/src/fastmcp/client/oauth_callback.py b/src/fastmcp/client/oauth_callback.py index 891e4cdb0..81386c406 100644 --- a/src/fastmcp/client/oauth_callback.py +++ b/src/fastmcp/client/oauth_callback.py @@ -252,6 +252,24 @@ def create_oauth_callback_server( status_code=400, ) + # Check for missing state parameter (indicates OAuth flow issue) + if callback_response.state is None: + # Resolve future with exception if provided + if response_future and not response_future.done(): + response_future.set_exception( + RuntimeError( + "OAuth server did not return state parameter - authentication failed" + ) + ) + + return HTMLResponse( + create_callback_html( + "FastMCP OAuth Error: Authentication failed
The OAuth server did not return the expected state parameter", + is_success=False, + ), + status_code=400, + ) + # Success case if response_future and not response_future.done(): response_future.set_result( diff --git a/src/fastmcp/contrib/component_manager/example.py b/src/fastmcp/contrib/component_manager/example.py index 3a04de493..5ce214b2b 100644 --- a/src/fastmcp/contrib/component_manager/example.py +++ b/src/fastmcp/contrib/component_manager/example.py @@ -1,6 +1,6 @@ from fastmcp import FastMCP from fastmcp.contrib.component_manager import set_up_component_manager -from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair key_pair = RSAKeyPair.generate() diff --git a/src/fastmcp/server/auth/__init__.py b/src/fastmcp/server/auth/__init__.py index 307d170f6..a6da4b7be 100644 --- a/src/fastmcp/server/auth/__init__.py +++ b/src/fastmcp/server/auth/__init__.py @@ -1,11 +1,10 @@ from .auth import OAuthProvider, TokenVerifier -from .verifiers import IntrospectionTokenVerifier, JWTVerifier, StaticTokenVerifier +from .providers.jwt import JWTVerifier, StaticTokenVerifier __all__ = [ "OAuthProvider", "TokenVerifier", - "IntrospectionTokenVerifier", "JWTVerifier", "StaticTokenVerifier", ] diff --git a/src/fastmcp/server/auth/auth.py b/src/fastmcp/server/auth/auth.py index 709d136de..9fee9a2ee 100644 --- a/src/fastmcp/server/auth/auth.py +++ b/src/fastmcp/server/auth/auth.py @@ -1,3 +1,7 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + from mcp.server.auth.provider import ( AccessToken, AuthorizationCode, @@ -12,10 +16,59 @@ from mcp.server.auth.settings import ( RevocationOptions, ) from pydantic import AnyHttpUrl +from starlette.routing import Route + +if TYPE_CHECKING: + pass -class TokenVerifier(TokenVerifierProtocol): - """Base class for token verifiers (Resource Servers).""" +class AuthProvider: + """Base class for all FastMCP authentication providers. + + This class provides a unified interface for all authentication providers, + whether they are simple token verifiers or full OAuth authorization servers. + All providers must be able to verify tokens and can optionally provide + custom authentication routes. + """ + + def __init__(self, required_scopes: list[str] | None = None): + """Initialize the auth provider.""" + self.required_scopes: list[str] = required_scopes or [] + + async def verify_token(self, token: str) -> AccessToken | None: + """Verify a bearer token and return access info if valid. + + All auth providers must implement token verification. + + Args: + token: The token string to validate + + Returns: + AccessToken object if valid, None if invalid or expired + """ + raise NotImplementedError("Subclasses must implement verify_token") + + def customize_auth_routes(self, routes: list[Route]) -> list[Route]: + """Customize authentication routes after standard creation. + + This method allows providers to modify or add to the standard OAuth routes. + The default implementation returns the routes unchanged. + + Args: + routes: List of standard routes (may be empty for token-only providers) + + Returns: + List of routes (potentially modified or extended) + """ + return routes + + +class TokenVerifier(AuthProvider, TokenVerifierProtocol): + """Base class for token verifiers (Resource Servers). + + This class provides token verification capability without OAuth server functionality. + Token verifiers typically don't provide authentication routes by default. + """ def __init__( self, @@ -29,6 +82,10 @@ class TokenVerifier(TokenVerifierProtocol): resource_server_url: The URL of this resource server (for RFC 8707 resource indicators) required_scopes: Scopes that are required for all requests """ + # Initialize AuthProvider (no args needed) + AuthProvider.__init__(self, required_scopes=required_scopes) + + # Handle our own resource_server_url and required_scopes self.resource_server_url: AnyHttpUrl | None if resource_server_url is None: self.resource_server_url = None @@ -36,7 +93,6 @@ class TokenVerifier(TokenVerifierProtocol): self.resource_server_url = AnyHttpUrl(resource_server_url) else: self.resource_server_url = resource_server_url - self.required_scopes = required_scopes or [] async def verify_token(self, token: str) -> AccessToken | None: """Verify a bearer token and return access info if valid.""" @@ -44,11 +100,20 @@ class TokenVerifier(TokenVerifierProtocol): class OAuthProvider( - OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken] + AuthProvider, + OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken], ): + """OAuth Authorization Server provider. + + This class provides full OAuth server functionality including client registration, + authorization flows, token issuance, and token verification. + """ + def __init__( self, - issuer_url: AnyHttpUrl | str, + *, + base_url: AnyHttpUrl | str, + issuer_url: AnyHttpUrl | str | None = None, service_documentation_url: AnyHttpUrl | str | None = None, client_registration_options: ClientRegistrationOptions | None = None, revocation_options: RevocationOptions | None = None, @@ -59,20 +124,74 @@ class OAuthProvider( Initialize the OAuth provider. Args: - issuer_url: The URL of the OAuth issuer. + base_url: The public URL of this FastMCP server + issuer_url: The issuer URL for OAuth metadata (defaults to base_url) service_documentation_url: The URL of the service documentation. client_registration_options: The client registration options. revocation_options: The revocation options. required_scopes: Scopes that are required for all requests. + resource_server_url: The URL of this resource server (for RFC 8707 resource indicators, defaults to base_url) """ + super().__init__() - if isinstance(issuer_url, str): - issuer_url = AnyHttpUrl(issuer_url) + + # Convert URLs to proper types + if isinstance(base_url, str): + base_url = AnyHttpUrl(base_url) + self.base_url = base_url + + if issuer_url is None: + self.issuer_url = base_url + elif isinstance(issuer_url, str): + self.issuer_url = AnyHttpUrl(issuer_url) + else: + self.issuer_url = issuer_url + + # Handle our own resource_server_url and required_scopes + if resource_server_url is None: + self.resource_server_url = base_url + elif isinstance(resource_server_url, str): + self.resource_server_url = AnyHttpUrl(resource_server_url) + else: + self.resource_server_url = resource_server_url + self.required_scopes = required_scopes or [] + + # Initialize OAuth Authorization Server Provider + OAuthAuthorizationServerProvider.__init__(self) + if isinstance(service_documentation_url, str): service_documentation_url = AnyHttpUrl(service_documentation_url) - self.issuer_url = issuer_url self.service_documentation_url = service_documentation_url self.client_registration_options = client_registration_options self.revocation_options = revocation_options - self.required_scopes = required_scopes + + async def verify_token(self, token: str) -> AccessToken | None: + """ + Verify a bearer token and return access info if valid. + + This method implements the TokenVerifier protocol by delegating + to our existing load_access_token method. + + Args: + token: The token string to validate + + Returns: + AccessToken object if valid, None if invalid or expired + """ + return await self.load_access_token(token) + + def customize_auth_routes(self, routes: list[Route]) -> list[Route]: + """Customize OAuth authentication routes after standard creation. + + This method allows providers to modify the standard OAuth routes + returned by create_auth_routes. The default implementation returns + the routes unchanged. + + Args: + routes: List of standard OAuth routes from create_auth_routes + + Returns: + List of routes (potentially modified) + """ + return routes diff --git a/src/fastmcp/server/auth/providers/bearer.py b/src/fastmcp/server/auth/providers/bearer.py index 5c13a8cbe..c37718801 100644 --- a/src/fastmcp/server/auth/providers/bearer.py +++ b/src/fastmcp/server/auth/providers/bearer.py @@ -1,14 +1,14 @@ """Backwards compatibility shim for BearerAuthProvider. -The BearerAuthProvider class has been moved to fastmcp.server.auth.verifiers.JWTVerifier +The BearerAuthProvider class has been moved to fastmcp.server.auth.providers.jwt.JWTVerifier for better organization. This module provides a backwards-compatible import. """ import warnings import fastmcp -from fastmcp.server.auth.verifiers import JWKData, JWKSData, RSAKeyPair -from fastmcp.server.auth.verifiers import JWTVerifier as BearerAuthProvider +from fastmcp.server.auth.providers.jwt import JWKData, JWKSData, RSAKeyPair +from fastmcp.server.auth.providers.jwt import JWTVerifier as BearerAuthProvider # Re-export for backwards compatibility __all__ = ["BearerAuthProvider", "RSAKeyPair", "JWKData", "JWKSData"] @@ -18,7 +18,7 @@ if fastmcp.settings.deprecation_warnings: warnings.warn( "The `fastmcp.server.auth.providers.bearer` module is deprecated " "and will be removed in a future version. " - "Please use `fastmcp.server.auth.verifiers.JWTVerifier` " + "Please use `fastmcp.server.auth.providers.jwt.JWTVerifier` " "instead of this module's BearerAuthProvider.", DeprecationWarning, stacklevel=2, diff --git a/src/fastmcp/server/auth/providers/in_memory.py b/src/fastmcp/server/auth/providers/in_memory.py index 533813522..56225bfd2 100644 --- a/src/fastmcp/server/auth/providers/in_memory.py +++ b/src/fastmcp/server/auth/providers/in_memory.py @@ -36,7 +36,7 @@ class InMemoryOAuthProvider(OAuthProvider): def __init__( self, - issuer_url: AnyHttpUrl | str | None = None, + base_url: AnyHttpUrl | str | None = None, service_documentation_url: AnyHttpUrl | str | None = None, client_registration_options: ClientRegistrationOptions | None = None, revocation_options: RevocationOptions | None = None, @@ -44,7 +44,7 @@ class InMemoryOAuthProvider(OAuthProvider): resource_server_url: AnyHttpUrl | str | None = None, ): super().__init__( - issuer_url=issuer_url or "http://fastmcp.example.com", + base_url=base_url or "http://fastmcp.example.com", service_documentation_url=service_documentation_url, client_registration_options=client_registration_options, revocation_options=revocation_options, diff --git a/src/fastmcp/server/auth/verifiers.py b/src/fastmcp/server/auth/providers/jwt.py similarity index 68% rename from src/fastmcp/server/auth/verifiers.py rename to src/fastmcp/server/auth/providers/jwt.py index 5def783d8..30cfb1ded 100644 --- a/src/fastmcp/server/auth/verifiers.py +++ b/src/fastmcp/server/auth/providers/jwt.py @@ -4,7 +4,7 @@ from __future__ import annotations import time from dataclasses import dataclass -from typing import Any +from typing import Any, cast import httpx from authlib.jose import JsonWebKey, JsonWebToken @@ -12,11 +12,12 @@ from authlib.jose.errors import JoseError from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from mcp.server.auth.provider import AccessToken -from pydantic import AnyHttpUrl, SecretStr, ValidationError +from pydantic import AnyHttpUrl, SecretStr from pydantic_settings import BaseSettings, SettingsConfigDict from typing_extensions import TypedDict from fastmcp.server.auth.auth import TokenVerifier +from fastmcp.server.auth.registry import register_provider from fastmcp.utilities.logging import get_logger from fastmcp.utilities.types import NotSet, NotSetT @@ -136,13 +137,29 @@ class RSAKeyPair: token_bytes = jwt_lib.encode( header, payload, self.private_key.get_secret_value() ) - return ( - token_bytes.decode("utf-8") - if isinstance(token_bytes, bytes) - else token_bytes - ) + + return token_bytes.decode("utf-8") +class JWTVerifierSettings(BaseSettings): + """Settings for JWT token verification.""" + + model_config = SettingsConfigDict( + env_prefix="FASTMCP_SERVER_AUTH_JWT_", + env_file=".env", + extra="ignore", + ) + + public_key: str | None = None + jwks_uri: str | None = None + issuer: str | None = None + algorithm: str | None = None + audience: str | list[str] | None = None + required_scopes: list[str] | None = None + resource_server_url: AnyHttpUrl | str | None = None + + +@register_provider("JWT") class JWTVerifier(TokenVerifier): """ JWT token verifier using public key or JWKS. @@ -161,13 +178,14 @@ class JWTVerifier(TokenVerifier): def __init__( self, - public_key: str | None = None, - jwks_uri: str | None = None, - issuer: str | None = None, - audience: str | list[str] | None = None, - algorithm: str | None = None, - required_scopes: list[str] | None = None, - resource_server_url: AnyHttpUrl | str | None = None, + *, + public_key: str | None | NotSetT = NotSet, + jwks_uri: str | None | NotSetT = NotSet, + issuer: str | None | NotSetT = NotSet, + audience: str | list[str] | None | NotSetT = NotSet, + algorithm: str | None | NotSetT = NotSet, + required_scopes: list[str] | None | NotSetT = NotSet, + resource_server_url: AnyHttpUrl | str | None | NotSetT = NotSet, ): """ Initialize the JWT token verifier. @@ -181,14 +199,29 @@ class JWTVerifier(TokenVerifier): required_scopes: Required scopes for all tokens resource_server_url: Resource server URL for TokenVerifier protocol """ - if not public_key and not jwks_uri: + settings = JWTVerifierSettings.model_validate( + { + k: v + for k, v in { + "public_key": public_key, + "jwks_uri": jwks_uri, + "issuer": issuer, + "audience": audience, + "algorithm": algorithm, + "required_scopes": required_scopes, + "resource_server_url": resource_server_url, + }.items() + if v is not NotSet + } + ) + + if not settings.public_key and not settings.jwks_uri: raise ValueError("Either public_key or jwks_uri must be provided") - if public_key and jwks_uri: + if settings.public_key and settings.jwks_uri: raise ValueError("Provide either public_key or jwks_uri, not both") - if not algorithm: - algorithm = "RS256" + algorithm = settings.algorithm or "RS256" if algorithm not in { "HS256", "HS384", @@ -207,14 +240,15 @@ class JWTVerifier(TokenVerifier): # Initialize parent TokenVerifier super().__init__( - resource_server_url=resource_server_url, required_scopes=required_scopes + resource_server_url=settings.resource_server_url, + required_scopes=settings.required_scopes, ) self.algorithm = algorithm - self.issuer = issuer - self.audience = audience - self.public_key = public_key - self.jwks_uri = jwks_uri + self.issuer = settings.issuer + self.audience = settings.audience + self.public_key = settings.public_key + self.jwks_uri = settings.jwks_uri self.jwt = JsonWebToken([self.algorithm]) self.logger = get_logger(__name__) @@ -377,7 +411,7 @@ class JWTVerifier(TokenVerifier): ) else: # aud is a string - check if it's in our expected list - audience_valid = aud in self.audience + audience_valid = aud in cast(list, self.audience) else: # self.audience is a string - use original logic if isinstance(aud, list): @@ -439,220 +473,6 @@ class JWTVerifier(TokenVerifier): return await self.load_access_token(token) -class JWTVerifierSettings(BaseSettings): - """Settings for the BearerAuthProvider.""" - - model_config = SettingsConfigDict( - env_prefix="FASTMCP_AUTH_JWT_", - env_file=".env", - extra="ignore", - ) - - public_key: str | None = None - jwks_uri: str | None = None - issuer: str | None = None - algorithm: str | None = None - audience: str | None = None - required_scopes: list[str] | None = None - resource_server_url: AnyHttpUrl | str | None = None - - -class EnvJWTVerifier(JWTVerifier): - def __init__( - self, - public_key: str | None | NotSetT = NotSet, - jwks_uri: str | None | NotSetT = NotSet, - issuer: str | None | NotSetT = NotSet, - audience: str | list[str] | None | NotSetT = NotSet, - algorithm: str | None | NotSetT = NotSet, - required_scopes: list[str] | None | NotSetT = NotSet, - resource_server_url: AnyHttpUrl | str | None | NotSetT = NotSet, - ): - kwargs = { - "public_key": public_key, - "jwks_uri": jwks_uri, - "issuer": issuer, - "algorithm": algorithm, - "audience": audience, - "required_scopes": required_scopes, - "resource_server_url": resource_server_url, - } - settings = JWTVerifierSettings( - **{k: v for k, v in kwargs.items() if v is not NotSet} - ) - super().__init__(**settings.model_dump()) - - -class IntrospectionTokenVerifier(TokenVerifier): - """ - OAuth 2.0 Token Introspection verifier (RFC 7662). - - This verifier validates tokens by making real-time calls to an OAuth 2.0 - authorization server's introspection endpoint. Unlike JWT verification, this - approach works with both opaque tokens and JWTs, and provides real-time - validation including immediate revocation support. - - Use this when: - - Your authorization server is separate from your FastMCP server - - You're using opaque (non-JWT) tokens - - You need real-time token validation and revocation support - - Your authorization server supports RFC 7662 introspection - - You want centralized token management without sharing secrets - """ - - def __init__( - self, - introspection_endpoint: AnyHttpUrl | str, - server_url: AnyHttpUrl | str, - client_id: str | None = None, - client_secret: str | None = None, - validate_resource: bool = False, - required_scopes: list[str] | None = None, - timeout: float = 10.0, - ): - """ - Initialize the introspection token verifier. - - Args: - introspection_endpoint: OAuth 2.0 introspection endpoint URL - server_url: This server's URL for resource validation - client_id: Client ID for introspection authentication - client_secret: Client secret for introspection authentication - validate_resource: Whether to validate RFC 8707 resource parameter - required_scopes: Required scopes for all tokens - timeout: HTTP request timeout in seconds - """ - try: - self.introspection_endpoint = AnyHttpUrl(introspection_endpoint) - server_url_validated = AnyHttpUrl(server_url) - except ValidationError as e: - raise ValueError(f"Invalid URL provided: {e}") from e - - # Basic SSRF protection - reject private/localhost URLs - if self._is_private_url(str(self.introspection_endpoint)): - raise ValueError("Introspection endpoint cannot be a private/localhost URL") - - # Initialize parent TokenVerifier with the resource server URL - super().__init__( - resource_server_url=server_url_validated, required_scopes=required_scopes - ) - - self.client_id = client_id - self.client_secret = client_secret - self.validate_resource = validate_resource - self.timeout = timeout - - # Create HTTP client with security settings - self._client = httpx.AsyncClient( - timeout=timeout, - verify=True, # Always verify SSL - limits=httpx.Limits(max_connections=10, max_keepalive_connections=5), - ) - - @property - def server_url(self) -> AnyHttpUrl: - """The resource server URL for this verifier.""" - if self.resource_server_url is None: - raise ValueError("Resource server URL not set") - return self.resource_server_url - - def _is_private_url(self, url: str) -> bool: - """Check if URL points to private/localhost addresses (basic SSRF protection).""" - import ipaddress - from urllib.parse import urlparse - - parsed = urlparse(url) - hostname = parsed.hostname - - if not hostname: - return False - - # Check for localhost - if hostname.lower() in ("localhost", "127.0.0.1", "::1"): - return True - - # Check for private IP ranges - try: - ip = ipaddress.ip_address(hostname) - return ip.is_private or ip.is_loopback - except ValueError: - # Not an IP address, assume it's a hostname - return False - - async def verify_token(self, token: str) -> AccessToken | None: - """Verify token using OAuth 2.0 introspection.""" - try: - # Prepare introspection request - data = {"token": token} - - # Add resource parameter if validation is enabled (RFC 8707) - if self.validate_resource and self.resource_server_url: - data["resource"] = str(self.resource_server_url) - - # Prepare authentication and make introspection request - if self.client_id and self.client_secret: - response = await self._client.post( - str(self.introspection_endpoint), - data=data, - auth=(self.client_id, self.client_secret), - headers={"Content-Type": "application/x-www-form-urlencoded"}, - ) - else: - response = await self._client.post( - str(self.introspection_endpoint), - data=data, - headers={"Content-Type": "application/x-www-form-urlencoded"}, - ) - response.raise_for_status() - - introspection_response = response.json() - - # Check if token is active - if not introspection_response.get("active", False): - return None - - # Extract token information - client_id = introspection_response.get("client_id", "unknown") - scopes = ( - introspection_response.get("scope", "").split() - if introspection_response.get("scope") - else [] - ) - exp = introspection_response.get("exp") - - # Check required scopes - if self.required_scopes: - token_scopes = set(scopes) - required_scopes = set(self.required_scopes) - if not required_scopes.issubset(token_scopes): - logger.debug( - f"Token missing required scopes. Has: {token_scopes}, Required: {required_scopes}" - ) - return None - - return AccessToken( - token=token, - client_id=client_id, - scopes=scopes, - expires_at=exp, - resource=str(self.resource_server_url) - if self.resource_server_url - else None, - ) - - except Exception as e: - logger.debug(f"Introspection verification failed: {e}") - return None - - async def __aenter__(self): - """Async context manager entry.""" - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - """Async context manager exit.""" - await self._client.aclose() - - class StaticTokenVerifier(TokenVerifier): """ Simple static token verifier for testing and development. diff --git a/src/fastmcp/server/auth/providers/workos.py b/src/fastmcp/server/auth/providers/workos.py new file mode 100644 index 000000000..7a71e4550 --- /dev/null +++ b/src/fastmcp/server/auth/providers/workos.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +import httpx +from mcp.server.auth.provider import ( + AccessToken, +) +from pydantic import AnyHttpUrl +from pydantic_settings import BaseSettings, SettingsConfigDict +from starlette.responses import JSONResponse +from starlette.routing import BaseRoute, Route + +from fastmcp.server.auth.auth import AuthProvider, TokenVerifier +from fastmcp.server.auth.providers.jwt import JWTVerifier +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 AuthKitProviderSettings(BaseSettings): + model_config = SettingsConfigDict( + env_prefix="FASTMCP_SERVER_AUTH_AUTHKITPROVIDER_", + env_file=".env", + extra="ignore", + ) + + authkit_domain: AnyHttpUrl + base_url: AnyHttpUrl + required_scopes: list[str] | None = None + + +@register_provider("AUTHKIT") +class AuthKitProvider(AuthProvider): + """WorkOS AuthKit metadata provider for DCR (Dynamic Client Registration). + + This provider implements WorkOS AuthKit integration using metadata forwarding + instead of OAuth proxying. This is the recommended approach for WorkOS DCR + as it allows WorkOS to handle the OAuth flow directly while FastMCP acts + as a resource server. + + IMPORTANT SETUP REQUIREMENTS: + + 1. Enable Dynamic Client Registration in WorkOS Dashboard: + - Go to Applications → Configuration + - Toggle "Dynamic Client Registration" to enabled + + 2. Configure your FastMCP server URL as a callback: + - Add your server URL to the Redirects tab in WorkOS dashboard + - Example: https://your-fastmcp-server.com/oauth2/callback + + For detailed setup instructions, see: + https://workos.com/docs/authkit/mcp/integrating/token-verification + + Example: + ```python + from fastmcp.server.auth.providers.workos import AuthKitProvider + + # Create WorkOS metadata provider (JWT verifier created automatically) + workos_auth = AuthKitProvider( + authkit_domain="https://your-workos-domain.authkit.app", + base_url="https://your-fastmcp-server.com", + ) + + # Use with FastMCP + mcp = FastMCP("My App", auth=workos_auth) + ``` + """ + + def __init__( + self, + *, + authkit_domain: AnyHttpUrl | str | NotSetT = NotSet, + base_url: AnyHttpUrl | str | NotSetT = NotSet, + required_scopes: list[str] | None | NotSetT = NotSet, + token_verifier: TokenVerifier | None = None, + ): + """Initialize WorkOS metadata provider. + + Args: + authkit_domain: Your WorkOS AuthKit domain (e.g., "https://your-app.authkit.app") + base_url: Public URL of this FastMCP server + required_scopes: Optional list of scopes to require for all requests + token_verifier: Optional token verifier. If None, creates JWT verifier for WorkOS + """ + super().__init__() + + settings = AuthKitProviderSettings.model_validate( + { + k: v + for k, v in { + "authkit_domain": authkit_domain, + "base_url": base_url, + "required_scopes": required_scopes, + }.items() + if v is not NotSet + } + ) + + self.authkit_domain = str(settings.authkit_domain).rstrip("/") + self.base_url = str(settings.base_url).rstrip("/") + + # Create default JWT verifier if none provided + if token_verifier is None: + token_verifier = JWTVerifier( + jwks_uri=f"{self.authkit_domain}/oauth2/jwks", + issuer=self.authkit_domain, + algorithm="RS256", + required_scopes=settings.required_scopes, + ) + + self.token_verifier = token_verifier + + async def verify_token(self, token: str) -> AccessToken | None: + """Verify a WorkOS token using the configured token verifier.""" + return await self.token_verifier.verify_token(token) + + def customize_auth_routes(self, routes: list[BaseRoute]) -> list[BaseRoute]: + """Add AuthKit metadata endpoints. + + This adds: + - /.well-known/oauth-authorization-server (forwards AuthKit metadata) + - /.well-known/oauth-protected-resource (returns FastMCP resource info) + """ + + async def oauth_authorization_server_metadata(request): + """Forward AuthKit OAuth authorization server metadata with FastMCP customizations.""" + try: + async with httpx.AsyncClient() as client: + response = await client.get( + f"{self.authkit_domain}/.well-known/oauth-authorization-server" + ) + 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 AuthKit metadata: {e}", + }, + status_code=500, + ) + + async def oauth_protected_resource_metadata(request): + """Return FastMCP resource server metadata.""" + return JSONResponse( + { + "resource": self.base_url, + "authorization_servers": [self.authkit_domain], + "bearer_methods_supported": ["header"], + } + ) + + routes.extend( + [ + Route( + "/.well-known/oauth-authorization-server", + endpoint=oauth_authorization_server_metadata, + methods=["GET"], + ), + Route( + "/.well-known/oauth-protected-resource", + endpoint=oauth_protected_resource_metadata, + methods=["GET"], + ), + ] + ) + + return routes diff --git a/src/fastmcp/server/auth/registry.py b/src/fastmcp/server/auth/registry.py new file mode 100644 index 000000000..9cbf9eb5e --- /dev/null +++ b/src/fastmcp/server/auth/registry.py @@ -0,0 +1,52 @@ +"""Provider registry for FastMCP auth providers.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING, TypeVar + +if TYPE_CHECKING: + from fastmcp.server.auth.auth import AuthProvider + +# Type variable for auth providers +T = TypeVar("T", bound="AuthProvider") + + +# Provider Registry +_PROVIDER_REGISTRY: dict[str, type[AuthProvider]] = {} + + +def register_provider(name: str) -> Callable[[type[T]], type[T]]: + """Decorator to register an auth provider with a given name. + + Args: + name: The name to register the provider under (e.g., 'AUTHKIT') + + Returns: + The decorated class + + Example: + @register_provider('AUTHKIT') + class AuthKitProvider(AuthProvider): + ... + """ + + def decorator(cls: type[T]) -> type[T]: + _PROVIDER_REGISTRY[name.upper()] = cls + return cls + + return decorator + + +def get_registered_provider(name: str) -> type[AuthProvider]: + """Get a registered provider by name. + + Args: + name: The provider name (case-insensitive) + + Returns: + The provider class if found, None otherwise + """ + if name.upper() in _PROVIDER_REGISTRY: + return _PROVIDER_REGISTRY[name.upper()] + raise ValueError(f"Provider {name!r} has not been registered.") diff --git a/src/fastmcp/server/http.py b/src/fastmcp/server/http.py index c3821a4c2..2192ac459 100644 --- a/src/fastmcp/server/http.py +++ b/src/fastmcp/server/http.py @@ -25,7 +25,7 @@ from starlette.responses import Response from starlette.routing import BaseRoute, Mount, Route from starlette.types import Lifespan, Receive, Scope, Send -from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier +from fastmcp.server.auth.auth import AuthProvider, OAuthProvider, TokenVerifier from fastmcp.utilities.logging import get_logger if TYPE_CHECKING: @@ -72,12 +72,12 @@ class RequestContextMiddleware: def setup_auth_middleware_and_routes( - auth: OAuthProvider | TokenVerifier, -) -> tuple[list[Middleware], list[BaseRoute], list[str]]: + auth: AuthProvider, +) -> tuple[list[Middleware], list[Route], list[str]]: """Set up authentication middleware and routes if auth is enabled. Args: - auth: Either an OAuthProvider or TokenVerifier for authentication + auth: An AuthProvider for authentication (TokenVerifier or OAuthProvider) Returns: Tuple of (middleware, auth_routes, required_scopes) @@ -90,27 +90,28 @@ def setup_auth_middleware_and_routes( Middleware(AuthContextMiddleware), ] - auth_routes: list[BaseRoute] = [] - required_scopes: list[str] = [] + auth_routes: list[Route] = [] + required_scopes: list[str] = auth.required_scopes or [] - # Handle TokenVerifier vs OAuthProvider - # Check if it's an OAuthProvider by looking for issuer_url attribute - if hasattr(auth, "issuer_url"): - # OAuthProvider: create auth routes and get required scopes - # We know this is an OAuthProvider because it has issuer_url - auth_routes = list( + # Check if it's an OAuthProvider (has OAuth server capability) + if isinstance(auth, OAuthProvider): + # OAuthProvider: create standard OAuth routes first + standard_routes = list( create_auth_routes( - provider=auth, # type: ignore[arg-type] - issuer_url=auth.issuer_url, # type: ignore[attr-defined] - service_documentation_url=auth.service_documentation_url, # type: ignore[attr-defined] - client_registration_options=auth.client_registration_options, # type: ignore[attr-defined] - revocation_options=auth.revocation_options, # type: ignore[attr-defined] + provider=auth, + issuer_url=auth.issuer_url, + service_documentation_url=auth.service_documentation_url, + client_registration_options=auth.client_registration_options, + revocation_options=auth.revocation_options, ) ) - required_scopes = auth.required_scopes or [] # type: ignore[attr-defined] + + # Allow provider to customize routes (e.g., for proxy behavior or metadata endpoints) + auth_routes = auth.customize_auth_routes(standard_routes) else: - # TokenVerifier: no auth routes but may have required scopes - required_scopes = getattr(auth, "required_scopes", None) or [] + # Simple AuthProvider or TokenVerifier: start with empty routes + # Allow provider to add custom routes (e.g., metadata endpoints) + auth_routes = auth.customize_auth_routes([]) return middleware, auth_routes, required_scopes @@ -147,7 +148,7 @@ def create_sse_app( server: FastMCP[LifespanResultT], message_path: str, sse_path: str, - auth: OAuthProvider | TokenVerifier | None = None, + auth: AuthProvider | None = None, debug: bool = False, routes: list[BaseRoute] | None = None, middleware: list[Middleware] | None = None, @@ -158,7 +159,7 @@ def create_sse_app( server: The FastMCP server instance message_path: Path for SSE messages sse_path: Path for SSE connections - auth: Optional authentication provider (OAuthProvider or TokenVerifier) + auth: Optional authentication provider (AuthProvider) debug: Whether to enable debug mode routes: Optional list of custom routes middleware: Optional list of middleware @@ -263,7 +264,7 @@ def create_streamable_http_app( server: FastMCP[LifespanResultT], streamable_http_path: str, event_store: EventStore | None = None, - auth: OAuthProvider | TokenVerifier | None = None, + auth: AuthProvider | None = None, json_response: bool = False, stateless_http: bool = False, debug: bool = False, @@ -276,7 +277,7 @@ def create_streamable_http_app( server: The FastMCP server instance streamable_http_path: Path for StreamableHTTP connections event_store: Optional event store for session management - auth: Optional authentication provider (OAuthProvider or TokenVerifier) + auth: Optional authentication provider (AuthProvider) json_response: Whether to use JSON response format stateless_http: Whether to use stateless mode (new transport per request) debug: Whether to enable debug mode diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index ff461e351..3fd9fa9eb 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -49,8 +49,8 @@ from fastmcp.prompts import Prompt, PromptManager from fastmcp.prompts.prompt import FunctionPrompt from fastmcp.resources import Resource, ResourceManager from fastmcp.resources.template import ResourceTemplate -from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier -from fastmcp.server.auth.verifiers import EnvJWTVerifier +from fastmcp.server.auth.auth import AuthProvider +from fastmcp.server.auth.registry import get_registered_provider from fastmcp.server.http import ( StarletteWithLifespan, create_sse_app, @@ -131,7 +131,7 @@ class FastMCP(Generic[LifespanResultT]): instructions: str | None = None, *, version: str | None = None, - auth: OAuthProvider | TokenVerifier | None = None, + auth: AuthProvider | None | NotSetT = NotSet, middleware: list[Middleware] | None = None, lifespan: ( Callable[ @@ -200,10 +200,14 @@ class FastMCP(Generic[LifespanResultT]): lifespan=_lifespan_wrapper(self, lifespan), ) - if auth is None and fastmcp.settings.default_auth_provider == "jwt-env": - auth = EnvJWTVerifier() - - self.auth = auth + # if auth is `NotSet`, try to create a provider from the environment + if auth is NotSet: + if fastmcp.settings.server_auth is not None: + provider_cls = get_registered_provider(fastmcp.settings.server_auth) + auth = provider_cls() + else: + auth = None + self.auth = cast(AuthProvider | None, auth) if tools: for tool in tools: diff --git a/src/fastmcp/settings.py b/src/fastmcp/settings.py index a24f2afaa..76f70e36f 100644 --- a/src/fastmcp/settings.py +++ b/src/fastmcp/settings.py @@ -259,19 +259,23 @@ class Settings(BaseSettings): ) # Auth settings - default_auth_provider: Annotated[ - Literal["jwt-env"] | None, + server_auth: Annotated[ + str | None, Field( description=inspect.cleandoc( """ - Configure the authentication provider. This setting is intended only to - be used for remote confirugation of providers that fully support - environment variable configuration. + Configure the authentication provider for the server. Auth + providers are registered with a specific key, and providing that + key here will cause the server to automatically configure the + provider from the environment. If None, no automatic configuration will take place. This setting is *always* overriden by any auth provider passed to the FastMCP constructor. + + Note that most auth providers require additional configuration + that must be provided via env vars. """ ), ), diff --git a/src/fastmcp/utilities/tests.py b/src/fastmcp/utilities/tests.py index 78b973b7d..4a2e6061e 100644 --- a/src/fastmcp/utilities/tests.py +++ b/src/fastmcp/utilities/tests.py @@ -8,10 +8,13 @@ import time from collections.abc import Callable, Generator from contextlib import contextmanager from typing import TYPE_CHECKING, Any, Literal +from urllib.parse import parse_qs, urlparse +import httpx import uvicorn from fastmcp import settings +from fastmcp.client.auth.oauth import OAuth from fastmcp.utilities.http import find_available_port if TYPE_CHECKING: @@ -139,3 +142,51 @@ def caplog_for_fastmcp(caplog): yield finally: logger.removeHandler(caplog.handler) + + +class HeadlessOAuth(OAuth): + """ + OAuth provider that bypasses browser interaction for testing. + + This simulates the complete OAuth flow programmatically by making HTTP requests + instead of opening a browser and running a callback server. Useful for automated testing. + """ + + def __init__(self, mcp_url: str, **kwargs): + """Initialize HeadlessOAuth with stored response tracking.""" + self._stored_response = None + super().__init__(mcp_url, **kwargs) + + async def redirect_handler(self, authorization_url: str) -> None: + """Make HTTP request to authorization URL and store response for callback handler.""" + async with httpx.AsyncClient() as client: + response = await client.get(authorization_url, follow_redirects=False) + self._stored_response = response + + async def callback_handler(self) -> tuple[str, str | None]: + """Parse stored response and return (auth_code, state).""" + if not self._stored_response: + raise RuntimeError( + "No authorization response stored. redirect_handler must be called first." + ) + + response = self._stored_response + + # Extract auth code from redirect location + if response.status_code == 302: + redirect_url = response.headers["location"] + parsed = urlparse(redirect_url) + query_params = parse_qs(parsed.query) + + if "error" in query_params: + error = query_params["error"][0] + error_desc = query_params.get("error_description", ["Unknown error"])[0] + raise RuntimeError( + f"OAuth authorization failed: {error} - {error_desc}" + ) + + auth_code = query_params["code"][0] + state = query_params.get("state", [None])[0] + return auth_code, state + else: + raise RuntimeError(f"Authorization failed: {response.status_code}") diff --git a/tests/auth/providers/test_token_verifier.py b/tests/auth/providers/test_token_verifier.py deleted file mode 100644 index ed4b598c6..000000000 --- a/tests/auth/providers/test_token_verifier.py +++ /dev/null @@ -1,175 +0,0 @@ -"""Tests for TokenVerifier protocol implementation in auth providers.""" - -import pytest -from mcp.server.auth.provider import AccessToken - -from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider -from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair - - -class TestJWTVerifierTokenVerifier: - """Test that JWTVerifier implements TokenVerifier protocol correctly.""" - - @pytest.fixture - def rsa_key_pair(self) -> RSAKeyPair: - """Generate RSA key pair for testing.""" - return RSAKeyPair.generate() - - @pytest.fixture - def jwt_verifier(self, rsa_key_pair: RSAKeyPair) -> JWTVerifier: - """Create JWTVerifier for testing.""" - return JWTVerifier( - public_key=rsa_key_pair.public_key, - issuer="https://test.example.com", - audience="https://api.example.com", - ) - - @pytest.fixture - def valid_token(self, rsa_key_pair: RSAKeyPair) -> str: - """Create a valid test token.""" - return rsa_key_pair.create_token( - subject="test-user", - issuer="https://test.example.com", - audience="https://api.example.com", - scopes=["read", "write"], - ) - - @pytest.fixture - def expired_token(self, rsa_key_pair: RSAKeyPair) -> str: - """Create an expired test token.""" - return rsa_key_pair.create_token( - subject="test-user", - issuer="https://test.example.com", - audience="https://api.example.com", - expires_in_seconds=-3600, # Expired 1 hour ago - ) - - async def test_verify_token_with_valid_token( - self, jwt_verifier: JWTVerifier, valid_token: str - ): - """Test that verify_token returns AccessToken for valid token.""" - result = await jwt_verifier.verify_token(valid_token) - - assert result is not None - assert isinstance(result, AccessToken) - assert result.token == valid_token - assert result.client_id == "test-user" - assert "read" in result.scopes - assert "write" in result.scopes - - async def test_verify_token_with_expired_token( - self, jwt_verifier: JWTVerifier, expired_token: str - ): - """Test that verify_token returns None for expired token.""" - result = await jwt_verifier.verify_token(expired_token) - assert result is None - - async def test_verify_token_with_invalid_token(self, jwt_verifier: JWTVerifier): - """Test that verify_token returns None for invalid token.""" - result = await jwt_verifier.verify_token("invalid.token.here") - assert result is None - - async def test_verify_token_with_malformed_token(self, jwt_verifier: JWTVerifier): - """Test that verify_token returns None for malformed token.""" - result = await jwt_verifier.verify_token("not-a-jwt") - assert result is None - - async def test_verify_token_delegation_to_load_access_token( - self, jwt_verifier: JWTVerifier, valid_token: str - ): - """Test that verify_token delegates to load_access_token.""" - # Both methods should return the same result - verify_result = await jwt_verifier.verify_token(valid_token) - load_result = await jwt_verifier.load_access_token(valid_token) - - assert verify_result == load_result - if verify_result is not None and load_result is not None: - assert verify_result.token == load_result.token - assert verify_result.client_id == load_result.client_id - assert verify_result.scopes == load_result.scopes - - -class TestInMemoryOAuthProviderTokenVerifier: - """Test that InMemoryOAuthProvider implements TokenVerifier protocol correctly.""" - - @pytest.fixture - def in_memory_provider(self) -> InMemoryOAuthProvider: - """Create InMemoryOAuthProvider for testing.""" - return InMemoryOAuthProvider( - issuer_url="https://test.example.com", - required_scopes=["user"], - ) - - async def test_verify_token_with_nonexistent_token( - self, in_memory_provider: InMemoryOAuthProvider - ): - """Test that verify_token returns None for nonexistent token.""" - result = await in_memory_provider.verify_token("nonexistent-token") - assert result is None - - async def test_verify_token_delegation_to_load_access_token( - self, in_memory_provider: InMemoryOAuthProvider - ): - """Test that verify_token delegates to load_access_token.""" - # Create a test token in the provider's storage - test_token = "test-access-token" - test_access_token = AccessToken( - token=test_token, - client_id="test-client", - scopes=["user"], - expires_at=None, # No expiry - ) - in_memory_provider.access_tokens[test_token] = test_access_token - - # Both methods should return the same result - verify_result = await in_memory_provider.verify_token(test_token) - load_result = await in_memory_provider.load_access_token(test_token) - - assert verify_result == load_result - assert verify_result is not None - assert verify_result.token == test_token - assert verify_result.client_id == "test-client" - assert verify_result.scopes == ["user"] - - async def test_verify_token_with_expired_token( - self, in_memory_provider: InMemoryOAuthProvider - ): - """Test that verify_token returns None for expired token.""" - import time - - # Create an expired token - expired_token = "expired-token" - expired_access_token = AccessToken( - token=expired_token, - client_id="test-client", - scopes=["user"], - expires_at=int(time.time()) - 3600, # Expired 1 hour ago - ) - in_memory_provider.access_tokens[expired_token] = expired_access_token - - result = await in_memory_provider.verify_token(expired_token) - assert result is None - - # Token should be cleaned up from storage - assert expired_token not in in_memory_provider.access_tokens - - -class TestTokenVerifierProtocolCompliance: - """Test that our providers properly implement the TokenVerifier protocol.""" - - async def test_bearer_provider_implements_protocol(self): - """Test that JWTVerifier can be used as TokenVerifier.""" - key_pair = RSAKeyPair.generate() - provider = JWTVerifier(public_key=key_pair.public_key) - - # Should have the required method for TokenVerifier protocol - assert hasattr(provider, "verify_token") - assert callable(provider.verify_token) - - async def test_in_memory_provider_implements_protocol(self): - """Test that InMemoryOAuthProvider can be used as TokenVerifier.""" - provider = InMemoryOAuthProvider() - - # Should have the required method for TokenVerifier protocol - assert hasattr(provider, "verify_token") - assert callable(provider.verify_token) diff --git a/tests/auth/test_oauth_client.py b/tests/auth/test_oauth_client.py deleted file mode 100644 index c12f80a1d..000000000 --- a/tests/auth/test_oauth_client.py +++ /dev/null @@ -1,267 +0,0 @@ -from collections.abc import Generator -from unittest.mock import patch -from urllib.parse import parse_qs, urlparse - -import httpx -import pytest - -import fastmcp.client.auth.oauth # Import module, not the function directly -from fastmcp.client import Client -from fastmcp.client.transports import StreamableHttpTransport -from fastmcp.server.auth.auth import ClientRegistrationOptions -from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider -from fastmcp.server.server import FastMCP -from fastmcp.utilities.tests import run_server_in_process - - -def fastmcp_server(issuer_url: str): - """Create a FastMCP server with OAuth authentication.""" - server = FastMCP( - "TestServer", - auth=InMemoryOAuthProvider( - issuer_url=issuer_url, - client_registration_options=ClientRegistrationOptions(enabled=True), - ), - ) - - @server.tool - def add(a: int, b: int) -> int: - """Add two numbers together.""" - return a + b - - @server.resource("resource://test") - def get_test_resource() -> str: - """Get a test resource.""" - return "Hello from authenticated resource!" - - return server - - -def run_server(host: str, port: int, **kwargs) -> None: - fastmcp_server(f"http://{host}:{port}").run(host=host, port=port, **kwargs) - - -@pytest.fixture(scope="module") -def streamable_http_server() -> Generator[str, None, None]: - with run_server_in_process(run_server, transport="http") as url: - yield f"{url}/mcp/" - - -@pytest.fixture() -def client_unauthorized(streamable_http_server: str) -> Client: - return Client(transport=StreamableHttpTransport(streamable_http_server)) - - -class HeadlessOAuthProvider(httpx.Auth): - """ - OAuth provider that bypasses browser interaction for testing. - - This simulates the complete OAuth flow programmatically by: - 1. Discovering OAuth metadata from the server - 2. Registering a client - 3. Getting an authorization code (simulates user approval) - 4. Exchanging it for an access token - 5. Adding Bearer token to all requests - - This enables testing OAuth-protected FastMCP servers without - requiring browser interaction or external OAuth providers. - """ - - def __init__(self, mcp_url: str): - self.mcp_url = mcp_url - parsed_url = urlparse(mcp_url) - self.server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" - self._access_token = None - - async def async_auth_flow(self, request): - """httpx.Auth interface - add Bearer token to requests.""" - if not self._access_token: - await self._obtain_token() - - if self._access_token: - request.headers["Authorization"] = f"Bearer {self._access_token}" - - yield request - - async def _obtain_token(self): - """Get a valid access token by simulating the OAuth flow.""" - import base64 - import hashlib - import secrets - - from mcp.shared.auth import OAuthClientInformationFull - from pydantic import AnyHttpUrl - - # Generate PKCE challenge/verifier - code_verifier = ( - base64.urlsafe_b64encode(secrets.token_bytes(32)).decode().rstrip("=") - ) - code_challenge = ( - base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()) - .decode() - .rstrip("=") - ) - - # Create HTTP client to talk to the server - async with httpx.AsyncClient() as http_client: - # 1. Discover OAuth metadata - metadata_url = ( - f"{self.server_base_url}/.well-known/oauth-authorization-server" - ) - response = await http_client.get(metadata_url) - response.raise_for_status() - metadata = response.json() - - # 2. Register a client - client_info = OAuthClientInformationFull( - client_id="test_client_headless", - client_secret="test_secret_headless", - redirect_uris=[AnyHttpUrl("http://localhost:8080/callback")], - ) - - register_response = await http_client.post( - metadata["registration_endpoint"], - json=client_info.model_dump(mode="json"), - ) - register_response.raise_for_status() - registered_client = register_response.json() - - # 3. Get authorization code (simulate user approval) - auth_params = { - "response_type": "code", - "client_id": registered_client["client_id"], - "redirect_uri": "http://localhost:8080/callback", - "code_challenge": code_challenge, - "code_challenge_method": "S256", - "state": "test_state_headless", - } - - auth_response = await http_client.get( - metadata["authorization_endpoint"], - params=auth_params, - follow_redirects=False, - ) - - # Extract auth code from redirect - if auth_response.status_code == 302: - redirect_url = auth_response.headers["location"] - parsed = urlparse(redirect_url) - query_params = parse_qs(parsed.query) - - if "error" in query_params: - error = query_params["error"][0] - error_desc = query_params.get( - "error_description", ["Unknown error"] - )[0] - raise RuntimeError( - f"OAuth authorization failed: {error} - {error_desc}" - ) - - auth_code = query_params["code"][0] - - # 4. Exchange auth code for access token - token_data = { - "grant_type": "authorization_code", - "client_id": registered_client["client_id"], - "client_secret": registered_client["client_secret"], - "code": auth_code, - "redirect_uri": "http://localhost:8080/callback", - "code_verifier": code_verifier, - } - - token_response = await http_client.post( - metadata["token_endpoint"], data=token_data - ) - token_response.raise_for_status() - token_info = token_response.json() - - self._access_token = token_info["access_token"] - else: - raise RuntimeError(f"Authorization failed: {auth_response.status_code}") - - -@pytest.fixture() -def client_with_headless_oauth( - streamable_http_server: str, -) -> Generator[Client, None, None]: - """Client with headless OAuth that bypasses browser interaction.""" - - # Patch the OAuth function to return our headless provider - def headless_oauth(*args, **kwargs): - mcp_url = args[0] if args else kwargs.get("mcp_url", "") - if not mcp_url: - raise ValueError("mcp_url is required") - return HeadlessOAuthProvider(mcp_url) - - with patch("fastmcp.client.auth.oauth.OAuth", side_effect=headless_oauth): - client = Client( - transport=StreamableHttpTransport(streamable_http_server), - auth=fastmcp.client.auth.oauth.OAuth(mcp_url=streamable_http_server), - ) - yield client - - -async def test_unauthorized(client_unauthorized: Client): - """Test that unauthenticated requests are rejected.""" - with pytest.raises(httpx.HTTPStatusError, match="401 Unauthorized"): - async with client_unauthorized: - pass - - -async def test_ping(client_with_headless_oauth: Client): - """Test that we can ping the server.""" - async with client_with_headless_oauth: - assert await client_with_headless_oauth.ping() - - -async def test_list_tools(client_with_headless_oauth: Client): - """Test that we can list tools.""" - async with client_with_headless_oauth: - tools = await client_with_headless_oauth.list_tools() - tool_names = [tool.name for tool in tools] - assert "add" in tool_names - - -async def test_call_tool(client_with_headless_oauth: Client): - """Test that we can call a tool.""" - async with client_with_headless_oauth: - result = await client_with_headless_oauth.call_tool("add", {"a": 5, "b": 3}) - # The add tool returns int which gets wrapped as structured output - # Client unwraps it and puts the actual int in the data field - assert result.data == 8 - - -async def test_list_resources(client_with_headless_oauth: Client): - """Test that we can list resources.""" - async with client_with_headless_oauth: - resources = await client_with_headless_oauth.list_resources() - resource_uris = [str(resource.uri) for resource in resources] - assert "resource://test" in resource_uris - - -async def test_read_resource(client_with_headless_oauth: Client): - """Test that we can read a resource.""" - async with client_with_headless_oauth: - resource = await client_with_headless_oauth.read_resource("resource://test") - assert resource[0].text == "Hello from authenticated resource!" # type: ignore[attr-defined] - - -async def test_oauth_server_metadata_discovery(streamable_http_server: str): - """Test that we can discover OAuth metadata from the running server.""" - parsed_url = urlparse(streamable_http_server) - server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" - - async with httpx.AsyncClient() as client: - # Test OAuth discovery endpoint - metadata_url = f"{server_base_url}/.well-known/oauth-authorization-server" - response = await client.get(metadata_url) - assert response.status_code == 200 - - metadata = response.json() - assert "authorization_endpoint" in metadata - assert "token_endpoint" in metadata - assert "registration_endpoint" in metadata - - # The endpoints should be properly formed URLs - assert metadata["authorization_endpoint"].startswith(server_base_url) - assert metadata["token_endpoint"].startswith(server_base_url) diff --git a/tests/auth/verifiers/test_env_jwt.py b/tests/auth/verifiers/test_env_jwt.py deleted file mode 100644 index d435dba47..000000000 --- a/tests/auth/verifiers/test_env_jwt.py +++ /dev/null @@ -1,89 +0,0 @@ -import pytest -from pydantic import ValidationError - -from fastmcp import FastMCP -from fastmcp.server.auth.verifiers import EnvJWTVerifier, JWTVerifier -from fastmcp.settings import Settings -from fastmcp.utilities.tests import temporary_settings - - -def test_load_bearer_env_from_env_var(monkeypatch): - mcp = FastMCP() - assert mcp.auth is None - - monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env") - monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key") - - with temporary_settings(**Settings().model_dump()): - mcp_with_auth = FastMCP() - assert isinstance(mcp_with_auth.auth, EnvJWTVerifier) - - -def test_load_bearer_env_from_env_var_requires_public_key_or_jwks_uri(monkeypatch): - mcp = FastMCP() - assert mcp.auth is None - - monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env") - - with temporary_settings(**Settings().model_dump()): - with pytest.raises( - ValueError, match="Either public_key or jwks_uri must be provided" - ): - FastMCP() - - -def test_configure_bearer_env_from_env_var(monkeypatch): - monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env") - monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key") - monkeypatch.setenv("FASTMCP_AUTH_JWT_ISSUER", "http://test-issuer") - monkeypatch.setenv("FASTMCP_AUTH_JWT_AUDIENCE", "test-audience") - monkeypatch.setenv( - "FASTMCP_AUTH_JWT_REQUIRED_SCOPES", '["test-scope1", "test-scope2"]' - ) - - with temporary_settings(**Settings().model_dump()): - mcp = FastMCP() - assert isinstance(mcp.auth, EnvJWTVerifier) - assert mcp.auth.public_key == "test-public-key" - assert mcp.auth.audience == "test-audience" - assert mcp.auth.required_scopes == ["test-scope1", "test-scope2"] - - -def test_list_of_scopes_must_be_a_list(monkeypatch): - monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env") - monkeypatch.setenv("FASTMCP_AUTH_JWT_REQUIRED_SCOPES", "test-scope1") - - with temporary_settings(**Settings().model_dump()): - with pytest.raises(ValidationError, match="Input should be a valid list"): - FastMCP() - - -def test_configure_bearer_env_jwks_uri_from_env_var(monkeypatch): - monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env") - monkeypatch.setenv("FASTMCP_AUTH_JWT_JWKS_URI", "test-jwks-uri") - - with temporary_settings(**Settings().model_dump()): - mcp = FastMCP() - assert isinstance(mcp.auth, EnvJWTVerifier) - assert mcp.auth.jwks_uri == "test-jwks-uri" - - -def test_configure_bearer_env_public_key_and_jwks_uri_error(monkeypatch): - monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env") - monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key") - monkeypatch.setenv("FASTMCP_AUTH_JWT_JWKS_URI", "test-jwks-uri") - - with temporary_settings(**Settings().model_dump()): - with pytest.raises(ValueError, match="Provide either public_key or jwks_uri"): - FastMCP() - - -def test_provided_auth_takes_precedence_over_env_vars(monkeypatch): - monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env") - monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key") - - with temporary_settings(**Settings().model_dump()): - mcp = FastMCP(auth=JWTVerifier(public_key="test-public-key-2")) - assert isinstance(mcp.auth, JWTVerifier) - assert not isinstance(mcp.auth, EnvJWTVerifier) - assert mcp.auth.public_key == "test-public-key-2" diff --git a/tests/auth/__init__.py b/tests/client/auth/__init__.py similarity index 100% rename from tests/auth/__init__.py rename to tests/client/auth/__init__.py diff --git a/tests/client/auth/test_oauth_client.py b/tests/client/auth/test_oauth_client.py new file mode 100644 index 000000000..742c0d723 --- /dev/null +++ b/tests/client/auth/test_oauth_client.py @@ -0,0 +1,128 @@ +from collections.abc import Generator +from urllib.parse import urlparse + +import httpx +import pytest + +from fastmcp.client import Client +from fastmcp.client.transports import StreamableHttpTransport +from fastmcp.server.auth.auth import ClientRegistrationOptions +from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider +from fastmcp.server.server import FastMCP +from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process + + +def fastmcp_server(issuer_url: str): + """Create a FastMCP server with OAuth authentication.""" + server = FastMCP( + "TestServer", + auth=InMemoryOAuthProvider( + base_url=issuer_url, + client_registration_options=ClientRegistrationOptions(enabled=True), + ), + ) + + @server.tool + def add(a: int, b: int) -> int: + """Add two numbers together.""" + return a + b + + @server.resource("resource://test") + def get_test_resource() -> str: + """Get a test resource.""" + return "Hello from authenticated resource!" + + return server + + +def run_server(host: str, port: int, **kwargs) -> None: + fastmcp_server(f"http://{host}:{port}").run(host=host, port=port, **kwargs) + + +@pytest.fixture(scope="module") +def streamable_http_server() -> Generator[str, None, None]: + with run_server_in_process(run_server, transport="http") as url: + yield f"{url}/mcp/" + + +@pytest.fixture() +def client_unauthorized(streamable_http_server: str) -> Client: + return Client(transport=StreamableHttpTransport(streamable_http_server)) + + +@pytest.fixture() +def client_with_headless_oauth( + streamable_http_server: str, +) -> Generator[Client, None, None]: + """Client with headless OAuth that bypasses browser interaction.""" + client = Client( + transport=StreamableHttpTransport(streamable_http_server), + auth=HeadlessOAuth(mcp_url=streamable_http_server), + ) + yield client + + +async def test_unauthorized(client_unauthorized: Client): + """Test that unauthenticated requests are rejected.""" + with pytest.raises(httpx.HTTPStatusError, match="401 Unauthorized"): + async with client_unauthorized: + pass + + +async def test_ping(client_with_headless_oauth: Client): + """Test that we can ping the server.""" + async with client_with_headless_oauth: + assert await client_with_headless_oauth.ping() + + +async def test_list_tools(client_with_headless_oauth: Client): + """Test that we can list tools.""" + async with client_with_headless_oauth: + tools = await client_with_headless_oauth.list_tools() + tool_names = [tool.name for tool in tools] + assert "add" in tool_names + + +async def test_call_tool(client_with_headless_oauth: Client): + """Test that we can call a tool.""" + async with client_with_headless_oauth: + result = await client_with_headless_oauth.call_tool("add", {"a": 5, "b": 3}) + # The add tool returns int which gets wrapped as structured output + # Client unwraps it and puts the actual int in the data field + assert result.data == 8 + + +async def test_list_resources(client_with_headless_oauth: Client): + """Test that we can list resources.""" + async with client_with_headless_oauth: + resources = await client_with_headless_oauth.list_resources() + resource_uris = [str(resource.uri) for resource in resources] + assert "resource://test" in resource_uris + + +async def test_read_resource(client_with_headless_oauth: Client): + """Test that we can read a resource.""" + async with client_with_headless_oauth: + resource = await client_with_headless_oauth.read_resource("resource://test") + assert resource[0].text == "Hello from authenticated resource!" # type: ignore[attr-defined] + + +async def test_oauth_server_metadata_discovery(streamable_http_server: str): + """Test that we can discover OAuth metadata from the running server.""" + parsed_url = urlparse(streamable_http_server) + server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" + + async with httpx.AsyncClient() as client: + # Test OAuth discovery endpoint + metadata_url = f"{server_base_url}/.well-known/oauth-authorization-server" + response = await client.get(metadata_url) + assert response.status_code == 200 + + metadata = response.json() + assert "authorization_endpoint" in metadata + assert "token_endpoint" in metadata + assert "registration_endpoint" in metadata + + # The endpoints should be properly formed URLs + assert metadata["authorization_endpoint"].startswith(server_base_url) + assert metadata["token_endpoint"].startswith(server_base_url) diff --git a/tests/contrib/test_component_manager.py b/tests/contrib/test_component_manager.py index d1941e171..56be8f625 100644 --- a/tests/contrib/test_component_manager.py +++ b/tests/contrib/test_component_manager.py @@ -4,7 +4,7 @@ from starlette.testclient import TestClient from fastmcp import FastMCP from fastmcp.contrib.component_manager import set_up_component_manager -from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair class TestComponentManagementRoutes: diff --git a/tests/deprecated/test_bearer_auth_provider.py b/tests/deprecated/test_bearer_auth_provider.py index e8b3b2ed4..5e02677d1 100644 --- a/tests/deprecated/test_bearer_auth_provider.py +++ b/tests/deprecated/test_bearer_auth_provider.py @@ -8,6 +8,6 @@ def test_bearer_auth_provider_deprecated(): """Test that BearerAuthProvider import shows deprecation warning.""" with pytest.warns( DeprecationWarning, - match="The `fastmcp.server.auth.providers.bearer` module is deprecated and will be removed in a future version. Please use `fastmcp.server.auth.verifiers.JWTVerifier` instead of this module's BearerAuthProvider.", + match="The `fastmcp.server.auth.providers.bearer` module is deprecated and will be removed in a future version. Please use `fastmcp.server.auth.providers.jwt.JWTVerifier` instead of this module's BearerAuthProvider.", ): from fastmcp.server.auth import BearerAuthProvider # noqa: F401 diff --git a/tests/deprecated/test_settings.py b/tests/deprecated/test_settings.py index d1c57f17b..9e6f3c335 100644 --- a/tests/deprecated/test_settings.py +++ b/tests/deprecated/test_settings.py @@ -232,6 +232,7 @@ class TestDeprecatedServerInitKwargs: "json_response": True, "stateless_http": True, } + mock_settings.server_auth = None # Add server_auth attribute server = FastMCP("TestServer") @@ -261,6 +262,7 @@ class TestDeprecatedServerInitKwargs: "json_response": True, "stateless_http": True, } + mock_settings.server_auth = None # Add server_auth attribute with warnings.catch_warnings(): warnings.simplefilter("ignore") # Ignore warnings for this test diff --git a/tests/auth/verifiers/test_jwt_verifier.py b/tests/server/auth/test_jwt_provider.py similarity index 97% rename from tests/auth/verifiers/test_jwt_verifier.py rename to tests/server/auth/test_jwt_provider.py index 0f853b765..79d99d6b9 100644 --- a/tests/auth/verifiers/test_jwt_verifier.py +++ b/tests/server/auth/test_jwt_provider.py @@ -7,7 +7,7 @@ from pytest_httpx import HTTPXMock from fastmcp import Client, FastMCP from fastmcp.client.auth.bearer import BearerAuth -from fastmcp.server.auth.verifiers import JWKData, JWKSData, JWTVerifier, RSAKeyPair +from fastmcp.server.auth.providers.jwt import JWKData, JWKSData, JWTVerifier, RSAKeyPair from fastmcp.utilities.tests import run_server_in_process @@ -776,3 +776,21 @@ class TestFastMCPBearerAuth: async with Client(mcp_server_url, auth=BearerAuth(token)) as client: tools = await client.list_tools() assert tools + + +class TestJWTVerifierImport: + """Test JWT token verifier can be imported and created.""" + + def test_jwt_verifier_requires_pyjwt(self): + """Test that JWTVerifier raises helpful error without PyJWT.""" + # Since PyJWT is likely installed in test environment, we'll just test construction + from fastmcp.server.auth.providers.jwt import JWTVerifier + + # This should work if PyJWT is available + try: + verifier = JWTVerifier(public_key="dummy-key") + assert verifier.public_key == "dummy-key" + assert verifier.algorithm == "RS256" + except ImportError as e: + # If PyJWT not available, should get helpful error + assert "PyJWT is required" in str(e) diff --git a/tests/server/auth/test_token_verifier_integration.py b/tests/server/auth/test_static_token_verifier.py similarity index 55% rename from tests/server/auth/test_token_verifier_integration.py rename to tests/server/auth/test_static_token_verifier.py index 2375d2650..51243dbee 100644 --- a/tests/server/auth/test_token_verifier_integration.py +++ b/tests/server/auth/test_static_token_verifier.py @@ -1,15 +1,14 @@ -"""Tests for TokenVerifier integration with FastMCP.""" +"""Tests for StaticTokenVerifier integration with FastMCP.""" import httpx -import pytest from mcp.server.auth.provider import AccessToken from fastmcp.server import FastMCP -from fastmcp.server.auth.verifiers import StaticTokenVerifier +from fastmcp.server.auth.providers.jwt import StaticTokenVerifier -class TestTokenVerifierIntegration: - """Test TokenVerifier integration with FastMCP server.""" +class TestStaticTokenVerifier: + """Test StaticTokenVerifier integration with FastMCP server.""" def test_static_token_verifier_creation(self): """Test creating a FastMCP server with StaticTokenVerifier.""" @@ -52,7 +51,7 @@ class TestTokenVerifierIntegration: assert result is None async def test_server_with_token_verifier_http_app(self): - """Test that FastMCP server works with TokenVerifier for HTTP requests.""" + """Test that FastMCP server works with StaticTokenVerifier for HTTP requests.""" verifier = StaticTokenVerifier( {"test-token": {"client_id": "test-client", "scopes": ["read", "write"]}} ) @@ -88,55 +87,3 @@ class TestTokenVerifierIntegration: # This should work - TokenVerifier server2 = FastMCP("Test2", auth=token_verifier) assert server2.auth is token_verifier - - -class TestJWTVerifierImport: - """Test JWT token verifier can be imported and created.""" - - def test_jwt_verifier_requires_pyjwt(self): - """Test that JWTVerifier raises helpful error without PyJWT.""" - # Since PyJWT is likely installed in test environment, we'll just test construction - from fastmcp.server.auth.verifiers import JWTVerifier - - # This should work if PyJWT is available - try: - verifier = JWTVerifier(public_key="dummy-key") - assert verifier.public_key == "dummy-key" - assert verifier.algorithm == "RS256" - except ImportError as e: - # If PyJWT not available, should get helpful error - assert "PyJWT is required" in str(e) - - -class TestIntrospectionTokenVerifierImport: - """Test introspection token verifier can be imported and created.""" - - def test_introspection_verifier_creation(self): - """Test IntrospectionTokenVerifier construction.""" - from fastmcp.server.auth.verifiers import IntrospectionTokenVerifier - - verifier = IntrospectionTokenVerifier( - "https://auth.example.com/introspect", "https://resource.example.com" - ) - - assert ( - str(verifier.introspection_endpoint) - == "https://auth.example.com/introspect" - ) - assert str(verifier.server_url) == "https://resource.example.com/" - assert verifier.validate_resource is False - assert verifier.required_scopes == [] - - def test_introspection_verifier_rejects_private_urls(self): - """Test that IntrospectionTokenVerifier rejects private URLs.""" - from fastmcp.server.auth.verifiers import IntrospectionTokenVerifier - - with pytest.raises(ValueError, match="private/localhost URL"): - IntrospectionTokenVerifier( - "http://localhost/introspect", "https://resource.example.com" - ) - - with pytest.raises(ValueError, match="private/localhost URL"): - IntrospectionTokenVerifier( - "http://127.0.0.1/introspect", "https://resource.example.com" - ) diff --git a/tests/server/auth/test_workos.py b/tests/server/auth/test_workos.py new file mode 100644 index 000000000..ff9c5daa1 --- /dev/null +++ b/tests/server/auth/test_workos.py @@ -0,0 +1,58 @@ +from collections.abc import Generator + +import httpx +import pytest + +from fastmcp import Client, FastMCP +from fastmcp.client.transports import StreamableHttpTransport +from fastmcp.server.auth.providers.workos import AuthKitProvider +from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process + + +def run_mcp_server(host: str, port: int) -> None: + mcp = FastMCP( + auth=AuthKitProvider( + authkit_domain="https://respectful-lullaby-34-staging.authkit.app", + base_url="http://localhost:4321", + ) + ) + + @mcp.tool + def add(a: int, b: int) -> int: + return a + b + + mcp.run(host=host, port=port, transport="http") + + +@pytest.fixture(scope="module") +def mcp_server_url() -> Generator[str]: + with run_server_in_process(run_mcp_server) as url: + yield f"{url}/mcp/" + + +@pytest.fixture() +def client_with_headless_oauth( + mcp_server_url: str, +) -> Generator[Client, None, None]: + """Client with headless OAuth that bypasses browser interaction.""" + client = Client( + transport=StreamableHttpTransport(mcp_server_url), + auth=HeadlessOAuth(mcp_url=mcp_server_url), + ) + yield client + + +class TestAuthKitProvider: + 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 exc_info.value.response.status_code == 401 + assert "tools" not in locals() + + # async def test_authorized_access(self, client_with_headless_oauth: Client): + # async with client_with_headless_oauth: + # tools = await client_with_headless_oauth.list_tools() + # assert tools is not None + # assert len(tools) > 0 + # assert "add" in tools diff --git a/tests/server/http/test_auth_setup.py b/tests/server/http/test_auth_setup.py index 277089a4c..695f05423 100644 --- a/tests/server/http/test_auth_setup.py +++ b/tests/server/http/test_auth_setup.py @@ -7,7 +7,7 @@ from starlette.middleware import Middleware from starlette.middleware.authentication import AuthenticationMiddleware from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider -from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair from fastmcp.server.http import setup_auth_middleware_and_routes @@ -29,7 +29,7 @@ class TestSetupAuthMiddlewareAndRoutes: def in_memory_provider(self) -> InMemoryOAuthProvider: """Create InMemoryOAuthProvider for testing.""" return InMemoryOAuthProvider( - issuer_url="https://test.example.com", + base_url="https://test.example.com", required_scopes=["user"], ) @@ -128,6 +128,10 @@ class MockOAuthProvider: ) return None + def customize_auth_routes(self, routes): + """Mock customize_auth_routes implementation.""" + return routes + class TestSetupWithMockProvider: """Test setup function with mock provider.""" diff --git a/tests/server/http/test_bearer_auth_backend.py b/tests/server/http/test_bearer_auth_backend.py index ed1d410db..f9ac065ca 100644 --- a/tests/server/http/test_bearer_auth_backend.py +++ b/tests/server/http/test_bearer_auth_backend.py @@ -5,7 +5,7 @@ from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend from mcp.server.auth.provider import AccessToken from starlette.requests import HTTPConnection -from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair class TestBearerAuthBackendTokenVerifierIntegration: diff --git a/tests/server/http/test_http_auth_middleware.py b/tests/server/http/test_http_auth_middleware.py index 6775b4354..10309036c 100644 --- a/tests/server/http/test_http_auth_middleware.py +++ b/tests/server/http/test_http_auth_middleware.py @@ -3,7 +3,7 @@ from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware from starlette.routing import Mount from fastmcp.server import FastMCP -from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair from fastmcp.server.http import create_streamable_http_app