Minor updates

This commit is contained in:
Jeremiah Lowin 2025-06-19 15:28:55 -04:00
commit 0beb79d11a
6 changed files with 37 additions and 32 deletions

View file

@ -78,6 +78,7 @@
"servers/auth/bearer"
]
},
"servers/middleware",
"servers/openapi",
"servers/proxy",
"servers/composition",

View file

@ -193,7 +193,7 @@ def hello():
return "hi"
# Mount directly
main.mount("sub", sub)
main.mount(sub, prefix="sub")
```
## Proxying Servers

View file

@ -1,18 +1,22 @@
---
title: MCP Middleware
sidebarTitle: Middleware
description: Add cross-cutting functionality to your MCP server with middleware that can inspect, modify, and respond to all MCP requests and responses.
description: Add custom functionality to your MCP server with middleware that can inspect, modify, and respond to all MCP requests and responses.
icon: layers
---
import { VersionBadge } from "/snippets/version-badge.mdx"
<VersionBadge version="2.8.0" />
<VersionBadge version="2.9.0" />
MCP middleware is a powerful concept that allows you to add cross-cutting functionality to your FastMCP server. Unlike traditional web middleware, MCP middleware is designed specifically for the Model Context Protocol, providing hooks for different types of MCP operations like tool calls, resource reads, and prompt requests.
<Warning>
<Tip>
MCP middleware is a FastMCP-specific concept and is not part of the official MCP protocol specification. This middleware system is designed to work with FastMCP servers and may not be compatible with other MCP implementations.
</Tip>
<Warning>
MCP middleware is a brand new concept and may be subject to breaking changes in future versions.
</Warning>
## What is MCP Middleware?
@ -58,13 +62,13 @@ The middleware hook system is designed to be extensible. As FastMCP evolves and
### Basic Middleware Structure
MCP middleware is implemented by subclassing the `MCPMiddleware` base class and overriding the hooks you need:
MCP middleware is implemented by subclassing the `Middleware` base class and overriding the hooks you need:
```python
from fastmcp import FastMCP
from fastmcp.server.middleware import MCPMiddleware, MiddlewareContext
from fastmcp.server.middleware import Middleware, MiddlewareContext
class LoggingMiddleware(MCPMiddleware):
class LoggingMiddleware(Middleware):
"""Middleware that logs all MCP operations."""
async def on_message(self, context: MiddlewareContext, call_next):
@ -97,7 +101,7 @@ mcp.add_middleware(LoggingMiddleware())
The `MiddlewareContext` object provides access to information about the current request:
```python
class InspectionMiddleware(MCPMiddleware):
class InspectionMiddleware(Middleware):
async def on_request(self, context: MiddlewareContext, call_next):
# Access request information
method = context.method # e.g., "tools/call"
@ -120,7 +124,7 @@ Each middleware hook receives a `MiddlewareContext` and a `call_next` function.
3. **Operation-specific hooks**: Called for specific MCP operations
```python
class ComprehensiveMiddleware(MCPMiddleware):
class ComprehensiveMiddleware(Middleware):
async def on_message(self, context: MiddlewareContext, call_next):
"""Called for ALL messages (requests and notifications)."""
print(f"Message: {context.method}")
@ -143,10 +147,10 @@ class ComprehensiveMiddleware(MCPMiddleware):
### Authentication Middleware
```python
from fastmcp.server.middleware import MCPMiddleware, MiddlewareContext
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolError
class AuthenticationMiddleware(MCPMiddleware):
class AuthenticationMiddleware(Middleware):
def __init__(self, required_token: str):
self.required_token = required_token
@ -184,7 +188,7 @@ mcp.add_middleware(AuthenticationMiddleware("secret-token-123"))
import time
import logging
class PerformanceMiddleware(MCPMiddleware):
class PerformanceMiddleware(Middleware):
def __init__(self):
self.logger = logging.getLogger("performance")
@ -214,7 +218,7 @@ class PerformanceMiddleware(MCPMiddleware):
### Request/Response Transformation Middleware
```python
class TransformationMiddleware(MCPMiddleware):
class TransformationMiddleware(Middleware):
async def on_call_tool(self, context: MiddlewareContext, call_next):
"""Transform tool arguments and results."""
@ -253,7 +257,7 @@ import asyncio
from collections import defaultdict
from datetime import datetime, timedelta
class RateLimitMiddleware(MCPMiddleware):
class RateLimitMiddleware(Middleware):
def __init__(self, max_requests: int = 100, window_minutes: int = 1):
self.max_requests = max_requests
self.window = timedelta(minutes=window_minutes)
@ -327,7 +331,7 @@ mcp.add_middleware(LoggingMiddleware())
### Conditional Middleware
```python
class ConditionalMiddleware(MCPMiddleware):
class ConditionalMiddleware(Middleware):
def __init__(self, condition_func):
self.should_process = condition_func
@ -352,7 +356,7 @@ mcp.add_middleware(ConditionalMiddleware(only_expensive_tools))
### Middleware with State
```python
class StatefulMiddleware(MCPMiddleware):
class StatefulMiddleware(Middleware):
def __init__(self):
self.call_count = 0
self.tools_used = set()
@ -377,7 +381,7 @@ class StatefulMiddleware(MCPMiddleware):
### Error Handling Middleware
```python
class ErrorHandlingMiddleware(MCPMiddleware):
class ErrorHandlingMiddleware(Middleware):
async def on_call_tool(self, context: MiddlewareContext, call_next):
"""Provide consistent error handling."""
try:
@ -400,7 +404,7 @@ class ErrorHandlingMiddleware(MCPMiddleware):
3. **Cache when possible**: Store frequently accessed data to avoid repeated lookups
```python
class EfficientMiddleware(MCPMiddleware):
class EfficientMiddleware(Middleware):
def __init__(self):
self._cache = {}
@ -428,7 +432,7 @@ class EfficientMiddleware(MCPMiddleware):
3. **Use `ToolError` for client-facing errors**: Keep internal errors internal
```python
class RobustMiddleware(MCPMiddleware):
class RobustMiddleware(Middleware):
async def on_request(self, context: MiddlewareContext, call_next):
"""Robust error handling example."""
try:

View file

@ -104,7 +104,7 @@ class MiddlewareContext(Generic[T]):
def make_middleware_wrapper(
middleware: MCPMiddleware, call_next: CallNext[T, R]
middleware: Middleware, call_next: CallNext[T, R]
) -> CallNext[T, R]:
"""Create a wrapper that applies a single middleware to a context. The
closure bakes in the middleware and call_next function, so it can be
@ -116,7 +116,7 @@ def make_middleware_wrapper(
return wrapper
class MCPMiddleware:
class Middleware:
"""Base class for FastMCP middleware with dispatching hooks."""
async def __call__(

View file

@ -35,7 +35,7 @@ from mcp.types import Resource as MCPResource
from mcp.types import ResourceTemplate as MCPResourceTemplate
from mcp.types import Tool as MCPTool
from pydantic import AnyUrl
from starlette.middleware import Middleware
from starlette.middleware import Middleware as ASGIMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import BaseRoute, Route
@ -55,7 +55,7 @@ from fastmcp.server.http import (
create_sse_app,
create_streamable_http_app,
)
from fastmcp.server.middleware import MCPMiddleware, MiddlewareContext
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.settings import Settings
from fastmcp.tools import ToolManager
from fastmcp.tools.tool import FunctionTool, Tool
@ -118,7 +118,7 @@ class FastMCP(Generic[LifespanResultT]):
*,
version: str | None = None,
auth: OAuthProvider | None = None,
middleware: list[MCPMiddleware] | None = None,
middleware: list[Middleware] | None = None,
lifespan: (
Callable[
[FastMCP[LifespanResultT]],
@ -335,7 +335,7 @@ class FastMCP(Generic[LifespanResultT]):
chain = partial(mw, call_next=chain)
return await chain(context)
def add_middleware(self, middleware: MCPMiddleware) -> None:
def add_middleware(self, middleware: Middleware) -> None:
self.middleware.append(middleware)
async def get_tools(self) -> dict[str, Tool]:
@ -917,7 +917,7 @@ class FastMCP(Generic[LifespanResultT]):
Args:
template: A ResourceTemplate instance to add
"""
self._resource_manager.add_template(template)
self._resource_manager.add_template(template, key=key)
def add_resource_fn(
self,
@ -1260,7 +1260,7 @@ class FastMCP(Generic[LifespanResultT]):
log_level: str | None = None,
path: str | None = None,
uvicorn_config: dict[str, Any] | None = None,
middleware: list[Middleware] | None = None,
middleware: list[ASGIMiddleware] | None = None,
) -> None:
"""Run the server using HTTP transport.
@ -1331,7 +1331,7 @@ class FastMCP(Generic[LifespanResultT]):
self,
path: str | None = None,
message_path: str | None = None,
middleware: list[Middleware] | None = None,
middleware: list[ASGIMiddleware] | None = None,
) -> StarletteWithLifespan:
"""
Create a Starlette app for the SSE server.
@ -1361,7 +1361,7 @@ class FastMCP(Generic[LifespanResultT]):
def streamable_http_app(
self,
path: str | None = None,
middleware: list[Middleware] | None = None,
middleware: list[ASGIMiddleware] | None = None,
) -> StarletteWithLifespan:
"""
Create a Starlette app for the StreamableHTTP server.
@ -1382,7 +1382,7 @@ class FastMCP(Generic[LifespanResultT]):
def http_app(
self,
path: str | None = None,
middleware: list[Middleware] | None = None,
middleware: list[ASGIMiddleware] | None = None,
json_response: bool | None = None,
stateless_http: bool | None = None,
transport: Literal["streamable-http", "sse"] = "streamable-http",

View file

@ -7,7 +7,7 @@ import pytest
from fastmcp import Client, FastMCP
from fastmcp.server.context import Context
from fastmcp.server.middleware import MCPMiddleware, MiddlewareContext
from fastmcp.server.middleware import Middleware, MiddlewareContext
@dataclass
@ -18,7 +18,7 @@ class Recording:
result: mcp.types.ServerResult | None
class RecordingMiddleware(MCPMiddleware):
class RecordingMiddleware(Middleware):
"""A middleware that automatically records all method calls."""
def __init__(self, name: str | None = None):