[MCP][07]logging和progress等功能说明
<h2 id="前言">前言</h2><p>截至目前（2025年9月19日），除了基础的Prompt、Resource和Tool概念，FastMCP还提供了以下功能：Sampling、Elicitation、Roots、Logging、Progress、Proxy、Middleware、Composition和Authentication等功能</p>
<ul>
<li>Sampling，采样，在server端调用client的llm，实现功能解耦</li>
<li>Elicition，征询，实现人工介入</li>
<li>Roots，Client告知Server可访问的资源</li>
<li>Logging，将Server日志发送给Client</li>
<li>Progress，Server端将进度发送给Client</li>
<li>Proxy，代理其它MCP Server</li>
<li>Middleware，拦截MCP通信中的请求和响应</li>
<li>Composition，Server端将多个servers组合成一个server对外提供</li>
<li>Authentication，Client和Server之间安全认证</li>
</ul>
<p>其中Sampling和Elicitation在我的实际开发中用到的比较多，所以我在前面章节中单独拎出来介绍了。FastMCP官方文档也说了Authentication还在迅速迭代中，虽然已经有了相关文档，但本文暂时就不涉及了，等这个功能稳定了再具体细说。剩下的功能会在本文中一次性全部介绍完，篇幅较长，可以根据章节名跳转到自己需要关注的内容。<strong>本文大部分参考自官方文档</strong>。</p>
<h2 id="roots">Roots</h2>
<p>Roots 是客户端向服务器告知其可访问资源的一种机制。服务器可利用此信息调整行为或提供更相关的响应。</p>
<h3 id="静态roots">静态Roots</h3>
<pre><code>from fastmcp import Client
client = Client(
"my_mcp_server.py",
roots=["/path/to/root1", "/path/to/root2"]
)
</code></pre>
<h3 id="动态roots">动态Roots</h3>
<pre><code class="language-python">from fastmcp import Client
from fastmcp.client.roots import RequestContext
async def roots_callback(context: RequestContext) -> list:
print(f"Server requested roots (Request ID: {context.request_id})")
return ["/path/to/root1", "/path/to/root2"]
client = Client(
"my_mcp_server.py",
roots=roots_callback
)
</code></pre>
<h2 id="logging">Logging</h2>
<p>Logging，从服务器向 MCP 客户端发送消息。FastMCP提供了一个logger（<code>fastmcp.utilities.logging.get_logger()</code>），也可以用python标准库的<code>logging</code>。</p>
<p>服务器日志功能允许 MCP 工具向客户端发送调试（debug）、信息（info）、警告（warning）和错误（error）级别的消息。这有助于用户了解函数执行过程，在开发和运行阶段辅助调试。一般用于以下场景：</p>
<ul>
<li><strong>调试</strong>：发送详细的执行信息，帮助诊断问题</li>
<li><strong>进度可见性</strong>：让用户了解工具当前正在执行的操作</li>
<li><strong>错误报告</strong>：向客户端传达问题及其上下文</li>
<li><strong>审计追踪</strong>：为合规或分析目的生成工具执行记录</li>
</ul>
<p>与标准 Python 日志不同，MCP 服务器 Logging 会直接将消息发送至客户端，使其在客户端界面或日志中可见。</p>
<h3 id="server-示例">Server 示例</h3>
<p>在任意tool函数中使用Context提供的日志方法：</p>
<pre><code class="language-python">from fastmcp import FastMCP, Context
mcp = FastMCP("custom")
@mcp.tool
async def analyze_data(data: list, ctx: Context) -> dict:
"""通过全面日志记录分析数值数据。"""
await ctx.debug("开始分析数值数据")
await ctx.info(f"正在分析 {len(data)} 个数据点")
try:
if not data:
await ctx.warning("提供了空数据列表")
return {"error": "空数据列表"}
result = sum(data) / len(data)
await ctx.info(f"分析完成，平均值为：{result}")
return {"average": result, "count": len(data)}
except Exception as e:
await ctx.error(f"分析失败：{str(e)}")
raise
if __name__ == "__main__":
mcp.run(transport="stdio", show_banner=False)
</code></pre>
<p>所有日志方法（<code>debug</code>、<code>info</code>、<code>warning</code>、<code>error</code>、<code>log</code>）现在均支持 <code>extra</code> 参数，该参数接受一个字典，用于传递任意结构化数据。这使得客户端可接收结构化日志，便于创建丰富且可查询的日志记录。</p>
<pre><code class="language-python">@mcp.tool
async def process_transaction(transaction_id: str, amount: float, ctx: Context):
await ctx.info(
f"正在处理交易 {transaction_id}",
extra={
"transaction_id": transaction_id,
"amount": amount,
"currency": "USD"
}
)
# ... 处理逻辑 ...
</code></pre>
<h3 id="client-示例">Client 示例</h3>
<pre><code class="language-python">import asyncio
from pathlib import Path
from fastmcp.client import Client, StdioTransport
from fastmcp.client.logging import LogMessage
import logging
import sys
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler(sys.stdout))
# This mapping is useful for converting MCP level strings to Python's levels
LOGGING_LEVEL_MAP = logging.getLevelNamesMapping()
class MCPClient:
def __init__(self):
self.mcp_client = Client(
StdioTransport(
command = str(Path(__file__).parent.parent / ".venv" / "bin" / "python"),
args = ["demo09-server.py"],
cwd = str(Path(__file__).parent)
),
log_handler=self.logging_handler,
)
async def logging_handler(self, message: LogMessage):
"""
Handles incoming logs from the MCP server and forwards them
to the standard Python logging system.
"""
msg = message.data.get('msg')
extra = message.data.get('extra')
# Convert the MCP log level to a Python log level
level = LOGGING_LEVEL_MAP.get(message.level.upper(), logging.INFO)
logger.log(level, msg, extra=extra)
async def generate(self):
async with self.mcp_client:
await self.mcp_client.ping()
rst = await self.mcp_client.call_tool("analyze_data", arguments={"data": })
print(rst)
async def main():
client = MCPClient()
await client.generate()
if __name__ == "__main__":
asyncio.run(main())
</code></pre>
<p>client运行输出</p>
<pre><code>开始分析数值数据
正在分析 5 个数据点
分析完成，平均值为：3.0
CallToolResult(content=, structured_content={'average': 3.0, 'count': 5}, data={'average': 3.0, 'count': 5}, is_error=False)
</code></pre>
<h2 id="progress">Progress</h2>
<p>Progress 功能允许 MCPtool 向 Client 通知长时间运行操作的当前进度。这使得Client能够显示进度指示器，从而在执行耗时任务时提供更佳的用户体验。Progress 在以下方面具有重要价值：</p>
<ul>
<li><strong>用户体验</strong>：让用户了解长时间运行操作的当前状态</li>
<li><strong>进度指示器</strong>：使客户端能够显示进度条或百分比</li>
<li><strong>防止超时</strong>：表明操作正在持续进行中，避免被误判为无响应</li>
<li><strong>调试用途</strong>：追踪执行进度，便于性能分析</li>
</ul>
<h3 id="server-示例-1">Server 示例</h3>
<pre><code class="language-python">from fastmcp import FastMCP, Context
import asyncio
mcp = FastMCP("custom")
@mcp.tool
async def process_items(items: list, ctx: Context) -> dict:
"""处理项目列表，并发送进度更新。"""
total = len(items)
results = []
for i, item in enumerate(items):
# 每处理一个项目，报告当前进度
await ctx.report_progress(progress=i, total=total)
# 模拟处理耗时
await asyncio.sleep(0.1)
results.append(item.upper())
# 报告 100% 完成
await ctx.report_progress(progress=total, total=total)
return {"processed": len(results), "results": results}
if __name__ == "__main__":
mcp.run(transport="stdio", show_banner=False)
</code></pre>
<h3 id="client-示例-1">Client 示例</h3>
<pre><code class="language-python">import asyncio
from pathlib import Path
from fastmcp.client import Client, StdioTransport
class MCPClient:
def __init__(self):
self.mcp_client = Client(
StdioTransport(
command = str(Path(__file__).parent.parent / ".venv" / "bin" / "python"),
args = ["demo09-server.py"],
cwd = str(Path(__file__).parent)
),
progress_handler=self.progress_handler,
)
async def progress_handler(self, progress: float, total: float | None, message: str | None) -> None:
if total is not None:
percentage = (progress / total) * 100
print(f"Progress: {percentage:.1f}% - {message or ''}")
else:
print(f"Progress: {progress} - {message or ''}")
async def generate(self):
async with self.mcp_client:
await self.mcp_client.ping()
rst = await self.mcp_client.call_tool("process_items", arguments={"items": ["item1", "item2", "item3", "item4", "item5", "item6", "item7", "item8", "item9", "item10", "item11", "item12", "item13", "item14", "item15"]})
print(rst)
async def main():
client = MCPClient()
await client.generate()
if __name__ == "__main__":
asyncio.run(main())
</code></pre>
<p>client运行输出</p>
<pre><code>Progress: 0.0% -
Progress: 6.7% -
Progress: 13.3% -
Progress: 20.0% -
Progress: 26.7% -
Progress: 33.3% -
Progress: 40.0% -
Progress: 46.7% -
Progress: 53.3% -
Progress: 60.0% -
Progress: 66.7% -
Progress: 73.3% -
Progress: 80.0% -
Progress: 86.7% -
Progress: 93.3% -
Progress: 100.0% -
CallToolResult(content=}', annotations=None, meta=None)], structured_content={'processed': 15, 'results': ['ITEM1', 'ITEM2', 'ITEM3', 'ITEM4', 'ITEM5', 'ITEM6', 'ITEM7', 'ITEM8', 'ITEM9', 'ITEM10', 'ITEM11', 'ITEM12', 'ITEM13', 'ITEM14', 'ITEM15']}, data={'processed': 15, 'results': ['ITEM1', 'ITEM2', 'ITEM3', 'ITEM4', 'ITEM5', 'ITEM6', 'ITEM7', 'ITEM8', 'ITEM9', 'ITEM10', 'ITEM11', 'ITEM12', 'ITEM13', 'ITEM14', 'ITEM15']}, is_error=False)
</code></pre>
<h2 id="proxy">Proxy</h2>
<p>FastMCP 的 Proxy 允许一个 FastMCP 服务器实例作为前端，代理另一个 MCP 服务器（该服务器可能是远程的、运行在不同传输协议上的，甚至是另一个 FastMCP 实例）。此功能通过 <code>FastMCP.as_proxy()</code> 类方法实现。作为代理服务器，它本身不直接实现工具或资源。当它接收到请求（如 <code>tools/call</code> 或 <code>resources/read</code>）时，会将该请求转发至一个_后端_ MCP 服务器，接收其响应，再将响应原样返回给原始客户端。</p>
<div class="mermaid">sequenceDiagram
participant ClientApp as 您的客户端（如 Claude Desktop）
participant FastMCPProxy as FastMCP 代理服务器
participant BackendServer as 后端 MCP 服务器（如远程 SSE）
ClientApp->>FastMCPProxy: MCP 请求（如 stdio）
Note over FastMCPProxy, BackendServer: 代理转发请求
FastMCPProxy->>BackendServer: MCP 请求（如 sse）
BackendServer-->>FastMCPProxy: MCP 响应（如 sse）
Note over ClientApp, FastMCPProxy: 代理转发响应
FastMCPProxy-->>ClientApp: MCP 响应（如 stdio）
</div><p>核心优势</p>
<ul>
<li><strong>会话隔离</strong>：每个请求拥有独立隔离的会话，确保并发操作安全</li>
<li><strong>传输协议桥接</strong>：通过一种传输协议暴露运行在另一种传输协议上的服务器</li>
<li><strong>高级 MCP 功能支持</strong>：自动转发采样（sampling）、引导（elicitation）、日志和进度报告</li>
<li><strong>安全性</strong>：作为后端服务器的受控网关</li>
<li><strong>简化架构</strong>：即使后端位置或传输协议变更，前端仍保持单一接入点</li>
</ul>
<blockquote>
<p>使用代理服务器时，特别是连接到基于 HTTP 的后端服务器时，需注意延迟可能显著增加。例如，<code>list_tools()</code> 操作可能耗时数百毫秒，而本地工具仅需 1–2 毫秒。挂载代理服务器时，此延迟会影响父服务器的所有操作，而不仅仅是与被代理工具的交互。<br>
如果您的使用场景对低延迟有严格要求，建议使用 <code>import_server()</code> 方法在启动时复制工具，而非在运行时进行代理。</p>
</blockquote>
<h3 id="快速入门">快速入门</h3>
<p>推荐使用 <code>ProxyClient</code> 创建代理，它提供完整的 MCP 功能支持，并自动实现会话隔离：</p>
<pre><code class="language-python">from fastmcp import FastMCP
from fastmcp.server.proxy import ProxyClient
# 创建支持完整 MCP 功能的代理
proxy = FastMCP.as_proxy(
ProxyClient("backend_server.py"),
name="MyProxy"
)
# 运行代理（例如，通过 stdio 供 Claude Desktop 使用）
if __name__ == "__main__":
proxy.run()
</code></pre>
<p>此单一设置即可提供：</p>
<ul>
<li>安全的并发请求处理</li>
<li>自动转发高级 MCP 功能（采样、引导等）</li>
<li>会话隔离，防止上下文混淆</li>
<li>与所有 MCP 客户端完全兼容</li>
</ul>
<h3 id="高级mcp功能支持">高级MCP功能支持</h3>
<p><code>ProxyClient</code> 会自动在后端服务器与连接到代理的客户端之间转发高级 MCP 协议功能，确保完整的 MCP 兼容性。支持的功能：</p>
<ul>
<li><strong>Roots</strong>：将文件系统根目录访问请求转发给客户端</li>
<li><strong>Sampling</strong>：将后端发起的 LLM 补全请求转发给客户端</li>
<li><strong>Elicitation</strong>：将用户输入请求转发给客户端</li>
<li><strong>Logging</strong>：将后端日志消息转发至客户端</li>
<li><strong>Progress</strong>：在长时间操作中转发进度通知</li>
</ul>
<p>也可以自定义功能支持，比如设置为None来选择性禁用转发</p>
<pre><code class="language-python"># 禁用采样，但保留其他功能
backend = ProxyClient(
"backend_server.py",
sampling_handler=None,# 禁用 LLM 采样转发
log_handler=None # 禁用日志转发
)
</code></pre>
<h3 id="基于配置的代理">基于配置的代理</h3>
<p>你可以直接从符合 MCPConfig 模式的配置字典创建代理。这对于快速设置指向远程服务器的代理非常有用，无需手动配置每个连接细节。</p>
<pre><code class="language-python">from fastmcp import FastMCP
# 直接从配置字典创建代理
config = {
"mcpServers": {
"default": {# 对于单服务器配置，通常使用 'default'
"url": "https://example.com/mcp",
"transport": "http"
}
}
}
# 创建指向配置服务器的代理（自动创建 ProxyClient）
proxy = FastMCP.as_proxy(config, name="Config-Based Proxy")
# 通过 stdio 传输协议本地运行
if __name__ == "__main__":
proxy.run()
</code></pre>
<h3 id="多服务器的设置">多服务器的设置</h3>
<p>你可以通过在配置中指定多个条目来创建指向多个服务器的代理。系统会自动以配置名称作为前缀挂载它们：</p>
<pre><code class="language-python"># 多服务器配置
config = {
"mcpServers": {
"weather": {
"url": "https://weather-api.example.com/mcp",
"transport": "http"
},
"calendar": {
"url": "https://calendar-api.example.com/mcp",
"transport": "http"
}
}
}
# 创建统一的多服务器代理
composite_proxy = FastMCP.as_proxy(config, name="Composite Proxy")
# 工具和资源可通过前缀访问：
# - weather_get_forecast, calendar_add_event
# - weather://weather/icons/sunny, calendar://calendar/events/today
</code></pre>
<h3 id="显式会话管理">显式会话管理</h3>
<p>在内部，<code>FastMCP.as_proxy()</code> 使用 <code>FastMCPProxy</code> 类。您通常无需直接与此类交互，但在高级场景下它可供使用。<code>FastMCPProxy</code> 要求显式会话管理——不会执行任何自动检测。您必须选择您的会话策略：</p>
<pre><code class="language-python"># 在所有请求间共享会话（并发时需谨慎）
shared_client = ProxyClient("backend_server.py")
def shared_session_factory():
return shared_client
proxy = FastMCPProxy(client_factory=shared_session_factory)
# 为每个请求创建新会话（推荐）
def fresh_session_factory():
return ProxyClient("backend_server.py")
proxy = FastMCPProxy(client_factory=fresh_session_factory)
</code></pre>
<p>如需自动选择会话策略，请使用便捷方法 <code>FastMCP.as_proxy()</code>。</p>
<pre><code class="language-python"># 带有特定配置的自定义工厂
def custom_client_factory():
client = ProxyClient("backend_server.py")
# 在此处添加任何自定义配置
return client
proxy = FastMCPProxy(client_factory=custom_client_factory)
</code></pre>
<h2 id="middleware">Middleware</h2>
<p>MCP 中间件允许您在请求和响应流经服务器时对其进行拦截和修改。可以将其视为一条管道，每个中间件均可检查当前操作、进行修改，然后将控制权传递给链中的下一个中间件。与传统的 Web 中间件不同，MCP 中间件专为 Model Context Protocol 设计，为各类 MCP 操作（如工具调用、资源读取和提示请求）提供专用钩子。</p>
<blockquote>
<p>MCP 中间件是一个全新概念，未来版本中可能发生破坏性变更。</p>
</blockquote>
<p>MCP 中间件的常见应用场景包括：</p>
<ul>
<li><strong>身份验证与授权</strong>：在执行操作前验证客户端权限</li>
<li><strong>日志与监控</strong>：追踪使用模式与性能指标</li>
<li><strong>速率限制</strong>：按客户端或操作类型控制请求频率</li>
<li><strong>请求/响应转换</strong>：在数据到达工具前或离开后对其进行修改</li>
<li><strong>缓存</strong>：存储频繁请求的数据以提升性能</li>
<li><strong>错误处理</strong>：为服务器提供一致的错误响应</li>
</ul>
<h3 id="中间件工作原理">中间件工作原理</h3>
<p>FastMCP 中间件基于管道模型运行。当请求进入时，它会按添加到服务器的顺序依次流经各个中间件。每个中间件均可：</p>
<pre><code>检查传入的请求及其上下文
在传递给下一个中间件或处理器前修改请求
通过调用 call_next() 执行链中的下一个中间件/处理器
在返回前检查并修改响应
处理执行过程中发生的错误
</code></pre>
<p>关键在于，中间件形成一条链，每个环节决定是继续处理还是完全终止链的执行。</p>
<p>如果你熟悉 ASGI 中间件，FastMCP 中间件的基本结构会感觉似曾相识。其核心是一个可调用类，接收一个包含当前 JSON-RPC 消息信息的上下文对象，以及一个用于继续中间件链的处理器函数。</p>
<p>重要的是要理解，MCP 基于 JSON-RPC 规范运行。虽然 FastMCP 以熟悉的方式呈现请求和响应，但其本质是 JSON-RPC 消息，而非 Web 应用中常见的 HTTP 请求/响应对。FastMCP 中间件适用于所有 传输类型 ，包括本地 stdio 传输和 HTTP 传输，但并非所有中间件实现都兼容所有传输类型（例如，检查 HTTP 头部的中间件无法在 stdio 传输中工作）。</p>
<p>实现中间件最基础的方式是重写 Middleware 基类的 <strong>call</strong> 方法：</p>
<pre><code class="language-python">from fastmcp.server.middleware import Middleware, MiddlewareContext
class 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
</code></pre>
<h3 id="中间件钩子">中间件钩子</h3>
<p>为便于用户针对特定类型的消息，FastMCP 中间件提供了一系列专用钩子。您可以重写特定的钩子方法（而非实现原始的 <code>__call__</code> 方法），这些方法仅在特定类型的操作时被调用，从而允许您精确地定位中间件逻辑所需的粒度。</p>
<h4 id="钩子层级与执行顺序">钩子层级与执行顺序</h4>
<p>FastMCP 提供多个按不同粒度调用的钩子。理解此层级结构对有效设计中间件至关重要。</p>
<p>当请求进入时，<strong>同一请求可能触发多个钩子调用</strong>，执行顺序由泛化到具体：</p>
<ol>
<li><code>on_message</code> - 为所有 MCP 消息（请求和通知）调用</li>
<li><code>on_request</code> 或 <code>on_notification</code> - 根据消息类型调用</li>
<li>操作特定钩子 - 为特定 MCP 操作调用，如 <code>on_call_tool</code></li>
</ol>
<p>例如，当客户端调用工具时，您的中间件将收到<strong>多次钩子调用</strong>：</p>
<ol>
<li><code>on_message</code> 和 <code>on_request</code> 用于任何初始工具发现操作（如 list_tools）</li>
<li><code>on_message</code>（因为它是任何 MCP 消息）用于工具调用本身</li>
<li><code>on_request</code>（因为工具调用期望响应）用于工具调用本身</li>
<li><code>on_call_tool</code>（因为它是具体的工具执行）用于工具调用本身</li>
</ol>
<p>请注意，MCP SDK 可能会执行额外操作（如为缓存目的列出工具），这将触发超出直接工具执行范围的额外中间件调用。</p>
<p>此层级结构允许您以适当的粒度定位中间件逻辑。对广泛关注点（如日志）使用 <code>on_message</code>，对身份验证使用 <code>on_request</code>，对工具特定逻辑（如性能监控）使用 <code>on_call_tool</code>。</p>
<h4 id="可用钩子">可用钩子</h4>
<ul>
<li><code>on_message</code>: 为所有 MCP 消息（请求和通知）调用</li>
<li><code>on_request</code>: 专为 MCP 请求（期望响应）调用</li>
<li><code>on_notification</code>: 专为 MCP 通知（即发即弃）调用</li>
<li><code>on_call_tool</code>: 在执行工具时调用</li>
<li><code>on_read_resource</code>: 在读取资源时调用</li>
<li><code>on_get_prompt</code>: 在获取提示时调用</li>
<li><code>on_list_tools</code>: 在列出可用工具时调用</li>
<li><code>on_list_resources</code>: 在列出可用资源时调用</li>
<li><code>on_list_resource_templates</code>: 在列出资源模板时调用</li>
<li><code>on_list_prompts</code>: 在列出可用提示时调用</li>
</ul>
<h3 id="中间件中的组件访问">中间件中的组件访问</h3>
<p>理解如何在中间件中访问组件信息（工具、资源、提示）对构建强大的中间件功能至关重要。访问模式在列出操作与执行操作之间存在显著差异。</p>
<h3 id="列出操作-vs-执行操作">列出操作 vs 执行操作</h3>
<p>FastMCP 中间件以不同方式处理两种类型的操作：</p>
<p><strong>列出操作</strong> (<code>on_list_tools</code>, <code>on_list_resources</code>, <code>on_list_prompts</code> 等)：</p>
<ul>
<li>中间件接收<strong>FastMCP 组件对象</strong>，包含完整元数据</li>
<li>这些对象包含 FastMCP 特有属性（如 <code>tags</code>），可直接从组件访问</li>
<li>结果在转换为 MCP 格式前包含完整组件信息</li>
<li>标签包含在返回给 MCP 客户端的组件 <code>meta</code> 字段中</li>
</ul>
<p><strong>执行操作</strong> (<code>on_call_tool</code>, <code>on_read_resource</code>, <code>on_get_prompt</code>)：</p>
<ul>
<li>中间件在<strong>组件执行前</strong>运行</li>
<li>中间件结果为执行结果，或组件未找到时的错误</li>
<li>组件元数据在钩子参数中不可直接访问</li>
</ul>
<h4 id="在执行期间访问组件元数据">在执行期间访问组件元数据</h4>
<p>如果需要在执行操作期间检查组件属性（如标签），请使用通过上下文获取的 FastMCP 服务器实例：</p>
<pre><code class="language-python">from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolError
class 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:
# 工具未找到或其他错误 - 让执行继续
# 并自然处理错误
pass
return await call_next(context)
</code></pre>
<p>相同模式适用于资源和提示：</p>
<pre><code class="language-python">from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ResourceError, PromptError
class 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:
pass
return 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:
pass
return await call_next(context)
</code></pre>
<h4 id="处理列出结果">处理列出结果</h4>
<p>对于列出操作，中间件 <code>call_next</code> 函数在组件转换为 MCP 格式前返回 FastMCP 组件列表。您可以过滤或修改此列表并将其返回给客户端。例如：</p>
<pre><code class="language-python">from fastmcp.server.middleware import Middleware, MiddlewareContext
class 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
</code></pre>
<p>此过滤在组件转换为 MCP 格式并返回给客户端前进行。标签在过滤期间可访问，并包含在最终列出响应的组件 <code>meta</code> 字段中。</p>
<blockquote>
<p>在列出操作中过滤组件时，请确保也在相应的执行钩子（<code>on_call_tool</code>、<code>on_read_resource</code>、<code>on_get_prompt</code>）中阻止已过滤组件的执行，以保持一致性。</p>
</blockquote>
<h4 id="工具调用拒绝">工具调用拒绝</h4>
<p>您可以通过在中间件中抛出 <code>ToolError</code> 来拒绝访问特定工具。这是阻止工具执行的正确方式，因为它与 FastMCP 错误处理系统正确集成</p>
<pre><code class="language-python">from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolError
class 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)
</code></pre>
<blockquote>
<p>拒绝工具调用时，务必抛出 <code>ToolError</code>，而非返回 <code>ToolResult</code> 对象或其他值。<code>ToolError</code> 确保错误通过中间件链正确传播，并转换为正确的 MCP 错误响应格式。</p>
</blockquote>
<h4 id="工具调用修改">工具调用修改</h4>
<p>对于工具调用等执行操作，您可以在执行前修改参数，或在执行后转换结果：</p>
<pre><code class="language-python">from fastmcp.server.middleware import Middleware, MiddlewareContext
class 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
</code></pre>
<p>对于更复杂的工具重写场景，请考虑使用 工具转换 模式，它为创建修改后的工具变体提供了更结构化的方法。</p>
<h3 id="钩子剖析">钩子剖析</h3>
<p>每个中间件钩子遵循相同的模式。让我们通过 <code>on_message</code> 钩子来理解其结构：</p>
<pre><code class="language-python">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
</code></pre>
<p>每个钩子接收两个参数：</p>
<ol>
<li>
<p><strong><code>context: MiddlewareContext</code></strong> - 包含当前请求信息：</p>
<ul>
<li><code>context.method</code> - MCP 方法名称（如 "tools/call"）</li>
<li><code>context.source</code> - 请求来源（"client" 或 "server"）</li>
<li><code>context.type</code> - 消息类型（"request" 或 "notification"）</li>
<li><code>context.message</code> - MCP 消息数据</li>
<li><code>context.timestamp</code> - 请求接收时间</li>
<li><code>context.fastmcp_context</code> - FastMCP Context 对象（如可用）</li>
</ul>
</li>
<li>
<p><strong><code>call_next</code></strong> - 用于继续中间件链的函数。除非您希望完全停止处理，否则<strong>必须</strong>调用此函数。</p>
</li>
</ol>
<p>开发者对请求流拥有完全控制权：</p>
<ul>
<li><strong>继续处理</strong>：调用 <code>await call_next(context)</code> 以继续</li>
<li><strong>修改请求</strong>：在调用 <code>call_next</code> 前更改上下文</li>
<li><strong>修改响应</strong>：在调用 <code>call_next</code> 后更改结果</li>
<li><strong>停止链</strong>：不调用 <code>call_next</code>（极少需要）</li>
<li><strong>处理错误</strong>：在 try/catch 块中包装 <code>call_next</code></li>
</ul>
<p>除了修改请求和响应，您还可以存储状态数据，供工具（可选）稍后访问。为此，请使用 FastMCP Context 适当调用 <code>set_state</code> 或 <code>get_state</code>。</p>
<h3 id="创建中间件">创建中间件</h3>
<p>FastMCP 中间件通过继承 <code>Middleware</code> 基类并重写所需钩子来实现。</p>
<pre><code class="language-python">from fastmcp import FastMCP
from fastmcp.server.middleware import Middleware, MiddlewareContext
class 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())
</code></pre>
<h3 id="向服务器添加中间件">向服务器添加中间件</h3>
<p>中间件按添加到服务器的顺序执行。最先添加的中间件在进入时最先运行，在退出时最后运行：</p>
<pre><code class="language-python">mcp = FastMCP("MyServer")
mcp.add_middleware(AuthenticationMiddleware("secret-token"))
mcp.add_middleware(PerformanceMiddleware())
mcp.add_middleware(LoggingMiddleware())
</code></pre>
<p>这将创建以下执行流：</p>
<ol>
<li>AuthenticationMiddleware（预处理）</li>
<li>PerformanceMiddleware（预处理）</li>
<li>LoggingMiddleware（预处理）</li>
<li>实际工具/资源处理器</li>
<li>LoggingMiddleware（后处理）</li>
<li>PerformanceMiddleware（后处理）</li>
<li>AuthenticationMiddleware（后处理）</li>
</ol>
<h3 id="组合服务器与中间件">组合服务器与中间件</h3>
<p>当使用 服务器组合（下面提的Composition） （如 <code>mount</code> 或 <code>import_server</code>）时，中间件行为遵循以下规则：</p>
<ol>
<li><strong>父服务器中间件</strong>为所有请求运行，包括路由到挂载服务器的请求</li>
<li><strong>挂载服务器中间件</strong>仅为由该特定服务器处理的请求运行</li>
<li><strong>中间件顺序</strong>在每个服务器内保持不变</li>
</ol>
<pre><code class="language-python"># 带有中间件的父服务器
parent = FastMCP("Parent")
parent.add_middleware(AuthenticationMiddleware("token"))
# 带有自身中间件的子服务器
child = FastMCP("Child")
child.add_middleware(LoggingMiddleware())
@child.tool
def child_tool() -> str:
return "from child"
# 挂载子服务器
parent.mount(child, prefix="child")
</code></pre>
<p>当客户端调用 "child_tool" 时，请求将首先流经父服务器的身份验证中间件，然后路由到子服务器，在子服务器中再经过其日志中间件。</p>
<h3 id="内置中间件">内置中间件</h3>
<p>FastMCP 包含多个中间件实现，展示了最佳实践并提供立即可用的功能。让我们通过构建简化版本来探索每种类型的工作原理，然后了解如何使用完整实现。</p>
<h4 id="计时中间件">计时中间件</h4>
<p>性能监控对于理解服务器行为和识别瓶颈至关重要。FastMCP 在 <code>fastmcp.server.middleware.timing</code> 中包含计时中间件。</p>
<p>以下是其工作方式的示例：</p>
<pre><code class="language-python">import time
from fastmcp.server.middleware import Middleware, MiddlewareContext
class 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) * 1000
print(f"请求 {context.method} 在 {duration_ms:.2f}ms 内完成")
return result
except Exception as e:
duration_ms = (time.perf_counter() - start_time) * 1000
print(f"请求 {context.method} 在 {duration_ms:.2f}ms 后失败：{e}")
raise
</code></pre>
<p>要使用具有正确日志和配置的完整版本：</p>
<pre><code class="language-python">from fastmcp.server.middleware.timing import (
TimingMiddleware,
DetailedTimingMiddleware
)
# 对所有请求进行基础计时
mcp.add_middleware(TimingMiddleware())
# 详细的操作级计时（工具、资源、提示）
mcp.add_middleware(DetailedTimingMiddleware())
</code></pre>
<p>内置版本包括自定义日志支持、正确格式化，且 <strong>DetailedTimingMiddleware</strong> 提供 <code>on_call_tool</code> 和 <code>on_read_resource</code> 等操作特定钩子，以实现精细计时。</p>
<h4 id="日志中间件">日志中间件</h4>
<p>请求和响应日志记录对于调试、监控和理解 MCP 服务器中的使用模式至关重要。FastMCP 在 <code>fastmcp.server.middleware.logging</code> 中提供全面的日志中间件。</p>
<p>以下是其工作方式的示例：</p>
<pre><code class="language-python">from fastmcp.server.middleware import Middleware, MiddlewareContext
class 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 result
except Exception as e:
print(f"{context.method} 失败：{e}")
raise
</code></pre>
<p>要使用具有高级功能的完整版本：</p>
<pre><code class="language-python">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))
</code></pre>
<p>内置版本包括负载日志、结构化 JSON 输出、自定义日志支持、负载大小限制以及用于精细控制的操作特定钩子。</p>
<h4 id="速率限制中间件">速率限制中间件</h4>
<p>速率限制对于保护服务器免受滥用、确保公平资源使用以及在负载下保持性能至关重要。FastMCP 在 <code>fastmcp.server.middleware.rate_limiting</code> 中包含复杂的速率限制中间件。</p>
<p>以下是其工作方式的示例：</p>
<pre><code class="language-python">import time
from collections import defaultdict
from fastmcp.server.middleware import Middleware, MiddlewareContext
from mcp import McpError
from mcp.types import ErrorData
class SimpleRateLimitMiddleware(Middleware):
def __init__(self, requests_per_minute: int = 60):
self.requests_per_minute = requests_per_minute
self.client_requests = defaultdict(list)
async def on_request(self, context: MiddlewareContext, call_next):
current_time = time.time()
client_id = "default"# 实际中，从头部或上下文中提取
# 清理旧请求并检查限制
cutoff_time = current_time - 60
self.client_requests = [
req_time for req_time in self.client_requests
if req_time > cutoff_time
]
if len(self.client_requests) >= self.requests_per_minute:
raise McpError(ErrorData(code=-32000, message="超出速率限制"))
self.client_requests.append(current_time)
return await call_next(context)
</code></pre>
<p>要使用具有高级算法的完整版本：</p>
<pre><code class="language-python">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
))
</code></pre>
<p>内置版本包括令牌桶算法、按客户端识别、全局速率限制以及具有可配置客户端识别功能的异步安全实现。</p>
<h4 id="错误处理中间件">错误处理中间件</h4>
<p>一致的错误处理和恢复对于健壮的 MCP 服务器至关重要。FastMCP 在 <code>fastmcp.server.middleware.error_handling</code> 中提供全面的错误处理中间件。</p>
<p>以下是其工作方式的示例：</p>
<pre><code class="language-python">import logging
from fastmcp.server.middleware import Middleware, MiddlewareContext
class 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 = self.error_counts.get(error_key, 0) + 1
self.logger.error(f"{context.method} 中发生错误：{type(error).__name__}: {error}")
raise
</code></pre>
<p>要使用具有高级功能的完整版本：</p>
<pre><code class="language-python">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)
))
</code></pre>
<p>内置版本包括错误转换、自定义回调、可配置的重试逻辑以及正确的 MCP 错误格式化。</p>
<h3 id="组合中间件">组合中间件</h3>
<pre><code class="language-python">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 ErrorHandlingMiddleware
mcp = FastMCP("Production Server")
# 按逻辑顺序添加中间件
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}"
</code></pre>
<h2 id="composition">Composition</h2>
<p>随着MCP 应用规模扩大，你可能希望将工具、资源和提示按逻辑模块组织，或复用现有的服务器组件。FastMCP 通过两种方法支持服务器组合：</p>
<ul>
<li><strong><code>import_server</code></strong>：一次性复制组件并添加前缀（静态组合）。</li>
<li><strong><code>mount</code></strong>：创建实时链接，主服务器在运行时将请求委托给子服务器（动态组合）。</li>
</ul>
<h3 id="为什么要组合服务器">为什么要组合服务器</h3>
<ul>
<li><strong>模块化</strong>：将大型应用拆分为更小、更专注的服务器（例如 <code>WeatherServer</code>、<code>DatabaseServer</code>、<code>CalendarServer</code>）。</li>
<li><strong>可复用性</strong>：创建通用工具服务器（例如 <code>TextProcessingServer</code>），并在需要时挂载。</li>
<li><strong>团队协作</strong>：不同团队可分别开发独立的 FastMCP 服务器，后期再进行组合。</li>
<li><strong>逻辑组织</strong>：将相关功能按逻辑分组，便于管理。</li>
</ul>
<h3 id="导入vs挂载">导入vs挂载</h3>
<p>选择导入还是挂载取决于您的具体用例和需求。</p>
<table>
<thead>
<tr>
<th style="text-align: center"><strong>特性</strong></th>
<th style="text-align: center"><strong>导入</strong></th>
<th style="text-align: center"><strong>挂载</strong></th>
</tr>
</thead>
<tbody>
<tr>
<td style="text-align: center"><strong>方法</strong></td>
<td style="text-align: center"><code>FastMCP.import_server(server, prefix=None)</code></td>
<td style="text-align: center"><code>FastMCP.mount(server, prefix=None)</code></td>
</tr>
<tr>
<td style="text-align: center"><strong>组合类型</strong></td>
<td style="text-align: center">一次性复制（静态）</td>
<td style="text-align: center">实时链接（动态）</td>
</tr>
<tr>
<td style="text-align: center"><strong>更新同步</strong></td>
<td style="text-align: center">子服务器的变更<strong>不会</strong>反映到主服务器</td>
<td style="text-align: center">子服务器的变更<strong>立即</strong>反映到主服务器</td>
</tr>
<tr>
<td style="text-align: center"><strong>性能</strong></td>
<td style="text-align: center">快速 — 无运行时委托开销</td>
<td style="text-align: center">较慢 — 受最慢挂载服务器影响</td>
</tr>
<tr>
<td style="text-align: center"><strong>前缀</strong></td>
<td style="text-align: center">可选 — 省略则保留原名称</td>
<td style="text-align: center">可选 — 省略则保留原名称</td>
</tr>
<tr>
<td style="text-align: center"><strong>适用场景</strong></td>
<td style="text-align: center">打包最终组件、性能敏感场景</td>
<td style="text-align: center">运行时模块化组合</td>
</tr>
</tbody>
</table>
<h3 id="导入">导入</h3>
<p><code>import_server()</code> 方法将一个 <code>FastMCP</code> 实例（<em>子服务器</em>）中的所有组件（工具、资源、模板、提示）复制到另一个实例（<em>主服务器</em>）中。可选提供 <code>prefix</code> 以避免命名冲突。若未提供前缀，组件将按原样导入。当多个服务器使用相同前缀（或无前缀）导入时，最后导入的服务器组件将覆盖先前导入的同名组件。</p>
<pre><code class="language-python">from fastmcp import FastMCP
import asyncio
# 定义子服务器
weather_mcp = FastMCP(name="WeatherService")
@weather_mcp.tool
def get_forecast(city: str) -> dict:
"""获取天气预报。"""
return {"city": city, "forecast": "Sunny"}
@weather_mcp.resource("data://cities/supported")
def list_supported_cities() -> list:
"""列出支持天气查询的城市。"""
return ["London", "Paris", "Tokyo"]
# 定义主服务器
main_mcp = FastMCP(name="MainApp")
# 导入子服务器
async def setup():
await main_mcp.import_server(weather_mcp, prefix="weather")
# 结果：main_mcp 现包含带前缀的组件：
# - 工具: "weather_get_forecast"
# - 资源: "data://weather/cities/supported"
if __name__ == "__main__":
asyncio.run(setup())
main_mcp.run()
</code></pre>
<h3 id="导入的工作原理">导入的工作原理</h3>
<p>当你调用 <code>await main_mcp.import_server(subserver, prefix={whatever})</code> 时：</p>
<ol>
<li><strong>工具</strong>：<code>subserver</code> 的所有工具被添加到 <code>main_mcp</code>，名称前缀为 <code>{prefix}_</code>。
<ul>
<li><code>subserver.tool(name="my_tool")</code> 变为 <code>main_mcp.tool(name="{prefix}_my_tool")</code>。</li>
</ul>
</li>
<li><strong>资源</strong>：所有资源的 URI 和名称均被添加前缀。
<ul>
<li>URI: <code>subserver.resource(uri="data://info")</code> 变为 <code>main_mcp.resource(uri="data://{prefix}/info")</code>。</li>
<li>名称: <code>resource.name</code> 变为 <code>"{prefix}_{resource.name}"</code>。</li>
</ul>
</li>
<li><strong>资源模板</strong>：模板的前缀规则与资源类似。
<ul>
<li>URI: <code>subserver.resource(uri="data://{id}")</code> 变为 <code>main_mcp.resource(uri="data://{prefix}/{id}")</code>。</li>
<li>名称: <code>template.name</code> 变为 <code>"{prefix}_{template.name}"</code>。</li>
</ul>
</li>
<li><strong>提示</strong>：所有提示的名称被添加前缀 <code>{prefix}_</code>。
<ul>
<li><code>subserver.prompt(name="my_prompt")</code> 变为 <code>main_mcp.prompt(name="{prefix}_my_prompt")</code>。</li>
</ul>
</li>
</ol>
<p>请注意，<code>import_server</code> 执行的是<strong>一次性复制</strong>。在导入<strong>之后</strong>对 <code>subserver</code> 所做的更改<strong>不会</strong>反映在 <code>main_mcp</code> 中。<code>subserver</code> 的 <code>lifespan</code> 上下文也<strong>不会</strong>由主服务器执行。</p>
<blockquote>
<p><code>prefix</code> 参数是可选的。如果省略，组件将按原样导入，不进行修改，这样组件将保留其原始名称。当导入多个具有相同前缀或无前缀的服务器时，<strong>最后导入</strong>的服务器的组件将优先。</p>
</blockquote>
<h3 id="挂载">挂载</h3>
<p><code>mount()</code> 方法在 <code>main_mcp</code> 服务器与 <code>subserver</code> 之间创建一个<strong>实时链接</strong>。它不复制组件，而是在运行时将匹配可选 <code>prefix</code> 的组件请求<strong>委托</strong>给 <code>subserver</code> 处理。若未提供前缀，则子服务器的组件可通过原始名称直接访问。当多个服务器使用相同前缀（或无前缀）挂载时，对于冲突的组件名称，最后挂载的服务器将优先。</p>
<pre><code class="language-python">import asyncio
from fastmcp import FastMCP, Client
# 定义子服务器
dynamic_mcp = FastMCP(name="DynamicService")
@dynamic_mcp.tool
def initial_tool():
"""初始工具演示。"""
return "Initial Tool Exists"
# 挂载子服务器（同步操作）
main_mcp = FastMCP(name="MainAppLive")
main_mcp.mount(dynamic_mcp, prefix="dynamic")
# 在挂载后添加工具 — 仍可通过 main_mcp 访问
@dynamic_mcp.tool
def added_later():
"""挂载后添加的工具。"""
return "Tool Added Dynamically!"
# 测试访问已挂载的工具
async def test_dynamic_mount():
tools = await main_mcp.get_tools()
print("可用工具：", list(tools.keys()))
# 输出：['dynamic_initial_tool', 'dynamic_added_later']
async with Client(main_mcp) as client:
result = await client.call_tool("dynamic_added_later")
print("结果：", result.data)
# 输出："Tool Added Dynamically!"
if __name__ == "__main__":
asyncio.run(test_dynamic_mount())
</code></pre>
<h3 id="挂载的工作原理">挂载的工作原理</h3>
<p>配置挂载后：</p>
<ol>
<li><strong>实时链接</strong>：父服务器与挂载的服务器建立连接。</li>
<li><strong>动态更新</strong>：对挂载服务器的更改在通过父服务器访问时立即生效。</li>
<li><strong>前缀访问</strong>：父服务器使用前缀将请求路由到挂载的服务器。</li>
<li><strong>委托</strong>：对匹配前缀的组件的请求在运行时委托给挂载的服务器处理。</li>
</ol>
<p>命名工具、资源、模板和提示的前缀规则与 <code>import_server</code> 相同。这包括为资源和模板的 URI/键及名称添加前缀，以便在多服务器配置中更好地识别。</p>
<p>由于“实时链接”的存在，父服务器上的 <code>list_tools()</code> 等操作会受到最慢挂载服务器速度的影响。特别是，基于 HTTP 的挂载服务器可能引入显著延迟（300-400ms，而本地工具仅需 1-2ms），并且这种减速会影响整个服务器，而不仅仅是与 HTTP 代理工具的交互。如果性能至关重要，通过 <code>import_server()</code> 导入工具可能是更合适的解决方案，因为它在启动时一次性复制组件，而不是在运行时委托请求。</p>
</div>
<div id="MySignature" role="contentinfo">
<p>本文来自博客园，作者：花酒锄作田，转载请注明原文链接：https://www.cnblogs.com/XY-Heruo/p/19102419</p><br><br>
来源：https://www.cnblogs.com/XY-Heruo/p/19102419
頁:
[1]