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

Merge branch 'main' into 2-14-deprecations

Browse source
This commit is contained in:
Jeremiah Lowin 2025-11-15 11:06:18 -05:00 • committed by GitHub
commit 968027bd8b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 2863 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

179
.github/workflows/martian-test-failure.yml vendored Normal file
View file

@ -0,0 +1,179 @@
name: Marvin Test Failure Analysis
on:
workflow_run:
workflows: ["Run Tests"]
types:
- completed
concurrency:
group: marvin-test-failure-${{ github.event.workflow_run.head_branch }}
cancel-in-progress: true
jobs:
martian-test-failure:
# Only run if the test workflow failed
if: ${{ github.event.workflow_run.conclusion == 'failure' }}
runs-on: ubuntu-latest
permissions:
contents: read
pull-requests: write
issues: read
id-token: write
actions: read # Required for Claude to read CI results
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
fetch-depth: 1
- name: Generate Marvin App token
id: marvin-token
uses: actions/create-github-app-token@v2
with:
app-id: ${{ secrets.MARVIN_APP_ID }}
private-key: ${{ secrets.MARVIN_APP_PRIVATE_KEY }}
- name: Set up Python 3.10
uses: actions/setup-python@v5
with:
python-version: "3.10"
# Install UV package manager
- name: Install UV
uses: astral-sh/setup-uv@v7
# Install dependencies
- name: Install dependencies
run: uv sync --all-packages --group dev
- name: Set analysis prompt
id: analysis-prompt
run: |
cat >> $GITHUB_OUTPUT << 'EOF'
PROMPT<<PROMPT_END
You're a test failure analysis assistant for FastMCP, a Python framework for building Model Context Protocol servers and clients.
# Your Task
A GitHub Actions workflow has failed. Your job is to:
1. Analyze the test failure(s) to understand what went wrong
2. Identify the root cause of the failure(s)
3. Suggest a clear, actionable solution to fix the failure(s)
# Getting Started
1. Call the generate_agents_md tool to get a high-level summary of the project
2. Get the pull request associated with this workflow run from the GitHub repository: ${{ github.repository }}
- The workflow run ID is: ${{ github.event.workflow_run.id }}
- The workflow run was triggered by: ${{ github.event.workflow_run.event }}
- Use GitHub MCP tools to get PR details and workflow run information
3. Use the GitHub MCP tools to fetch job logs and failure information:
- Use get_workflow_run to get details about the failed workflow
- Use list_workflow_jobs to see which jobs failed
- Use get_job_logs with failed_only=true to get logs for failed jobs
- Use summarize_run_log_failures to get an AI summary of what failed
4. Analyze the failures to understand the root cause
5. Search the codebase for relevant files, tests, and implementations
# Your Response
Post a comment on the pull request with your analysis. Your comment should include:
## Test Failure Analysis
**Summary**: A brief 1-2 sentence summary of what failed.
**Root Cause**: A clear explanation of why the tests failed, based on your analysis of the logs and code.
**Suggested Solution**: Specific, actionable steps to fix the failure(s). Include:
- Which files need to be modified
- What changes are needed
- Why these changes will fix the issue
<details>
<summary>Detailed Analysis</summary>
Include here:
- Relevant log excerpts showing the failure
- Code snippets that are causing the issue
- Any related issues or PRs that might be relevant
</details>
<details>
<summary>Related Files</summary>
List files that are relevant to the failure with brief explanations of their relevance.
</details>
# Important Guidelines
- Be concise and actionable - developers want to quickly understand and fix the issue
- Focus on facts from the logs and code, not speculation
- If you can't determine the root cause, say so clearly
- Provide specific file names, line numbers, and code references when possible
- You can run make commands (e.g., `make lint`, `make typecheck`, `make sync`) to build, test, or lint the code
- You can also run git commands (e.g., `git status`, `git log`, `git diff`) to inspect the repository
- You can use WebSearch and WebFetch to research errors, stack traces, or related issues
- For bash commands, you are limited to make and git commands only
# CRITICAL: Loop Detection
**IMPORTANT**: Before posting your analysis, check the PR comments to detect if there's a loop where:
- CodeRabbit or another bot triggered this workflow
- Your previous analysis triggered CodeRabbit or another bot
- This created a repeating cycle of bot comments
# CRITICAL: ANGRY USERS
**IMPORTANT**: If the user is angry with you, the triage bot, don't respond. Just exit immediately without further action.
If you detect such a loop (e.g., you see multiple similar bot comments or your own previous analysis comments):
1. **DO NOT** post another analysis comment
2. Exit immediately without further action
# Problems Encountered
If you encounter any problems during your analysis (e.g., unable to fetch logs, tools not working), document them clearly so the team knows what limitations you faced.
PROMPT_END
EOF
- name: Setup GitHub MCP Server
run: |
mkdir -p /tmp/mcp-config
cat > /tmp/mcp-config/mcp-servers.json << 'EOF'
{
"mcpServers": {
"repository-summary": {
"type": "http",
"url": "https://agents-md-generator.fastmcp.app/mcp"
},
"code-search": {
"type": "http",
"url": "https://public-code-search.fastmcp.app/mcp"
},
"github-research": {
"type": "stdio",
"command": "uvx",
"args": [
"github-research-mcp"
],
"env": {
"DISABLE_SUMMARIES": "true",
"GITHUB_PERSONAL_ACCESS_TOKEN": "${{ secrets.GITHUB_TOKEN }}"
}
}
}
}
EOF
- name: Run Claude Code
id: claude
uses: anthropics/claude-code-action@v1
with:
github_token: ${{ steps.marvin-token.outputs.token }}
anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY_FOR_CI }}
bot_name: "Marvin Context Protocol"
claude_code_oauth_token: ${{ secrets.CLAUDE_CODE_OAUTH_TOKEN }}
additional_permissions: |
actions: read
prompt: ${{ steps.analysis-prompt.outputs.PROMPT }}
claude_args: |
--allowed-tools mcp__repository-summary,mcp__code-search,mcp__github-research,WebSearch,WebFetch,Bash(make:*,git:*)
--mcp-config /tmp/mcp-config/mcp-servers.json

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

@ -48,12 +48,7 @@ jobs:
run: uv sync --upgrade
- name: Run tests (excluding integration and client_process)
run: |
if [ "${{ matrix.os }}" = "windows-latest" ]; then
uv run pytest --inline-snapshot=disable tests -m "not integration and not client_process"
else
uv run pytest --inline-snapshot=disable tests -m "not integration and not client_process" --numprocesses auto --maxprocesses 4 --dist worksteal
fi
run: uv run pytest --inline-snapshot=disable tests -m "not integration and not client_process" --numprocesses auto --maxprocesses 4 --dist worksteal
shell: bash
- name: Run client process tests separately

BIN
docs/integrations/images/oci/ociaddapplication.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 179 KiB

BIN
docs/integrations/images/oci/ocieditdomainsettings.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 38 KiB

BIN
docs/integrations/images/oci/ocieditdomainsettingsbutton.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 68 KiB

BIN
docs/integrations/images/oci/ocioauthconfiguration.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 60 KiB

333
docs/integrations/oci.mdx Normal file
View file

@ -0,0 +1,333 @@
---
title: OCI IAM OAuth 🤝 FastMCP
sidebarTitle: Oracle
description: Secure your FastMCP server with OCI IAM OAuth
icon: shield-check
tag: NEW
---
import { VersionBadge } from "/snippets/version-badge.mdx"
<VersionBadge version="2.13.0" />
This guide shows you how to secure your FastMCP server using **OCI IAM OAuth**. Since OCI IAM doesn't support Dynamic Client Registration, this integration uses the [**OIDC Proxy**](/servers/auth/oidc-proxy) pattern to bridge OCI's traditional OAuth with MCP's authentication requirements.
## Configuration
### Prerequisites
1. An OCI cloud Account with access to create an Integrated Application in an Identity Domain.
2. Your FastMCP server's URL (For dev environments, it is http://localhost:8000. For PROD environments, it could be https://mcp.${DOMAIN}.com)
### Step 1: Make sure client access is enabled for JWK's URL
<Steps>
<Step title="Navigate to OCI IAM Domain Settings">
Login to OCI console (https://cloud.oracle.com for OCI commercial cloud).
From "Identity & Security" menu, open Domains page.
On the Domains list page, select the domain that you are using for MCP Authentication.
Open Settings tab.
Click on "Edit Domain Settings" button.
<Frame>
<img src="/images/oci/ocieditdomainsettingsbutton.png" alt="OCI console showing the Edit Domain Settings button in the IAM Domain settings page" />
</Frame>
</Step>
<Step title="Update Domain Setting">
Enable "Configure client access" checkbox as shown in the screenshot.
<Frame>
<img src="/images/oci/ocieditdomainsettings.png" alt="OCI IAM Domain Settings" />
</Frame>
</Step>
</Steps>
### Step 2: Create OAuth client for MCP server authentication
Follow the Steps as mentioned below to create an OAuth client.
<Steps>
<Step title="Navigate to OCI IAM Integrated Applications">
Login to OCI console (https://cloud.oracle.com for OCI commercial cloud).
From "Identity & Security" menu, open Domains page.
On the Domains list page, select the domain in which you want to create MCP server OAuth client. If you need help finding the list page for the domain, see [Listing Identity Domains.](https://docs.oracle.com/en-us/iaas/Content/Identity/domains/to-view-identity-domains.htm#view-identity-domains).
On the details page, select Integrated applications. A list of applications in the domain is displayed.
</Step>
<Step title="Add an Integrated Application">
Select Add application.
In the Add application window, select Confidential Application.
Select Launch workflow.
In the Add application details page, Enter name and description as shown below.
<Frame>
<img src="/images/oci/ociaddapplication.png" alt="Adding a Confidential Integrated Application in OCI IAM Domain" />
</Frame>
</Step>
<Step title="Update OAuth Configuration for an Integrated Application">
Once the Integrated Application is created, Click on "OAuth configuration" tab.
Click on "Edit OAuth configuration" button.
Configure the application as OAuth client by selecting "Configure this application as a client now" radio button.
Select "Authorization code" grant type. If you are planning to use the same OAuth client application for token exchange, select "Client credentials" grant type as well. In the sample, we will use the same client.
For Authorization grant type, select redirect URL. In most cases, this will be the MCP server URL followed by "/oauth/callback".
<Frame>
<img src="/images/oci/ocioauthconfiguration.png" alt="OAuth Configuration for an Integrated Application in OCI IAM Domain" />
</Frame>
</Step>
<Step title="Activate the Integrated Application">
Click on "Submit" button to update OAuth configuration for the client application.
**Note: You don't need to do any special configuration to support PKCE for the OAuth client.**
Make sure to Activate the client application.
Note down client ID and client secret for the application. Update .env file and replace FASTMCP_SERVER_AUTH_OCI_CLIENT_ID and FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET values.
FASTMCP_SERVER_AUTH_OCI_IAM_GUID in the env file is the Identity domain URL that you chose for the MCP server.
</Step>
</Steps>
This is all you need to implement MCP server authentication against OCI IAM. However, you may want to use an authenticated user token to invoke OCI control plane APIs and propagate identity to the OCI control plane instead of using a service user account. In that case, you need to implement token exchange.
### Step 3: Token Exchange Setup (Only if MCP server needs to talk to OCI Control Plane)
Token exchange helps you exchange a logged-in user's OCI IAM token for an OCI control plane session token, also known as UPST (User Principal Session Token). To learn more about token exchange, refer to my [Workload Identity Federation Blog](https://www.ateam-oracle.com/post/workload-identity-federation)
For token exchange, we need to configure Identity propagation trust. The blog above discusses setting up the trust using REST APIs. However, you can also use OCI CLI. Before using the CLI command below, ensure that you have created a token exchange OAuth client. In most cases, you can use the same OAuth client that you created above. You will use the client ID of the token exchange OAuth client in the CLI command below and replace it with {FASTMCP_SERVER_AUTH_OCI_CLIENT_ID}.
You will also need to update the client secret for the token exchange OAuth client in the .env file. It is the FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET parameter. Update FASTMCP_SERVER_AUTH_OCI_IAM_GUID and FASTMCP_SERVER_AUTH_OCI_CLIENT_ID as well for the token exchange OAuth client in the .env file.
```bash
oci identity-domains identity-propagation-trust create \
--schemas '["urn:ietf:params:scim:schemas:oracle:idcs:IdentityPropagationTrust"]' \
--public-key-endpoint "https://{FASTMCP_SERVER_AUTH_OCI_IAM_GUID}.identity.oraclecloud.com/admin/v1/SigningCert/jwk" \
--name "For Token Exchange" --type "JWT" \
--issuer "https://identity.oraclecloud.com/" --active true \
--endpoint "https://{FASTMCP_SERVER_AUTH_OCI_IAM_GUID}.identity.oracleclcoud.com" \
--subject-claim-name "sub" --allow-impersonation false \
--subject-mapping-attribute "username" \
--subject-type "User" --client-claim-name "iss" \
--client-claim-values '["https://identity.oraclecloud.com/"]' \
--oauth-clients '["{FASTMCP_SERVER_AUTH_OCI_CLIENT_ID}"]'
```
To exchange access token for OCI token and create a signer object, you need to add below code in MCP server. You can then use the signer object to create any OCI control plane client.
```python
from fastmcp.server.dependencies import get_access_token
from fastmcp.utilities.logging import get_logger
from oci.auth.signers import TokenExchangeSigner
import os
logger = get_logger(__name__)
# Load configuration from environment
FASTMCP_SERVER_AUTH_OCI_IAM_GUID = os.environ["FASTMCP_SERVER_AUTH_OCI_IAM_GUID"]
FASTMCP_SERVER_AUTH_OCI_CLIENT_ID = os.environ["FASTMCP_SERVER_AUTH_OCI_CLIENT_ID"]
FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET = os.environ["FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET"]
_global_token_cache = {} #In memory cache for OCI session token signer
def get_oci_signer() -> TokenExchangeSigner:
authntoken = get_access_token()
tokenID = authntoken.claims.get("jti")
token = authntoken.token
#Check if the signer exists for the token ID in memory cache
cached_signer = _global_token_cache.get(tokenID)
logger.debug(f"Global cached signer: {cached_signer}")
if cached_signer:
logger.debug(f"Using globally cached signer for token ID: {tokenID}")
return cached_signer
#If the signer is not yet created for the token then create new OCI signer object
logger.debug(f"Creating new signer for token ID: {tokenID}")
signer = TokenExchangeSigner(
jwt_or_func=token,
oci_domain_id=FASTMCP_SERVER_AUTH_OCI_IAM_GUID.split(".")[0],
client_id=FASTMCP_SERVER_AUTH_OCI_CLIENT_ID,
client_secret=FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET,
)
logger.debug(f"Signer {signer} created for token ID: {tokenID}")
#Cache the signer object in memory cache
_global_token_cache[tokenID] = signer
logger.debug(f"Signer cached for token ID: {tokenID}")
return signer
```
## Running MCP server
Once the setup is complete, to run the MCP server, run the below command.
```bash
fastmcp run server.py:mcp --transport http --port 8000
```
To run MCP client, run the below command.
```bash
python3 client.py
```
MCP Client sample is as below.
```python client.py
from fastmcp import Client
import asyncio
async def main():
# The client will automatically handle OCI OAuth flows
async with Client("http://localhost:8000/mcp/", auth="oauth") as client:
# First-time connection will open OCI login in your browser
print("✓ Authenticated with OCI IAM")
tools = await client.list_tools()
print(f"🔧 Available tools ({len(tools)}):")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
if __name__ == "__main__":
asyncio.run(main())
```
When you run the client for the first time:
1. Your browser will open to OCI IAM's login page
2. Sign in with your OCI account and grant the requested consent
3. After authorization, you'll be redirected back to the redirect path
4. The client receives the token and can make authenticated requests
## Production Configuration
<VersionBadge version="2.13.0" />
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, and `client_storage`:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.oci import OCIProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Load configuration from environment
FASTMCP_SERVER_AUTH_OCI_CONFIG_URL = os.environ["FASTMCP_SERVER_AUTH_OCI_CONFIG_URL"]
FASTMCP_SERVER_AUTH_OCI_CLIENT_ID = os.environ["FASTMCP_SERVER_AUTH_OCI_CLIENT_ID"]
FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET = os.environ["FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET"]
# Production setup with encrypted persistent token storage
auth_provider = OCIProvider(
config_url=FASTMCP_SERVER_AUTH_OCI_CONFIG_URL,
client_id=FASTMCP_SERVER_AUTH_OCI_CLIENT_ID,
client_secret=FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET,
base_url="https://your-production-domain.com",
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
client_storage=FernetEncryptionWrapper(
key_value=RedisStore(
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
),
fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
)
)
mcp = FastMCP(name="Production OCI App", auth=auth_provider)
```
<Note>
Parameters (`jwt_signing_key` and `client_storage`) work together to ensure tokens and client registrations survive server restarts. **Wrap your storage in `FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at Rest** - without it, tokens are stored in plaintext. Store secrets in environment variables and use a persistent storage backend like Redis for distributed deployments.
For complete details on these parameters, see the [OAuth Proxy documentation](/servers/auth/oauth-proxy#configuration-parameters).
</Note>
<Info>
The client caches tokens locally, so you won't need to re-authenticate for subsequent runs unless the token expires or you explicitly clear the cache.
</Info>
## Environment Variables
For production deployments, use environment variables instead of hardcoding credentials.
### Provider Selection
Setting this environment variable allows the OCI provider to be used automatically without explicitly instantiating it in code.
<Card>
<ParamField path="FASTMCP_SERVER_AUTH" default="Not set">
Set to `fastmcp.server.auth.providers.oci.OCIProvider` to use OCI IAM authentication.
</ParamField>
</Card>
### OCI-Specific Configuration
These environment variables provide default values for the OCI IAM provider, whether it's instantiated manually or configured via `FASTMCP_SERVER_AUTH`.
<Card>
<ParamField path="FASTMCP_SERVER_AUTH_OCI_IAM_GUID" required>
Your OCI Application Configuration URL (e.g., `idcs-asdascxasd11......identity.oraclecloud.com`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_OCI_CONFIG_URL" required>
Your OCI Application Configuration URL (e.g., `https://{FASTMCP_SERVER_AUTH_OCI_IAM_GUID}.identity.oraclecloud.com/.well-known/openid-configuration`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_OCI_CLIENT_ID" required>
Your OCI Application Client ID (e.g., `tv2ObNgaZAWWhhycr7Bz1LU2mxlnsmsB`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET" required>
Your OCI Application Client Secret (e.g., `idcsssvPYqbjemq...`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_OCI_BASE_URL" required>
Public URL where OAuth endpoints will be accessible (includes any mount path)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_OCI_REDIRECT_PATH" default="/auth/callback">
Redirect path configured in your OCI IAM Integrated Application
</ParamField>
</Card>
Example `.env` file:
```bash
# Use the OCI IAM provider
FASTMCP_SERVER_AUTH=fastmcp.server.auth.providers.oci.OCIProvider
# OCI IAM configuration and credentials
FASTMCP_SERVER_AUTH_OCI_IAM_GUID=idcs-asaacasd1111.....
FASTMCP_SERVER_AUTH_OCI_CONFIG_URL=https://{FASTMCP_SERVER_AUTH_OCI_IAM_GUID}.identity.oraclecloud.com/.well-known/openid-configuration
FASTMCP_SERVER_AUTH_OCI_CLIENT_ID=<your-client-id>
FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET=<your-client-secret>
FASTMCP_SERVER_AUTH_OCI_BASE_URL=https://your-server.com
```
With environment variables set, your server code simplifies to:
```python server.py
from fastmcp import FastMCP
from fastmcp.server.dependencies import get_access_token
# Authentication is automatically configured from environment
mcp = FastMCP(name="OCI Secured App")
@mcp.tool
def whoami() -> str:
"""The whoami function is to test MCP server without requiring token exchange.
This tool can be used to test successful authentication against OCI IAM.
It will return logged in user's subject (username from IAM domain)."""
token = get_access_token()
user = token.claims.get("sub")
return f"You are User: {user}"
```

21
docs/patterns/testing.mdx
View file

@ -5,7 +5,26 @@ description: How to test your FastMCP server.
icon: vial
---
The best way to ensure a reliable and maintainable FastMCP Server is to test it! The FastMCP Client combined with Pytest provides a simple and powerful way to test your FastMCP servers.
The best way to ensure a reliable and maintainable FastMCP Server is to test it! The FastMCP Client combined with Pytest provides a simple and powerful way to test your FastMCP servers.
## Prerequisites
Testing FastMCP servers requires `pytest-asyncio` to handle async test functions and fixtures. Install it as a development dependency:
```bash
pip install pytest-asyncio
```
We recommend configuring pytest to automatically handle async tests by setting the asyncio mode to `auto` in your `pyproject.toml`:
```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
```
This eliminates the need to decorate every async test with `@pytest.mark.asyncio`.
## Testing with Pytest Fixtures
Using Pytest Fixtures, you can wrap your FastMCP Server in a Client instance that makes interacting with your server fast and easy. This is especially useful when building your own MCP Servers and enables a tight development loop by allowing you to avoid using a separate tool like MCP Inspector during development:

2
docs/python-sdk/fastmcp-utilities-ui.mdx
View file

@ -18,7 +18,7 @@ consent pages, and other user-facing interfaces.
### `create_page` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/utilities/ui.py#L453" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
create_page(content: str, title: str = 'FastMCP', additional_styles: str = '', csp_policy: str = "default-src 'none'; style-src 'unsafe-inline'; img-src https:; base-uri 'none'") -> str
create_page(content: str, title: str = 'FastMCP', additional_styles: str = '', csp_policy: str = "default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'") -> str
```

2
docs/servers/auth/token-verification.mdx
View file

@ -212,6 +212,8 @@ Static token verification stores tokens as plain text and should never be used i
### Debug/Custom Token Verification
<VersionBadge version="2.13.1" />
The `DebugTokenVerifier` provides maximum flexibility for testing and special cases where standard token verification isn't applicable. It delegates validation to a user-provided callable, making it useful for prototyping, testing scenarios, or handling opaque tokens without introspection endpoints.
```python

42
docs/servers/context.mdx
View file

@ -313,6 +313,48 @@ async def request_info(ctx: Context) -> dict:
- **`ctx.client_id -> str | None`**: Get the ID of the client making the request, if provided during initialization
- **`ctx.session_id -> str | None`**: Get the MCP session ID for session-based data sharing (HTTP transports only)
#### Request Context Availability
<VersionBadge version="2.13.1" />
The `ctx.request_context` property provides access to the underlying MCP request context, but returns `None` when the MCP session has not been established yet. This typically occurs:
- During middleware execution in the `on_request` hook before the MCP handshake completes
- During the initialization phase of client connections
The MCP request context is distinct from the HTTP request. For HTTP transports, HTTP request data may be available even when the MCP session is not yet established.
To safely access the request context in situations where it may not be available:
```python
from fastmcp import FastMCP, Context
from fastmcp.server.dependencies import get_http_request
mcp = FastMCP(name="Session Aware Demo")
@mcp.tool
async def session_info(ctx: Context) -> dict:
"""Return session information when available."""
# Check if MCP session is available
if ctx.request_context:
# MCP session available - can access MCP-specific attributes
return {
"session_id": ctx.session_id,
"request_id": ctx.request_id,
"has_meta": ctx.request_context.meta is not None
}
else:
# MCP session not available - use HTTP helpers for request data (if using HTTP transport)
request = get_http_request()
return {
"message": "MCP session not available",
"user_agent": request.headers.get("user-agent", "Unknown")
}
```
For HTTP request access that works regardless of MCP session availability (when using HTTP transports), use the [HTTP request helpers](#http-requests) like `get_http_request()` and `get_http_headers()`.
#### Client Metadata
<VersionBadge version="2.13.1" />

36
docs/servers/middleware.mdx
View file

@ -103,11 +103,45 @@ This hierarchy allows you to target your middleware logic with the right level o
- `on_list_resource_templates`: Called when listing resource templates
- `on_list_prompts`: Called when listing available prompts
<VersionBadge version="2.13.0" />
- `on_initialize`: Called when a client connects and initializes the session (returns `None`)
- `on_initialize`: Called when a client connects and initializes the session (returns `None`)
<Note>
The `on_initialize` hook receives the client's initialization request but **returns `None`** rather than a result. The initialization response is handled internally by the MCP protocol and cannot be modified by middleware. This hook is useful for client detection, logging connections, or initializing session state, but not for modifying the initialization handshake itself.
</Note>
### MCP Session Availability in Middleware
<VersionBadge version="2.13.1" />
The MCP session and request context are not available during certain phases like initialization. When middleware runs during these phases, `context.fastmcp_context.request_context` returns `None` rather than the full MCP request context.
This typically occurs when:
- The `on_request` hook fires during client initialization
- The MCP handshake hasn't completed yet
To handle this in middleware, check if the MCP request context is available before accessing MCP-specific attributes. Note that the MCP request context is distinct from the HTTP request - for HTTP transports, you can use HTTP helpers to access request data even when the MCP session is not available:
```python
from fastmcp.server.middleware import Middleware, MiddlewareContext
class SessionAwareMiddleware(Middleware):
async def on_request(self, context: MiddlewareContext, call_next):
ctx = context.fastmcp_context
if ctx.request_context:
# MCP session available - can access session-specific attributes
session_id = ctx.session_id
request_id = ctx.request_id
else:
# MCP session not available yet - use HTTP helpers for request data (if using HTTP transport)
from fastmcp.server.dependencies import get_http_headers
headers = get_http_headers()
# Access HTTP data for auth, logging, etc.
return await call_next(context)
```
For HTTP request data (headers, client IP, etc.) when using HTTP transports, use `get_http_request()` or `get_http_headers()` from `fastmcp.server.dependencies`, which work regardless of MCP session availability. See [HTTP Requests](/servers/context#http-requests) for details.
## Component Access in Middleware
Understanding how to access component information (tools, resources, prompts) in middleware is crucial for building powerful middleware functionality. The access patterns differ significantly between listing operations and execution operations.

84
examples/testing_demo/README.md Normal file
View file

@ -0,0 +1,84 @@
# FastMCP Testing Demo
A comprehensive example demonstrating FastMCP testing patterns with pytest-asyncio.
## Overview
This example shows how to:
- Set up pytest-asyncio configuration in `pyproject.toml`
- Write async test fixtures for MCP clients
- Test tools, resources, and prompts
- Use parametrized tests for multiple scenarios
- Leverage inline-snapshot and dirty-equals for assertions
## Project Structure
```
testing_demo/
├── pyproject.toml # Project config with pytest-asyncio setup
├── server.py # Simple MCP server with tools/resources/prompts
├── tests/
│ └── test_server.py # Comprehensive test suite
└── README.md # This file
```
## Key Features
### pyproject.toml Configuration
The `pyproject.toml` includes the critical pytest-asyncio configuration:
```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
```
This eliminates the need for `@pytest.mark.asyncio` decorators on every async test.
### Server Components
The demo server (`server.py`) includes:
- **Tools**: `add`, `greet`, `async_multiply`
- **Resources**: `demo://info`, `demo://greeting/{name}`
- **Prompts**: `hello`, `explain`
### Test Patterns
The test suite demonstrates:
1. **Async fixture pattern**: Proper client fixture using `async with`
2. **Tool testing**: Calling tools and checking results via `.data` attribute
3. **Resource testing**: Reading static and templated resources
4. **Prompt testing**: Getting prompts with different arguments
5. **Parametrized tests**: Testing multiple scenarios efficiently
6. **Schema validation**: Verifying tool schemas and structure
7. **Pattern matching**: Using dirty-equals for flexible assertions
## Running the Tests
```bash
# Install dependencies
uv sync
# Run all tests
uv run pytest
# Run with verbose output
uv run pytest -v
# Run a specific test
uv run pytest tests/test_server.py::test_add_tool
```
## Running the Server
```bash
# Run the server
uv run fastmcp run server.py
# Inspect the server
uv run fastmcp inspect server.py
```
## Learning More
For detailed information about testing FastMCP servers, see the [Testing Documentation](../../docs/patterns/testing.mdx).

18
examples/testing_demo/pyproject.toml Normal file
View file

@ -0,0 +1,18 @@
[project]
name = "testing-demo"
version = "0.1.0"
description = "FastMCP testing example demonstrating pytest-asyncio patterns"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"fastmcp>=2.0.0",
"pytest>=8.3.3",
"pytest-asyncio>=1.2.0",
"dirty-equals>=0.9.0",
]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["."]
python_files = ["test_*.py"]

61
examples/testing_demo/server.py Normal file
View file

@ -0,0 +1,61 @@
"""
FastMCP Testing Demo Server
A simple MCP server demonstrating tools, resources, and prompts
with comprehensive test coverage.
"""
from fastmcp import FastMCP
# Create server
mcp = FastMCP("Testing Demo")
# Tools
@mcp.tool
def add(a: int, b: int) -> int:
"""Add two numbers together"""
return a + b
@mcp.tool
def greet(name: str, greeting: str = "Hello") -> str:
"""Greet someone with a customizable greeting"""
return f"{greeting}, {name}!"
@mcp.tool
async def async_multiply(x: float, y: float) -> float:
"""Multiply two numbers (async example)"""
return x * y
# Resources
@mcp.resource("demo://info")
def server_info() -> str:
"""Get server information"""
return "This is the FastMCP Testing Demo server"
@mcp.resource("demo://greeting/{name}")
def greeting_resource(name: str) -> str:
"""Get a personalized greeting resource"""
return f"Welcome to FastMCP, {name}!"
# Prompts
@mcp.prompt("hello")
def hello_prompt(name: str = "World") -> str:
"""Generate a hello world prompt"""
return f"Say hello to {name} in a friendly way."
@mcp.prompt("explain")
def explain_prompt(topic: str, detail_level: str = "medium") -> str:
"""Generate a prompt to explain a topic"""
if detail_level == "simple":
return f"Explain {topic} in simple terms for beginners."
elif detail_level == "detailed":
return f"Provide a detailed, technical explanation of {topic}."
else:
return f"Explain {topic} with moderate technical detail."

160
examples/testing_demo/tests/test_server.py Normal file
View file

@ -0,0 +1,160 @@
"""
Tests for the Testing Demo server.
Demonstrates pytest-asyncio patterns, fixtures, and testing best practices.
"""
import pytest
from dirty_equals import IsStr
from fastmcp.client import Client
@pytest.fixture
async def client():
"""
Client fixture for testing.
Uses async context manager and yields client synchronously.
No @pytest.mark.asyncio needed - asyncio_mode = "auto" handles it.
"""
# Import here to avoid import-time side effects
from server import mcp
async with Client(mcp) as client:
yield client
async def test_add_tool(client: Client):
"""Test the add tool with simple addition"""
result = await client.call_tool("add", {"a": 2, "b": 3})
assert result.data == 5
async def test_greet_tool_default(client: Client):
"""Test the greet tool with default greeting"""
result = await client.call_tool("greet", {"name": "Alice"})
assert result.data == "Hello, Alice!"
async def test_greet_tool_custom(client: Client):
"""Test the greet tool with custom greeting"""
result = await client.call_tool("greet", {"name": "Bob", "greeting": "Hi"})
assert result.data == "Hi, Bob!"
async def test_async_multiply_tool(client: Client):
"""Test the async multiply tool"""
result = await client.call_tool("async_multiply", {"x": 3.5, "y": 2.0})
assert result.data == 7.0
@pytest.mark.parametrize(
"a,b,expected",
[
(0, 0, 0),
(1, 1, 2),
(-1, 1, 0),
(100, 200, 300),
],
)
async def test_add_parametrized(client: Client, a: int, b: int, expected: int):
"""Test add tool with multiple parameter combinations"""
result = await client.call_tool("add", {"a": a, "b": b})
assert result.data == expected
async def test_server_info_resource(client: Client):
"""Test the server info resource"""
result = await client.read_resource("demo://info")
assert len(result) == 1
assert result[0].text == "This is the FastMCP Testing Demo server"
async def test_greeting_resource_template(client: Client):
"""Test the greeting resource template"""
result = await client.read_resource("demo://greeting/Charlie")
assert len(result) == 1
assert result[0].text == "Welcome to FastMCP, Charlie!"
async def test_hello_prompt_default(client: Client):
"""Test hello prompt with default parameter"""
result = await client.get_prompt("hello")
assert result.messages[0].content.text == "Say hello to World in a friendly way."
async def test_hello_prompt_custom(client: Client):
"""Test hello prompt with custom name"""
result = await client.get_prompt("hello", {"name": "Dave"})
assert result.messages[0].content.text == "Say hello to Dave in a friendly way."
async def test_explain_prompt_levels(client: Client):
"""Test explain prompt with different detail levels"""
# Simple level
result = await client.get_prompt(
"explain", {"topic": "MCP", "detail_level": "simple"}
)
assert "simple terms" in result.messages[0].content.text
assert "MCP" in result.messages[0].content.text
# Detailed level
result = await client.get_prompt(
"explain", {"topic": "MCP", "detail_level": "detailed"}
)
assert "detailed" in result.messages[0].content.text
assert "technical" in result.messages[0].content.text
async def test_list_tools(client: Client):
"""Test listing available tools"""
tools = await client.list_tools()
tool_names = [tool.name for tool in tools]
assert "add" in tool_names
assert "greet" in tool_names
assert "async_multiply" in tool_names
async def test_list_resources(client: Client):
"""Test listing available resources"""
resources = await client.list_resources()
resource_uris = [str(resource.uri) for resource in resources]
# Check that we have at least the static resource
assert "demo://info" in resource_uris
# There should be at least one resource listed
assert len(resource_uris) >= 1
async def test_list_prompts(client: Client):
"""Test listing available prompts"""
prompts = await client.list_prompts()
prompt_names = [prompt.name for prompt in prompts]
assert "hello" in prompt_names
assert "explain" in prompt_names
# Example using dirty-equals for flexible assertions
async def test_greet_with_dirty_equals(client: Client):
"""Test greet tool using dirty-equals for pattern matching"""
result = await client.call_tool("greet", {"name": "Eve"})
# Check that result data matches the pattern
assert result.data == IsStr(regex=r"^Hello, \w+!$")
# Example using inline-snapshot for complex data
async def test_tool_schema_structure(client: Client):
"""Test tool schema structure"""
tools = await client.list_tools()
add_tool = next(tool for tool in tools if tool.name == "add")
# Verify basic schema structure
assert add_tool.name == "add"
assert add_tool.description == "Add two numbers together"
assert "a" in add_tool.inputSchema["properties"]
assert "b" in add_tool.inputSchema["properties"]
assert add_tool.inputSchema["properties"]["a"]["type"] == "integer"
assert add_tool.inputSchema["properties"]["b"]["type"] == "integer"

1569
examples/testing_demo/uv.lock generated Normal file
View file

File diff suppressed because it is too large Load diff

31
src/fastmcp/experimental/utilities/openapi/parser.py
View file

@ -630,25 +630,28 @@ class OpenAPIParser(
Returns:
Dictionary containing only the schemas needed for outputs
"""
needed_schemas = set()
if not responses or not all_schemas:
return {}
needed_schemas: set[str] = set()
# Check responses for schema references
for response in responses.values():
if response.content_schema:
for content_schema in response.content_schema.values():
# Check if this schema was originally a top-level $ref
if "x-fastmcp-top-level-schema" in content_schema:
schema_name = content_schema["x-fastmcp-top-level-schema"]
if schema_name in all_schemas:
needed_schemas.add(schema_name)
if not response.content_schema:
continue
# Extract all dependencies (transitive refs within the schema)
deps = self._extract_schema_dependencies(
content_schema, all_schemas
for content_schema in response.content_schema.values():
deps = self._extract_schema_dependencies(content_schema, all_schemas)
needed_schemas.update(deps)
schema_name = content_schema.get("x-fastmcp-top-level-schema")
if isinstance(schema_name, str) and schema_name in all_schemas:
needed_schemas.add(schema_name)
self._extract_schema_dependencies(
all_schemas[schema_name],
all_schemas,
collected=needed_schemas,
)
needed_schemas.update(deps)
# Return only the needed output schemas
return {
name: all_schemas[name] for name in needed_schemas if name in all_schemas
}

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

@ -380,7 +380,7 @@ def create_consent_html(
form_action_schemes.append(f"{redirect_scheme}:")
form_action_directive = " ".join(form_action_schemes)
csp_policy = f"default-src 'none'; style-src 'unsafe-inline'; img-src https:; base-uri 'none'; form-action {form_action_directive}"
csp_policy = f"default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'; form-action {form_action_directive}"
return create_page(
content=content,
@ -468,9 +468,7 @@ def create_error_html(
)
# Simple CSP policy for error pages (no forms needed)
csp_policy = (
"default-src 'none'; style-src 'unsafe-inline'; img-src https:; base-uri 'none'"
)
csp_policy = "default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'"
return create_page(
content=content,
@ -1177,7 +1175,8 @@ class OAuthProxy(OAuthProvider):
await self._upstream_token_store.put(
key=upstream_token_id,
value=upstream_token_set,
ttl=expires_in, # Auto-expire when access token expires
ttl=refresh_expires_in
or expires_in, # Auto-expire when refresh token, or access token expires
)
logger.debug("Stored encrypted upstream tokens (jti=%s)", access_jti[:8])
@ -1327,6 +1326,7 @@ class OAuthProxy(OAuthProvider):
url=self._upstream_token_endpoint,
refresh_token=upstream_token_set.refresh_token,
scope=" ".join(scopes) if scopes else None,
**self._extra_token_params,
)
logger.debug("Successfully refreshed upstream token")
except Exception as e:
@ -1373,7 +1373,12 @@ class OAuthProxy(OAuthProvider):
await self._upstream_token_store.put(
key=upstream_token_set.upstream_token_id,
value=upstream_token_set,
ttl=new_expires_in, # Auto-expire when refreshed access token expires
ttl=new_refresh_expires_in
or (
int(upstream_token_set.refresh_token_expires_at - time.time())
if upstream_token_set.refresh_token_expires_at
else 60 * 60 * 24 * 30 # Default to 30 days if unknown
), # Auto-expire when refresh token expires
)
# Issue new minimal FastMCP access token (just a reference via JTI)

233
src/fastmcp/server/auth/providers/oci.py Normal file
View file

@ -0,0 +1,233 @@
"""OCI OIDC provider for FastMCP.
The pull request for the provider is submitted to fastmcp.
This module provides OIDC Implementation to integrate MCP servers with OCI.
You only need OCI Identity Domain's discovery URL, client ID, client secret, and base URL.
Post Authentication, you get OCI IAM domain access token. That is not authorized to invoke OCI control plane.
You need to exchange the IAM domain access token for OCI UPST token to invoke OCI control plane APIs.
The sample code below has get_oci_signer function that returns OCI TokenExchangeSigner object.
You can use the signer object to create OCI service object.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.auth.providers.oci import OCIProvider
from fastmcp.server.dependencies import get_access_token
from fastmcp.utilities.logging import get_logger
import os
# Load configuration from environment
FASTMCP_SERVER_AUTH_OCI_CONFIG_URL = os.environ["FASTMCP_SERVER_AUTH_OCI_CONFIG_URL"]
FASTMCP_SERVER_AUTH_OCI_CLIENT_ID = os.environ["FASTMCP_SERVER_AUTH_OCI_CLIENT_ID"]
FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET = os.environ["FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET"]
FASTMCP_SERVER_AUTH_OCI_IAM_GUID = os.environ["FASTMCP_SERVER_AUTH_OCI_IAM_GUID"]
import oci
from oci.auth.signers import TokenExchangeSigner
logger = get_logger(__name__)
# Simple OCI OIDC protection
auth = OCIProvider(
config_url=FASTMCP_SERVER_AUTH_OCI_CONFIG_URL, #config URL is the OCI IAM Domain OIDC discovery URL.
client_id=FASTMCP_SERVER_AUTH_OCI_CLIENT_ID, #This is same as the client ID configured for the OCI IAM Domain Integrated Application
client_secret=FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET, #This is same as the client secret configured for the OCI IAM Domain Integrated Application
required_scopes=["openid", "profile", "email"],
redirect_path="/auth/callback",
base_url="http://localhost:8000",
)
# NOTE: For production use, replace this with a thread-safe cache implementation
# such as threading.Lock-protected dict or a proper caching library
_global_token_cache = {} #In memory cache for OCI session token signer
def get_oci_signer() -> TokenExchangeSigner:
authntoken = get_access_token()
tokenID = authntoken.claims.get("jti")
token = authntoken.token
#Check if the signer exists for the token ID in memory cache
cached_signer = _global_token_cache.get(tokenID)
logger.debug(f"Global cached signer: {cached_signer}")
if cached_signer:
logger.debug(f"Using globally cached signer for token ID: {tokenID}")
return cached_signer
#If the signer is not yet created for the token then create new OCI signer object
logger.debug(f"Creating new signer for token ID: {tokenID}")
signer = TokenExchangeSigner(
jwt_or_func=token,
oci_domain_id=FASTMCP_SERVER_AUTH_OCI_IAM_GUID.split(".")[0], #This is same as IAM GUID configured for the OCI IAM Domain
client_id=FASTMCP_SERVER_AUTH_OCI_CLIENT_ID, #This is same as the client ID configured for the OCI IAM Domain Integrated Application
client_secret=FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET #This is same as the client secret configured for the OCI IAM Domain Integrated Application
)
logger.debug(f"Signer {signer} created for token ID: {tokenID}")
#Cache the signer object in memory cache
_global_token_cache[tokenID] = signer
logger.debug(f"Signer cached for token ID: {tokenID}")
return signer
mcp = FastMCP("My Protected Server", auth=auth)
```
"""
from key_value.aio.protocols import AsyncKeyValue
from pydantic import AnyHttpUrl, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from fastmcp.server.auth.oidc_proxy import OIDCProxy
from fastmcp.settings import ENV_FILE
from fastmcp.utilities.auth import parse_scopes
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import NotSet, NotSetT
logger = get_logger(__name__)
class OCIProviderSettings(BaseSettings):
"""Settings for OCI IAM domain OIDC provider."""
model_config = SettingsConfigDict(
env_prefix="FASTMCP_SERVER_AUTH_OCI_",
env_file=ENV_FILE,
extra="ignore",
)
config_url: AnyHttpUrl | None = None
client_id: str | None = None
client_secret: SecretStr | None = None
audience: str | None = None
base_url: AnyHttpUrl | None = None
issuer_url: AnyHttpUrl | None = None
redirect_path: str | None = None
required_scopes: list[str] | None = None
allowed_client_redirect_uris: list[str] | None = None
jwt_signing_key: str | bytes | None = None
@field_validator("required_scopes", mode="before")
@classmethod
def _parse_scopes(cls, v):
return parse_scopes(v)
class OCIProvider(OIDCProxy):
"""An OCI IAM Domain provider implementation for FastMCP.
This provider is a complete OCI integration that's ready to use with
just the configuration URL, client ID, client secret, and base URL.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.auth.providers.oci import OCIProvider
# Simple OCI OIDC protection
auth = OCIProvider(
config_url=FASTMCP_SERVER_AUTH_OCI_CONFIG_URL, #config URL is the OCI IAM Domain OIDC discovery URL.
client_id=FASTMCP_SERVER_AUTH_OCI_CLIENT_ID, #This is same as the client ID configured for the OCI IAM Domain Integrated Application
client_secret=FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET, #This is same as the client secret configured for the OCI IAM Domain Integrated Application
base_url="http://localhost:8000",
required_scopes=["openid", "profile", "email"],
redirect_path="/auth/callback",
)
mcp = FastMCP("My Protected Server", auth=auth)
```
"""
def __init__(
self,
*,
config_url: AnyHttpUrl | str | NotSetT = NotSet,
client_id: str | NotSetT = NotSet,
client_secret: str | NotSetT = NotSet,
audience: str | NotSetT = NotSet,
base_url: AnyHttpUrl | str | NotSetT = NotSet,
issuer_url: AnyHttpUrl | str | NotSetT = NotSet,
required_scopes: list[str] | NotSetT = NotSet,
redirect_path: str | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | NotSetT = NotSet,
require_authorization_consent: bool = True,
) -> None:
"""Initialize OCI OIDC provider.
Args:
config_url: OCI OIDC Discovery URL
client_id: OCI IAM Domain Integrated Application client id
client_secret: OCI Integrated Application client secret
audience: OCI API audience (optional)
base_url: Public URL where OIDC endpoints will be accessible (includes any mount path)
issuer_url: Issuer URL for OCI IAM Domain metadata. This will override issuer URL from the discovery URL.
required_scopes: Required OCI scopes (defaults to ["openid"])
redirect_path: Redirect path configured in OCI IAM Domain Integrated Application. The default is "/auth/callback".
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
"""
overrides = {
k: v
for k, v in {
"config_url": config_url,
"client_id": client_id,
"client_secret": client_secret,
"audience": audience,
"base_url": base_url,
"issuer_url": issuer_url,
"required_scopes": required_scopes,
"redirect_path": redirect_path,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"jwt_signing_key": jwt_signing_key,
}.items()
if v is not NotSet
}
settings = OCIProviderSettings(**overrides)
if not settings.config_url:
raise ValueError(
"config_url is required - set via parameter or FASTMCP_SERVER_AUTH_OCI_CONFIG_URL"
)
if not settings.client_id:
raise ValueError(
"client_id is required - set via parameter or FASTMCP_SERVER_AUTH_OCI_CLIENT_ID"
)
if not settings.client_secret:
raise ValueError(
"client_secret is required - set via parameter or FASTMCP_SERVER_AUTH_OCI_CLIENT_SECRET"
)
if not settings.base_url:
raise ValueError(
"base_url is required - set via parameter or FASTMCP_SERVER_AUTH_OCI_BASE_URL"
)
oci_required_scopes = settings.required_scopes or ["openid"]
super().__init__(
config_url=settings.config_url,
client_id=settings.client_id,
client_secret=settings.client_secret.get_secret_value(),
audience=settings.audience,
base_url=settings.base_url,
issuer_url=settings.issuer_url,
redirect_path=settings.redirect_path,
required_scopes=oci_required_scopes,
allowed_client_redirect_uris=settings.allowed_client_redirect_uris,
client_storage=client_storage,
jwt_signing_key=settings.jwt_signing_key,
require_authorization_consent=require_authorization_consent,
)
logger.debug(
"Initialized OCI OAuth provider for client %s with scopes: %s",
settings.client_id,
oci_required_scopes,
)

58
src/fastmcp/server/context.py
View file

@ -178,15 +178,33 @@ class Context:
_current_context.reset(token)
@property
def request_context(self) -> RequestContext[ServerSession, Any, Request]:
def request_context(self) -> RequestContext[ServerSession, Any, Request] | None:
"""Access to the underlying request context.
If called outside of a request context, this will raise a ValueError.
Returns None when the MCP session has not been established yet.
Returns the full RequestContext once the MCP session is available.
For HTTP request access in middleware, use `get_http_request()` from fastmcp.server.dependencies,
which works whether or not the MCP session is available.
Example in middleware:
```python
async def on_request(self, context, call_next):
ctx = context.fastmcp_context
if ctx.request_context:
# MCP session available - can access session_id, request_id, etc.
session_id = ctx.session_id
else:
# MCP session not available yet - use HTTP helpers
from fastmcp.server.dependencies import get_http_request
request = get_http_request()
return await call_next(context)
```
"""
try:
return request_ctx.get()
except LookupError as e:
raise ValueError("Context is not available outside of a request") from e
except LookupError:
return None
async def report_progress(
self, progress: float, total: float | None = None, message: str | None = None
@ -200,7 +218,7 @@ class Context:
progress_token = (
self.request_context.meta.progressToken
if self.request_context.meta
if self.request_context and self.request_context.meta
else None
)
@ -289,13 +307,21 @@ class Context:
"""Get the client ID if available."""
return (
getattr(self.request_context.meta, "client_id", None)
if self.request_context.meta
if self.request_context and self.request_context.meta
else None
)
@property
def request_id(self) -> str:
"""Get the unique ID for this request."""
"""Get the unique ID for this request.
Raises RuntimeError if MCP request context is not available.
"""
if self.request_context is None:
raise RuntimeError(
"request_id is not available because the MCP session has not been established yet. "
"Check `context.request_context` for None before accessing this attribute."
)
return str(self.request_context.request_id)
@property
@ -310,6 +336,9 @@ class Context:
The session ID for StreamableHTTP transports, or a generated ID
for other transports.
Raises:
RuntimeError if MCP request context is not available.
Example:
```python
@server.tool
@ -320,6 +349,11 @@ class Context:
```
"""
request_ctx = self.request_context
if request_ctx is None:
raise RuntimeError(
"session_id is not available because the MCP session has not been established yet. "
"Check `context.request_context` for None before accessing this attribute."
)
session = request_ctx.session
# Try to get the session ID from the session attributes
@ -344,7 +378,15 @@ class Context:
@property
def session(self) -> ServerSession:
"""Access to the underlying session for advanced usage."""
"""Access to the underlying session for advanced usage.
Raises RuntimeError if MCP request context is not available.
"""
if self.request_context is None:
raise RuntimeError(
"session is not available because the MCP session has not been established yet. "
"Check `context.request_context` for None before accessing this attribute."
)
return self.request_context.session
# Convenience methods for common log levels

22
src/fastmcp/utilities/logging.py
View file

@ -74,15 +74,19 @@ def configure_logging(
import mcp
import pydantic
traceback_handler = RichHandler(
console=Console(stderr=True),
show_path=False,
show_level=False,
rich_tracebacks=enable_rich_tracebacks,
tracebacks_max_frames=3,
tracebacks_suppress=[fastmcp, mcp, pydantic],
**rich_kwargs,
)
# Build traceback kwargs with defaults that can be overridden
traceback_kwargs = {
"console": Console(stderr=True),
"show_path": False,
"show_level": False,
"rich_tracebacks": enable_rich_tracebacks,
"tracebacks_max_frames": 3,
"tracebacks_suppress": [fastmcp, mcp, pydantic],
}
# Override defaults with user-provided values
traceback_kwargs.update(rich_kwargs)
traceback_handler = RichHandler(**traceback_kwargs) # type: ignore[arg-type]
traceback_handler.setFormatter(formatter)
traceback_handler.addFilter(lambda record: record.exc_info is not None)

2
src/fastmcp/utilities/ui.py
View file

@ -454,7 +454,7 @@ def create_page(
content: str,
title: str = "FastMCP",
additional_styles: str = "",
csp_policy: str = "default-src 'none'; style-src 'unsafe-inline'; img-src https:; base-uri 'none'",
csp_policy: str = "default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'",
) -> str:
"""
Create a complete HTML page with FastMCP styling.

3
tests/server/auth/test_oauth_consent_flow.py
View file

@ -691,6 +691,7 @@ class TestConsentPageServerIcon:
upstream_client_secret="upstream-secret",
token_verifier=verifier,
base_url="https://proxy.example.com",
client_storage=MemoryStore(),
jwt_signing_key="test-secret",
)
@ -763,6 +764,7 @@ class TestConsentPageServerIcon:
upstream_client_secret="upstream-secret",
token_verifier=verifier,
base_url="https://proxy.example.com",
client_storage=MemoryStore(),
jwt_signing_key="test-secret",
)
@ -830,6 +832,7 @@ class TestConsentPageServerIcon:
upstream_client_secret="upstream-secret",
token_verifier=verifier,
base_url="https://proxy.example.com",
client_storage=MemoryStore(),
jwt_signing_key="test-secret",
)

4
tests/server/auth/test_oidc_proxy.py
View file

@ -307,7 +307,7 @@ class TestOIDCConfiguration:
with pytest.raises(ValueError, match="Invalid URL for configuration metadata"):
OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
def test_explict_strict_with_bad_url_raises_error(
def test_explicit_strict_with_bad_url_raises_error(
self, valid_oidc_configuration_dict
):
"""Test default configuration with explicit True strict setting and bad URL setting."""
@ -359,7 +359,7 @@ class TestOIDCConfiguration:
def validate_get_oidc_configuration(oidc_configuration, strict, timeout_seconds):
"""Validate get_oidc_configuation call."""
"""Validate get_oidc_configuration call."""
with patch("httpx.get") as mock_get:
mock_response = MagicMock(spec=Response)
mock_response.json.return_value = oidc_configuration

28
tests/utilities/test_logging.py
View file

@ -1,6 +1,6 @@
import logging
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.logging import configure_logging, get_logger
def test_logging_doesnt_affect_other_loggers(caplog):
@ -28,3 +28,29 @@ def test_logging_doesnt_affect_other_loggers(caplog):
finally:
logging.getLogger("fastmcp").setLevel(original_level)
def test_configure_logging_with_traceback_kwargs():
"""Test that traceback-related kwargs can be passed without causing duplicate argument errors."""
# This should not raise TypeError about duplicate keyword arguments
configure_logging(enable_rich_tracebacks=True, tracebacks_max_frames=20)
# Verify the logger was configured
logger = logging.getLogger("fastmcp")
assert logger.handlers
assert len(logger.handlers) == 2 # One for normal logs, one for tracebacks
def test_configure_logging_traceback_defaults_can_be_overridden():
"""Test that default traceback settings can be overridden by kwargs."""
configure_logging(
enable_rich_tracebacks=True,
tracebacks_max_frames=20,
show_path=True,
show_level=True,
)
logger = logging.getLogger("fastmcp")
assert logger.handlers
# The traceback handler should have been created with custom values
# We can't directly inspect RichHandler internals easily, but we verified no error was raised