验证输入参数仅仅是工具执行生命周期的第一步。一旦参数符合Pydantic模型,服务器就会调用注册的处理函数。此函数充当JSON-RPC协议与外部系统之间的桥梁。在模型上下文协议(MCP)的背景下,这座桥梁通常涉及对第三方REST或GraphQL API的异步网络请求。实现这些处理程序需要仔细管理异步上下文、数据转换和网络延迟。与本地文件操作不同,外部API调用会带来不可预测性。服务器必须高效处理这些交互,以便为与大型语言模型(LLM)交互的用户维持良好的响应体验。异步工具处理程序大多数用Python实现的MCP服务器使用asyncio来处理并发请求。定义工具处理程序时,应使用async def语法。这允许服务器在等待外部API响应时让出控制权,从而避免整个服务器在网络请求期间阻塞。使用Python MCP SDK时,工具执行逻辑被封装在一个装饰函数中,用于处理CallToolRequestSchema。在此函数内部,您通常会采用异步HTTP客户端,例如httpx或aiohttp。下图展示了LLM调用封装外部API的工具时的执行流程。digraph G { rankdir=TB; node [fontname="Helvetica,Arial,sans-serif", shape=box, style=filled, color="#dee2e6"]; edge [fontname="Helvetica,Arial,sans-serif"]; subgraph cluster_0 { label = "MCP 服务器范围"; style = filled; color = "#f8f9fa"; Handler [label="工具处理程序\n(异步函数)", fillcolor="#a5d8ff"]; Validation [label="Pydantic\n校验", fillcolor="#b2f2bb"]; Transform [label="数据\n转换", fillcolor="#e599f7"]; } Client [label="MCP 客户端\n(LLM)", fillcolor="#ffc9c9", shape=ellipse]; ExternalAPI [label="外部 REST API\n(例如:天气/股票)", fillcolor="#ffec99", shape=ellipse]; Client -> Validation [label="工具调用请求"]; Validation -> Handler [label="有效参数"]; Handler -> ExternalAPI [label="HTTP GET/POST"]; ExternalAPI -> Handler [label="JSON 响应"]; Handler -> Transform [label="原始数据"]; Transform -> Client [label="格式化文本"]; }工具请求从客户端发出,经过校验和外部执行,最后以文本形式返回给客户端的流程。实现HTTP封装实现的核心在于使用工具调用提供的参数来构建HTTP请求。最佳实践是在服务器生命周期内实例化一个HTTP客户端实例,或者为单个请求使用上下文管理器,以确保连接正确关闭。设想一个场景,我们希望提供一个名为get_crypto_price的工具,用于获取加密货币的当前价格。该工具接受一个ticker符号(例如“BTC”)和一个currency(例如“USD”)。以下是使用httpx的实现方式:import httpx import mcp.types as types from mcp.server import Server app = Server("crypto-server") # 实例化服务器 async def fetch_crypto_price(ticker: str, currency: str) -> str: url = f"https://api.coingecko.com/api/v3/simple/price" params = { "ids": ticker.lower(), "vs_currencies": currency.lower() } async with httpx.AsyncClient() as client: # 设置超时,避免LLM生成停滞 response = await client.get(url, params=params, timeout=10.0) response.raise_for_status() data = response.json() # 提取具体数值 price = data.get(ticker.lower(), {}).get(currency.lower()) if price: return f"The current price of {ticker} is {price} {currency.upper()}." else: return f"Could not retrieve price for {ticker} in {currency}." @app.call_tool() async def handle_tool_call( name: str, arguments: dict ) -> list[types.TextContent]: if name == "get_crypto_price": # 此时参数已由schema校验 ticker = arguments.get("ticker") currency = arguments.get("currency", "usd") result = await fetch_crypto_price(ticker, currency) return [ types.TextContent( type="text", text=result ) ] raise ValueError(f"Unknown tool: {name}")在此实现中,fetch_crypto_price函数封装了外部逻辑。它处理URL构建、查询参数和网络调用。handle_tool_call函数充当路由器,解包字典参数并委托给特定的逻辑函数。数据转换和上下文窗口外部API通常会返回冗长的JSON对象,其中包含元数据、时间戳和大量细节,这些可能与用户的查询无关。大型语言模型具有有限的上下文窗口。将50KB的原始JSON响应直接放入对话历史中会降低模型性能并增加成本(如果按token付费)。工具实现的一个重要职责是将这些原始数据过滤并转换成简洁、易读的格式或最小化的JSON结构。例如,一个天气API可能返回:{ "coord": { "lon": -0.13, "lat": 51.51 }, "weather": [{ "id": 300, "main": "Drizzle", "description": "light intensity drizzle", "icon": "09d" }], "base": "stations", "main": { "temp": 280.32, "pressure": 1012, "humidity": 81, "temp_min": 279.15, "temp_max": 281.15 }, "visibility": 10000, "wind": { "speed": 4.1, "deg": 80 }, "clouds": { "all": 90 }, "dt": 1485789600, "sys": { "type": 1, "id": 5091, "message": 0.0103, "country": "GB", "sunrise": 1485762037, "sunset": 1485794875 }, "id": 2643743, "name": "London", "cod": 200 }传递整个对象效率不高。工具应解析此内容并返回:“伦敦当前天气:小雨,温度:280.32K,湿度:81%。”下图比较了原始API响应与转换后响应的token使用影响。最小化负载大小对于维持长时间对话非常重要。{ "layout": { "title": "Token使用量:原始JSON与转换后输出对比", "xaxis": { "title": "响应类型", "fixedrange": true }, "yaxis": { "title": "预估Token数", "fixedrange": true }, "margin": { "l": 50, "r": 50, "t": 50, "b": 50 }, "height": 400, "width": 600, "bargap": 0.5, "showlegend": false }, "data": [ { "x": ["原始JSON负载", "转换后文本"], "y": [450, 45], "type": "bar", "marker": { "color": ["#ced4da", "#339af0"] } } ] }原始API响应与精选文本响应之间的预估token消耗量比较。安全措施:SSRF和输入净化当工具接受影响网络请求(特指URL或主机名)的输入时,会带来服务器端请求伪造(SSRF)的风险。在SSRF攻击中,攻击者(或出现幻觉的模型)诱导服务器向不应从外部访问的内部资源发出请求。例如,如果您实现了一个工具fetch_url_content(url: str),用户可能会要求LLM获取http://localhost:8080/admin。如果您的MCP服务器运行在开发者的机器上,或者一个带有内部服务的云容器中,该工具可能会泄露敏感的内部配置数据。为减轻此风险,在实现外部调用时:允许列表: 如果可行,将工具限制在一组特定域。如果工具旨在查询维基百科,请确保URL以https://wikipedia.org开头。避免通用代理: 避免构建充当开放代理的工具(例如,“获取[URL]内容”)。而应构建特定工具,例如“获取Jira工单”或“搜索GitHub仓库”。校验协议: 确保请求只使用http或https,阻止诸如file://或gopher://等可能访问本地文件系统的方案。超时管理LLM为用户实时运行。如果一个工具执行需要30秒,用户可能会认为生成失败。Python的httpx库默认为超时,但通常比较宽松。您应该为外部调用明确设置严格的超时。5到10秒的超时通常适用于MCP工具。如果外部API慢于此速度,工具应返回消息说明操作耗时超出预期或服务不可用,而不是无限期保持连接打开。在前面提供的httpx示例中,timeout=10.0确保如果CoinGecko API挂起,工具处理程序会抛出异常(必须捕获),而不是使MCP服务器卡住。# 推荐的超时配置 timeout_config = httpx.Timeout(10.0, connect=5.0) async with httpx.AsyncClient(timeout=timeout_config) as client: # ... 执行请求此配置允许5秒用于建立连接,操作总共10秒,确保服务器能够迅速从网络问题中恢复。