当前位置：首页 > news >正文

14_FastMCP 2.x 中文文档之FastMCP高级功能：MCP中间件详解

news 2025/11/8 7:18:11

一、MCP 中间件

通过中间件为您的 MCP 服务器添加横切关注功能，该中间件可以检查、修改和响应所有 MCP 请求和响应。

新版本：2.9.0 功能

MCP 中间件是一个强大的概念，允许您为 FastMCP 服务器添加横切关注功能。与传统的 Web 中间件不同，MCP 中间件专门为模型上下文协议设计，为不同类型的 MCP 操作（如工具调用、资源读取和提示请求）提供钩子。

提示：MCP 中间件是 FastMCP 特有的概念，不是官方 MCP 协议规范的一部分。此中间件系统设计用于 FastMCP 服务器，可能与其他 MCP 实现不兼容。

警告：MCP 中间件是一个全新的概念，在未来的版本中可能会有破坏性变更。

二、什么是 MCP 中间件？

MCP 中间件允许您在请求和响应流经服务器时拦截和修改它们。可以将其视为一个管道，每个中间件都可以检查正在发生的情况，进行更改，然后将控制权传递给链中的下一个中间件。

MCP 中间件的常见用例包括：

身份验证和授权：在执行操作之前验证客户端权限
日志记录和监控：跟踪使用模式和性能指标
速率限制：控制每个客户端或操作类型的请求频率
请求/响应转换：在数据到达工具之前或离开之后修改数据
缓存：存储频繁请求的数据以提高性能
错误处理：在整个服务器中提供一致的错误响应

三、中间件如何工作

FastMCP 中间件基于管道模型运行。当请求进入时，它会按照添加到服务器的顺序流经您的中间件。每个中间件可以：

检查传入请求及其上下文
在将请求传递给下一个中间件或处理程序之前修改请求
通过调用 call_next() 执行链中的下一个中间件/处理程序
在返回响应之前检查和修改响应
处理处理过程中发生的错误

关键见解是中间件形成一个链，其中每个部分决定是继续处理还是完全停止链。

如果您熟悉 ASGI 中间件，FastMCP 中间件的基本结构会让您感到熟悉。核心上，中间件是一个可调用类，它接收包含当前 JSON-RPC 消息信息的上下文对象和一个用于继续中间件链的处理程序函数。

重要的是要理解 MCP 基于 JSON-RPC 规范运行。虽然 FastMCP 以熟悉的方式呈现请求和响应，但这些本质上是 JSON-RPC 消息，而不是您在 Web 应用程序中可能习惯的 HTTP 请求/响应对。FastMCP 中间件适用于所有传输类型，包括本地 stdio 传输和 HTTP 传输，尽管并非所有中间件实现都兼容所有传输（例如，检查 HTTP 头的中间件不适用于 stdio 传输）。

实现中间件的最基本方法是重写 Middleware 基类上的 call 方法：

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContextclass RawMiddleware(Middleware):async def __call__(self, context: MiddlewareContext, call_next):# 此方法接收所有消息，无论类型如何print(f"原始中间件处理：{context.method}")result = await call_next(context)print(f"原始中间件完成：{context.method}")return result

这使您可以完全控制流经服务器的每条消息，但需要您手动处理所有消息类型。

四、中间件钩子

为了使用户更容易针对特定类型的消息，FastMCP 中间件提供了各种专门的钩子。您可以通过重写特定的钩子方法而不是实现原始的 call 方法，这些钩子方法仅针对特定类型的操作调用，允许您精确定位中间件逻辑所需的特异性级别。

4.1 钩子层次结构和执行顺序

FastMCP 提供了多个具有不同特异性级别的钩子。理解此层次结构对于有效的中间件设计至关重要。

当请求进入时，同一请求可能会调用多个钩子，从一般到具体：

on_message - 为所有 MCP 消息调用（请求和通知）
on_request 或 on_notification - 根据消息类型调用
特定于操作的钩子 - 为特定的 MCP 操作调用，如 on_call_tool

例如，当客户端调用工具时，您的中间件将接收多个钩子调用：

任何初始工具发现操作（list_tools）的 on_message 和 on_request
工具调用本身的 on_message（因为它是任何 MCP 消息）
工具调用本身的 on_request（因为工具调用期望响应）
工具调用本身的 on_call_tool（因为它具体是工具执行）

请注意，MCP SDK 可能会执行其他操作，如出于缓存目的列出工具，这将触发除了直接工具执行之外的额外中间件调用。

此层次结构允许您以正确的特异性级别定位中间件逻辑。使用 on_message 处理广泛关注点（如日志记录），使用 on_request 处理身份验证，使用 on_call_tool 处理特定于工具的逻辑（如性能监控）。

4.2 可用钩子

版本 2.9.0 新增：

on_message：为所有 MCP 消息调用（请求和通知）
on_request：专门为 MCP 请求调用（期望响应）
on_notification：专门为 MCP 通知调用（即发即忘）
on_call_tool：在执行工具时调用
on_read_resource：在读取资源时调用
on_get_prompt：在检索提示时调用
on_list_tools：在列出可用工具时调用
on_list_resources：在列出可用资源时调用
on_list_resource_templates：在列出资源模板时调用
on_list_prompts：在列出可用提示时调用

版本 2.13.0 新增：

on_initialize：在客户端连接并初始化会话时调用（返回 None）

注意：on_initialize 钩子接收客户端的初始化请求但返回 None 而不是结果。初始化响应由 MCP 协议内部处理，不能被中间件修改。此钩子对于客户端检测、记录连接或初始化会话状态很有用，但不适用于修改初始化握手本身。

五、中间件中的组件访问

理解如何在中间件中访问组件信息（工具、资源、提示）对于构建强大的中间件功能至关重要。列出操作和执行操作的访问模式显著不同。

5.1 列出操作与执行操作

FastMCP 中间件以不同方式处理两种类型的操作：

列出操作（on_list_tools、on_list_resources、on_list_prompts 等）：

中间件接收具有完整元数据的 FastMCP 组件对象
这些对象包括 FastMCP 特定的属性，如可以直接从组件访问的 tags
结果包含在转换为 MCP 格式之前的完整组件信息
标签包含在返回给 MCP 客户端的列表响应中组件的 meta 字段中

执行操作（on_call_tool、on_read_resource、on_get_prompt）：

中间件在组件执行之前运行
中间件结果要么是执行结果，要么是如果未找到组件则返回错误
组件元数据在钩子参数中不可直接访问

5.2 执行期间访问组件元数据

如果需要在执行操作期间检查组件属性（如标签），请使用通过上下文可用的 FastMCP 服务器实例：

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolErrorclass TagBasedMiddleware(Middleware):async def on_call_tool(self, context: MiddlewareContext, call_next):# 访问工具对象以检查其元数据if context.fastmcp_context:try:tool = await context.fastmcp_context.fastmcp.get_tool(context.message.name)# 检查此工具是否有 "private" 标签if "private" in tool.tags:raise ToolError("访问被拒绝：私有工具")# 检查工具是否启用if not tool.enabled:raise ToolError("工具当前已禁用")except Exception:# 未找到工具或其他错误 - 让执行继续并自然处理错误passreturn await call_next(context)

相同的模式适用于资源和提示：

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ResourceError, PromptErrorclass ComponentAccessMiddleware(Middleware):async def on_read_resource(self, context: MiddlewareContext, call_next):if context.fastmcp_context:try:resource = await context.fastmcp_context.fastmcp.get_resource(context.message.uri)if "restricted" in resource.tags:raise ResourceError("访问被拒绝：受限资源")except Exception:passreturn await call_next(context)async def on_get_prompt(self, context: MiddlewareContext, call_next):if context.fastmcp_context:try:prompt = await context.fastmcp_context.fastmcp.get_prompt(context.message.name)if not prompt.enabled:raise PromptError("提示当前已禁用")except Exception:passreturn await call_next(context)

5.3 处理列出结果

对于列出操作，中间件 call_next 函数返回在转换为 MCP 格式之前的 FastMCP 组件列表。您可以过滤或修改此列表并将其返回给客户端。例如：

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContextclass ListingFilterMiddleware(Middleware):async def on_list_tools(self, context: MiddlewareContext, call_next):result = await call_next(context)# 过滤掉带有 "private" 标签的工具filtered_tools = [tool for tool in result if "private" not in tool.tags]# 返回修改后的列表return filtered_tools

此过滤发生在组件转换为 MCP 格式并返回给客户端之前。标签在过滤期间可访问，并包含在最终列表响应中组件的 meta 字段中。

提示：在列出操作中过滤组件时，请确保还在相应的执行钩子（on_call_tool、on_read_resource、on_get_prompt）中防止过滤组件的执行，以保持一致性。

5.4 工具调用拒绝

您可以通过在中间件中引发 ToolError 来拒绝对特定工具的访问。这是阻止工具执行的正确方法，因为它与 FastMCP 错误处理系统正确集成。

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolErrorclass AuthMiddleware(Middleware):async def on_call_tool(self, context: MiddlewareContext, call_next):tool_name = context.message.name# 拒绝对受限工具的访问if tool_name.lower() in ["delete", "admin_config"]:raise ToolError("访问被拒绝：工具需要管理员权限")# 允许其他工具继续return await call_next(context)

警告：拒绝工具调用时，始终引发 ToolError 而不是返回 ToolResult 对象或其他值。ToolError 确保通过中间件链正确传播错误，并转换为正确的 MCP 错误响应格式。

5.5 工具调用修改

对于工具调用等执行操作，您可以在执行前修改参数或在执行后转换结果：

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContextclass ToolCallMiddleware(Middleware):async def on_call_tool(self, context: MiddlewareContext, call_next):# 在执行前修改参数if context.message.name == "calculate":# 确保正输入if context.message.arguments.get("value", 0) < 0:context.message.arguments["value"] = abs(context.message.arguments["value"])result = await call_next(context)# 在执行后转换结果if context.message.name == "get_data":# 向结果添加元数据if result.structured_content:result.structured_content["processed_at"] = "2024-01-01T00:00:00Z"return result

提示：对于更复杂的工具重写场景，请考虑使用工具转换模式，它提供了创建修改工具变体的更结构化方法。

六、钩子剖析

每个中间件钩子都遵循相同的模式。让我们检查 on_message 钩子以了解结构：

示例代码：

async def on_message(self, context: MiddlewareContext, call_next):# 1. 预处理：检查并可选择修改请求print(f"处理 {context.method}")# 2. 链继续：调用下一个中间件/处理程序result = await call_next(context)# 3. 后处理：检查并可选择修改响应print(f"完成 {context.method}")# 4. 返回结果（可能已修改）return result

6.1 钩子参数

每个钩子接收两个参数：

context: MiddlewareContext - 包含当前请求信息：
context.method - MCP 方法名称（例如 “tools/call”）
context.source - 请求来源（“client” 或 “server”）
context.type - 消息类型（“request” 或 “notification”）
context.message - MCP 消息数据
context.timestamp - 接收请求的时间
context.fastmcp_context - FastMCP 上下文对象（如果可用）
call_next - 继续中间件链的函数。您必须调用此函数以继续，除非您想完全停止处理。

6.2 控制流

您对请求流有完全控制：

继续处理：调用 await call_next(context) 以继续
修改请求：在调用 call_next 之前更改上下文
修改响应：在调用 call_next 之后更改结果
停止链：不调用 call_next（很少需要）
处理错误：在 try/catch 块中包装 call_next

状态管理

版本 2.11.0 新增：

除了修改请求和响应之外，您还可以存储状态数据，您的工具可以（可选地）稍后访问。为此，使用 FastMCP 上下文适当地 set_state 或 get_state。有关更多信息，请参阅上下文状态管理文档。

七、创建中间件

FastMCP 中间件通过子类化 Middleware 基类并重写您需要的钩子来实现。您只需要实现与您的用例相关的钩子。

示例代码：

from fastmcp import FastMCP
from fastmcp.server.middleware import Middleware, MiddlewareContextclass LoggingMiddleware(Middleware):"""记录所有 MCP 操作的中间件。"""async def on_message(self, context: MiddlewareContext, call_next):"""为所有 MCP 消息调用。"""print(f"处理来自 {context.source} 的 {context.method}")result = await call_next(context)print(f"完成 {context.method}")return result# 将中间件添加到您的服务器
mcp = FastMCP("MyServer")
mcp.add_middleware(LoggingMiddleware())

这将创建一个基本的日志记录中间件，它将打印流经服务器的每个请求的信息。

7.1 将中间件添加到服务器

单个中间件

将中间件添加到服务器很简单：

示例代码：

mcp = FastMCP("MyServer")
mcp.add_middleware(LoggingMiddleware())

多个中间件

中间件按照添加到服务器的顺序执行。第一个添加的中间件在进入时首先运行，在退出时最后运行：

示例代码：

mcp = FastMCP("MyServer")mcp.add_middleware(AuthenticationMiddleware("secret-token"))
mcp.add_middleware(PerformanceMiddleware())
mcp.add_middleware(LoggingMiddleware())

这将创建以下执行流：

AuthenticationMiddleware（预处理）
PerformanceMiddleware（预处理）
LoggingMiddleware（预处理）
实际工具/资源处理程序
LoggingMiddleware（后处理）
PerformanceMiddleware（后处理）
AuthenticationMiddleware（后处理）

八、服务器组合和中间件

当使用带有 mount 或 import_server 的服务器组合时，中间件行为遵循以下规则：

父服务器中间件为所有请求运行，包括路由到已挂载服务器的请求
已挂载服务器中间件仅为由该特定服务器处理的请求运行
在每个服务器内保留中间件顺序

这允许您创建分层中间件架构，其中父服务器处理横切关注点（如身份验证），而子服务器专注于特定领域的中间件。

示例代码：

# 带有中间件的父服务器
parent = FastMCP("Parent")
parent.add_middleware(AuthenticationMiddleware("token"))# 带有自己中间件的子服务器
child = FastMCP("Child")
child.add_middleware(LoggingMiddleware())@child.tool
def child_tool() -> str:return "来自子服务器"# 挂载子服务器
parent.mount(child, prefix="child")

当客户端调用 “child_tool” 时，请求将首先流经父服务器的身份验证中间件，然后路由到子服务器，在那里它将通过子服务器的日志记录中间件。

8.1 内置中间件示例

FastMCP 包含几个中间件实现，演示了最佳实践并提供了立即可用的功能。让我们通过构建简化版本来探索每种类型的工作原理，然后了解如何使用完整实现。

8.2 计时中间件

性能监控对于理解服务器行为和识别瓶颈至关重要。FastMCP 在fastmcp.server.middleware.timing 包含计时中间件。

以下是其工作原理示例：

示例代码：

import time
from fastmcp.server.middleware import Middleware, MiddlewareContextclass SimpleTimingMiddleware(Middleware):async def on_request(self, context: MiddlewareContext, call_next):start_time = time.perf_counter()try:result = await call_next(context)duration_ms = (time.perf_counter() - start_time) * 1000print(f"请求 {context.method} 在 {duration_ms:.2f}ms 内完成")return resultexcept Exception as e:duration_ms = (time.perf_counter() - start_time) * 1000print(f"请求 {context.method} 在 {duration_ms:.2f}ms 后失败：{e}")raise

要使用具有适当日志记录和配置的完整版本：

示例代码：

from fastmcp.server.middleware.timing import (TimingMiddleware,DetailedTimingMiddleware
)# 所有请求的基本计时
mcp.add_middleware(TimingMiddleware())# 每个操作的详细计时（工具、资源、提示）
mcp.add_middleware(DetailedTimingMiddleware())

内置版本包括自定义记录器支持、正确格式化，并且 DetailedTimingMiddleware 提供了特定于操作的钩子，如 on_call_tool 和 on_read_resource，用于细粒度计时。

8.3 工具注入中间件

工具注入中间件是在请求生命周期期间向服务器注入工具的中间件：

示例代码：

from fastmcp.server.middleware.tool_injection import ToolInjectionMiddlewaredef my_tool_fn(a: int, b: int) -> int:return a + bmy_tool = Tool.from_function(fn=my_tool_fn, name="my_tool")mcp.add_middleware(ToolInjectionMiddleware(tools=[my_tool]))

8.4 提示工具中间件

提示工具中间件是针对无法列出或获取提示的客户端的兼容性中间件。它提供了两个工具：list_prompts 和 get_prompt，允许客户端仅使用工具调用列出和获取提示。

示例代码：

from fastmcp.server.middleware.tool_injection import PromptToolMiddlewaremcp.add_middleware(PromptToolMiddleware())

8.5 资源工具中间件

资源工具中间件是针对无法列出或读取资源的客户端的兼容性中间件。它提供了两个工具：list_resources 和 read_resource，允许客户端仅使用工具调用列出和读取资源。

示例代码：

from fastmcp.server.middleware.tool_injection import ResourceToolMiddlewaremcp.add_middleware(ResourceToolMiddleware())

8.6 缓存中间件

缓存中间件对于提高性能和减少服务器负载至关重要。FastMCP 在 fastmcp.server.middleware.caching 提供缓存中间件。

以下是使用完整版本的方法：

示例代码：

from fastmcp.server.middleware.caching import ResponseCachingMiddlewaremcp.add_middleware(ResponseCachingMiddleware())

开箱即用，它将调用/列表工具、资源和提示缓存到具有基于 TTL 的过期的内存缓存中。缓存条目基于其 TTL 过期；没有基于事件的缓存失效。列表调用存储在全局键下——在多个服务器之间共享存储后端时，考虑命名空间集合以防止冲突。有关高级配置选项，请参阅存储后端。

可以单独配置每个方法，例如，将列表工具缓存 30 秒，将缓存限制为特定工具，并禁用资源读取的缓存：

示例代码：

from fastmcp.server.middleware.caching import ResponseCachingMiddleware, CallToolSettings, ListToolsSettings, ReadResourceSettingsmcp.add_middleware(ResponseCachingMiddleware(list_tools_settings=ListToolsSettings(ttl=30,),call_tool_settings=CallToolSettings(included_tools=["tool1"],),read_resource_settings=ReadResourceSettings(enabled=False)
))

8.6.1 存储后端

默认情况下，缓存使用内存存储，这很快但不会在重启之间持久化。对于生产或在服务器重启之间持久缓存，请配置不同的存储后端。有关完整选项，包括磁盘、Redis、DynamoDB 和自定义实现，请参阅存储后端。

基于磁盘的缓存示例：

示例代码：

from fastmcp.server.middleware.caching import ResponseCachingMiddleware
from key_value.aio.stores.disk import DiskStoremcp.add_middleware(ResponseCachingMiddleware(cache_storage=DiskStore(directory="cache"),
))

用于分布式部署的 Redis：

示例代码：

from fastmcp.server.middleware.caching import ResponseCachingMiddleware
from key_value.aio.stores.redis import RedisStoremcp.add_middleware(ResponseCachingMiddleware(cache_storage=RedisStore(host="redis.example.com", port=6379),
))

8.6.2 缓存统计信息

缓存中间件通过底层存储层收集操作统计信息（命中、未命中等）。从中间件实例访问统计信息：

示例代码：

from fastmcp.server.middleware.caching import ResponseCachingMiddlewaremiddleware = ResponseCachingMiddleware()
mcp.add_middleware(middleware)# 稍后，检索统计信息
stats = middleware.statistics()
print(f"总缓存操作：{stats}")

8.7 日志记录中间件

请求和响应日志记录对于调试、监控和理解 MCP 服务器中的使用模式至关重要。FastMCP 在 fastmcp.server.middleware.logging 提供全面的日志记录中间件。

以下是其工作原理示例：

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContextclass SimpleLoggingMiddleware(Middleware):async def on_message(self, context: MiddlewareContext, call_next):print(f"处理来自 {context.source} 的 {context.method}")try:result = await call_next(context)print(f"完成 {context.method}")return resultexcept Exception as e:print(f"失败 {context.method}：{e}")raise

要使用具有高级功能的完整版本：

示例代码：

from fastmcp.server.middleware.logging import (LoggingMiddleware,StructuredLoggingMiddleware
)# 人类可读的日志记录，支持负载
mcp.add_middleware(LoggingMiddleware(include_payloads=True,max_payload_length=1000
))# 用于日志聚合工具的 JSON 结构化日志记录
mcp.add_middleware(StructuredLoggingMiddleware(include_payloads=True))

内置版本包括负载日志记录、结构化 JSON 输出、自定义记录器支持、负载大小限制和用于细粒度控制的特定于操作的钩子。

8.8 速率限制中间件

速率限制对于保护服务器免受滥用、确保公平资源使用和在负载下保持性能至关重要。FastMCP 在 fastmcp.server.middleware.rate_limiting 包含复杂的速率限制中间件。

以下是其工作原理示例：

示例代码：

import time
from collections import defaultdict
from fastmcp.server.middleware import Middleware, MiddlewareContext
from mcp import McpError
from mcp.types import ErrorDataclass SimpleRateLimitMiddleware(Middleware):def __init__(self, requests_per_minute: int = 60):self.requests_per_minute = requests_per_minuteself.client_requests = defaultdict(list)async def on_request(self, context: MiddlewareContext, call_next):current_time = time.time()client_id = "default"  # 实际上，从头或上下文提取# 清理旧请求并检查限制cutoff_time = current_time - 60self.client_requests[client_id] = [req_time for req_time in self.client_requests[client_id]if req_time > cutoff_time]if len(self.client_requests[client_id]) >= self.requests_per_minute:raise McpError(ErrorData(code=-32000, message="超出速率限制"))self.client_requests[client_id].append(current_time)return await call_next(context)

要使用具有高级算法的完整版本：

示例代码：

from fastmcp.server.middleware.rate_limiting import (RateLimitingMiddleware,SlidingWindowRateLimitingMiddleware
)# 令牌桶速率限制（允许受控突发）
mcp.add_middleware(RateLimitingMiddleware(max_requests_per_second=10.0,burst_capacity=20
))# 滑动窗口速率限制（精确的基于时间的控制）
mcp.add_middleware(SlidingWindowRateLimitingMiddleware(max_requests=100,window_minutes=1
))

内置版本包括令牌桶算法、每个客户端识别、全局速率限制和具有可配置客户端识别函数的异步安全实现。

8.9 错误处理中间件

一致的错误处理和恢复对于健壮的 MCP 服务器至关重要。FastMCP 在 fastmcp.server.middleware.error_handling 提供全面的错误处理中间件。

以下是其工作原理示例：

示例代码：

import logging
from fastmcp.server.middleware import Middleware, MiddlewareContextclass SimpleErrorHandlingMiddleware(Middleware):def __init__(self):self.logger = logging.getLogger("errors")self.error_counts = {}async def on_message(self, context: MiddlewareContext, call_next):try:return await call_next(context)except Exception as error:# 记录错误并跟踪统计信息error_key = f"{type(error).__name__}:{context.method}"self.error_counts[error_key] = self.error_counts.get(error_key, 0) + 1self.logger.error(f"{context.method} 中的错误：{type(error).__name__}：{error}")raise

要使用具有高级功能的完整版本：

示例代码：

from fastmcp.server.middleware.error_handling import (ErrorHandlingMiddleware,RetryMiddleware
)# 全面的错误日志记录和转换
mcp.add_middleware(ErrorHandlingMiddleware(include_traceback=True,transform_errors=True,error_callback=my_error_callback
))# 具有指数退避的自动重试
mcp.add_middleware(RetryMiddleware(max_retries=3,retry_exceptions=(ConnectionError, TimeoutError)
))

内置版本包括错误转换、自定义回调、可配置的重试逻辑和正确的 MCP 错误格式化。

8.10 组合中间件

这些中间件无缝地协同工作：

示例代码：

from fastmcp import FastMCP
from fastmcp.server.middleware.timing import TimingMiddleware
from fastmcp.server.middleware.logging import LoggingMiddleware
from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware
from fastmcp.server.middleware.error_handling import ErrorHandlingMiddlewaremcp = FastMCP("生产服务器")# 以逻辑顺序添加中间件
mcp.add_middleware(ErrorHandlingMiddleware())  # 首先处理错误
mcp.add_middleware(RateLimitingMiddleware(max_requests_per_second=50))
mcp.add_middleware(TimingMiddleware())  # 计时实际执行
mcp.add_middleware(LoggingMiddleware())  # 记录所有内容@mcp.tool
def my_tool(data: str) -> str:return f"已处理：{data}"

此配置为您的 MCP 服务器提供了全面的监控、保护和可观察性。

8.11 自定义中间件示例

您还可以通过扩展基类创建自定义中间件：

示例代码：

from fastmcp.server.middleware import Middleware, MiddlewareContextclass CustomHeaderMiddleware(Middleware):async def on_request(self, context: MiddlewareContext, call_next):# 在此处添加自定义逻辑print(f"处理 {context.method}")result = await call_next(context)print(f"完成 {context.method}")return resultmcp.add_middleware(CustomHeaderMiddleware())