1. 项目概述为什么MCP服务的异步通信安全如此重要最近在几个生产环境的Python项目中我反复遇到了一个看似基础、实则暗藏玄机的问题基于异步通信的MCP服务在安全加固上总是存在各种疏漏。MCP或者说模型上下文协议在AI工具链和智能体生态中越来越常见它本质上是一个让不同工具、模型和服务器之间进行高效通信的桥梁。很多开发者包括我自己在早期都认为只要功能跑通了加上基础的认证和加密就万事大吉。但现实是从零构建一个能扛住生产环境压力的MCP服务尤其是在异步通信这个核心环节有太多细节被我们习惯性地忽略了。这些细节不是那种“有最好没有也行”的锦上添花而是直接关系到服务稳定性、数据安全性和系统健壮性的基石。比如你有没有想过一个简单的连接超时设置不当可能会导致整个服务线程池被拖死或者你以为用了SSL/TLS就高枕无忧却忽略了证书验证和协议版本的细微差别给中间人攻击留下了可乘之机这些问题在开发测试阶段可能风平浪静一旦上线面对真实的、复杂的网络环境和潜在的恶意请求就会瞬间暴露出来。这篇文章我就结合自己踩过的坑和积累的经验拆解七个在构建生产级Python MCP服务时关于异步通信安全最容易被忽略的加固细节。无论你是正在从零搭建一个新的MCP Server还是在优化一个已有的服务希望这些“血泪教训”能帮你绕开那些隐蔽的陷阱。2. 核心需求解析生产级MCP服务的安全挑战在深入细节之前我们得先搞清楚一个“生产级”的MCP服务到底面临着哪些不同于本地开发或测试环境的安全挑战。这决定了我们加固的侧重点和力度。2.1 网络环境的复杂性与不可预测性开发环境通常是纯净的、可控的局域网。而生产环境则暴露在公网或复杂的内部网络中会面临网络抖动与延迟连接可能随时中断、重连数据包可能乱序、重复或丢失。异步框架如asyncio、aiohttp虽然擅长处理高并发但如果没有正确的超时、重试和连接池管理机制一次网络波动就可能引发雪崩。不可信的客户端客户端可能是来自任何地方的代码可能存在恶意如尝试注入攻击、耗尽服务器资源或存在缺陷如发送畸形数据、不按协议通信。中间人攻击风险数据在传输过程中可能被窃听或篡改。仅仅使用http而不是https是显而易见的错误但即使用了https配置不当同样危险。2.2 资源管理的严峻性一个MCP服务往往需要处理模型推理、数据查询等相对耗时的操作。拒绝服务风险恶意客户端可以通过快速建立大量连接、发送超大负载或故意慢速读取响应来耗尽服务器的连接数、内存或CPU资源。异步任务泄露在异步编程中如果创建的Task没有被正确await或取消它可能会一直留在内存中导致内存泄漏。这在长时间运行的服务中会逐渐累积最终拖垮服务。连接池与限流缺失对下游服务如数据库、其他API的调用如果没有连接池和限流一个上游的流量洪峰就可能击穿下游依赖导致级联故障。2.3 协议与数据层面的隐蔽漏洞MCP协议本身可能定义良好但实现上容易出问题。消息边界与反序列化如何界定一个完整的请求/响应消息使用JSON反序列化时是否对数据大小、深度、类型做了限制一个精心构造的超大或嵌套极深的JSON可能直接导致服务器解析时内存溢出或CPU死循环。输入验证的缺失是否默认信任了客户端传来的所有参数一个路径遍历攻击../../../etc/passwd或命令注入攻击可能就源于对输入参数没有进行严格的净化和验证。错误信息的过度暴露当服务内部出错时返回给客户端的错误信息是否包含了堆栈跟踪、内部文件路径、数据库结构等敏感信息这相当于给攻击者画了一张“攻击路线图”。理解了这些挑战我们接下来的加固措施就有了明确的目标抵御不可信网络、守护有限资源、净化输入输出。3. 细节一连接超时与空闲超时的精细化配置这是最基础也最容易被设为一个随意值比如timeout30的细节。不合理的超时设置要么导致用户体验卡顿要么导致资源无法释放。为什么需要区分多种超时连接超时指从发起连接到成功建立TCP连接所允许的最长时间。主要用于应对网络不通或目标服务挂掉的情况。读超时指从发送请求后等待接收响应数据的最大时间。用于防止慢速客户端或网络延迟导致的线程阻塞。写超时指发送请求数据的最大时间。防止向一个写入缓冲区已满的慢速连接发送数据时被无限期阻塞。空闲超时指连接建立后没有数据交换的最大空闲时间。用于清理那些“僵死”的连接释放资源。在aiohttpServer中的实战配置from aiohttp import web import asyncio async def handle(request): # 你的业务逻辑 await asyncio.sleep(0.1) return web.Response(textOK) app web.Application() # 关键配置底层TCP连接器和超时 timeout aiohttp.ClientTimeout( total60, # 整个请求连接读写的总超时最后防线 connect10, # 连接超时网络不通时快速失败 sock_read30, # 读超时防止慢客户端 sock_connect10, # socket连接超时 ) # 创建自定义连接器并配置空闲超时和连接数限制 connector aiohttp.TCPConnector( limit100, # 全局最大连接数防DoS limit_per_host10, # 对单个目标主机的最大连接数 keepalive_timeout30, # 空闲连接保持时间不宜过长 enable_cleanup_closedTrue, # 清理已关闭的连接避免内存泄漏 force_closeFalse, # 默认不强制关闭使用keep-alive ttl_dns_cache300, # DNS缓存时间 ) # 将超时和连接器配置到app的router中对于客户端请求或通过middleware管理对于服务端 # 服务端更关键的是配置请求处理超时和慢客户端断开 app[client_timeout] timeout app[connector] connector # 使用中间件来实施读超时和慢客户端保护 web.middleware async def timeout_middleware(request, handler): try: # 为每个请求设置处理超时 return await asyncio.wait_for(handler(request), timeoutrequest.app[client_timeout].sock_read) except asyncio.TimeoutError: # 记录日志并返回408超时响应 request.app.logger.warning(fRequest timeout: {request.path}) return web.Response(status408, textRequest Timeout) app.middlewares.append(timeout_middleware) web.run_app(app, host0.0.0.0, port8080)实操心得不要依赖默认值aiohttp或httpx等库的默认超时可能不适合你的生产环境。务必显式声明。总超时是安全网total超时应略大于connect sock_read的和作为最终保障防止某些极端情况。空闲超时与连接池keepalive_timeout设置过短会增加重建连接的开销过长则占用资源。根据你的平均请求间隔来调整通常在30-60秒。监控与调整上线后密切监控超时错误率。如果connect timeout错误多可能是网络或下游服务问题如果read timeout错误多可能是你的处理逻辑太慢或客户端太慢需要优化代码或调整超时值。4. 细节二TLS/SSL配置的“魔鬼细节”使用httpswss是必须的但ssl.create_default_context()的默认配置可能并不安全。常见陷阱弱协议和弱密码套件默认可能为了兼容性启用了如TLSv1.0,TLSv1.1或不安全的密码套件。证书验证不严格默认可能不验证主机名或接受自签名证书在测试环境方便在生产环境是风险。缺乏OCSP装订在线证书状态协议装订能提高TLS握手性能并增强撤销检查但需要额外配置。安全加固的SSL上下文配置import ssl from aiohttp import web import certifi def create_secure_ssl_context(certfile, keyfile): 创建一个生产环境安全的SSL上下文。 certfile: 你的服务器证书链文件路径 keyfile: 你的服务器私钥文件路径 ssl_context ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_context.load_cert_chain(certfilecertfile, keyfilekeyfile) # 1. 禁用不安全的协议版本根据你的客户端兼容性调整 ssl_context.minimum_version ssl.TLSVersion.TLSv1_2 # 强制要求TLSv1.2 # ssl_context.maximum_version ssl.TLSVersion.TLSv1_3 # 可以限制最高版本 # 2. 设置安全的密码套件这是一个推荐的安全集合可能需要根据环境调整 # 优先使用前向保密的密码套件 ssl_context.set_ciphers(ECDHEAESGCM:ECDHECHACHA20:DHEAESGCM:DHECHACHA20) # 3. 强制进行主机名验证对于客户端模式 # ssl_context.check_hostname True # 作为客户端时启用 # ssl_context.verify_mode ssl.CERT_REQUIRED # 作为客户端时启用 # 4. 可选但推荐加载受信任的CA证书用于验证客户端证书双向认证或外部服务 ssl_context.load_verify_locations(cafilecertifi.where()) return ssl_context # 在启动服务器时使用 ssl_context create_secure_ssl_context(server.crt, server.key) web.run_app(app, host0.0.0.0, port8443, ssl_contextssl_context)对于客户端连接如MCP Server连接数据库或其他APIimport aiohttp import ssl import certifi ssl_context ssl.create_default_context(cafilecertifi.where()) ssl_context.check_hostname True ssl_context.verify_mode ssl.CERT_REQUIRED connector aiohttp.TCPConnector(ssl_contextssl_context) async with aiohttp.ClientSession(connectorconnector) as session: async with session.get(https://external-api.com/data) as resp: ...注意事项密码套件选择使用ssl_context.set_ciphers()可以精确控制。建议使用ECDHE密钥交换和AES-GCM加密的套件它们提供前向保密。你可以用openssl ciphers -v命令查看可用的套件并参考Mozilla的SSL配置生成器。证书管理生产环境务必使用由可信CA签发的证书。Let‘s Encrypt是免费且自动化的好选择。定期检查证书过期时间并设置自动续期。双向认证对于内部服务间的高安全要求通信可以考虑配置双向TLS即服务器也验证客户端的证书。这需要在服务器端设置ssl_context.verify_mode ssl.CERT_REQUIRED并加载信任的CA证书来验证客户端证书。协议版本尽早弃用TLSv1.0和TLSv1.1。TLSv1.2是当前最低安全要求TLSv1.3性能和安全更好应优先支持。5. 细节三请求大小限制与流式处理这是防止资源耗尽攻击如超大JSON导致内存溢出的关键防线。你不能假设客户端发送的数据总是合理的。为什么需要限制一个恶意客户端可以发送一个几十GB的“JSON”文件试图一次性撑爆你的服务器内存。即使不是恶意用户误操作上传大文件也可能导致服务不稳定。在aiohttp中实施全局请求大小限制from aiohttp import web import aiohttp app web.Application(client_max_size1024 * 1024 * 10) # 限制请求体最大为10MB # 或者更精细地在路由层面限制 async def handle_large_upload(request): # 这个处理器允许更大的上传比如50MB reader await request.multipart() # ... 使用流式方式处理parts避免一次性加载到内存更安全的做法流式处理与早期拒绝对于可能接收较大数据的端点如文件上传最佳实践是使用流式处理并在读取一定量数据后如果判断内容非法如不是预期的JSON则立即断开连接。from aiohttp import web, hdrs import json import asyncio async def safe_json_handler(request): 一个安全的JSON处理器限制大小并流式读取。 max_size 1024 * 1024 # 1MB reader request.content body bytearray() chunk_size 8192 while True: chunk await reader.read(chunk_size) if not chunk: break body.extend(chunk) if len(body) max_size: # 超过大小限制立即返回错误并断开 raise web.HTTPRequestEntityTooLarge(max_sizemax_size, actual_sizelen(body)) if not body: return web.json_response({}) try: # 使用json.loads的object_hook或自定义解析器来限制深度和防止其他攻击 data json.loads(body.decode(utf-8)) except json.JSONDecodeError: raise web.HTTPBadRequest(textInvalid JSON) # 进一步验证数据结构和内容... return web.json_response({status: processed}) # 使用中间件为所有请求应用大小限制 web.middleware async def size_limit_middleware(request, handler): content_length request.content_length if content_length and content_length 10 * 1024 * 1024: # 10MB raise web.HTTPRequestEntityTooLarge(max_size10*1024*1024, actual_sizecontent_length) return await handler(request) app.middlewares.append(size_limit_middleware)实操心得设置合理的默认值全局client_max_size应该设置一个保守的值如1-10MB对于需要处理大文件的特定路由再单独放宽。利用HTTP协议特性检查Content-Length头部可以在读取body之前就拒绝过大的请求节省服务器资源。aiohttp的HTTPRequestEntityTooLarge异常会自动处理Connection: close。流式处理是王道对于文件上传、大日志接收等场景一定要设计为流式处理request.content.read(chunk_size)边读边处理或写入磁盘避免整个文件驻留内存。JSON解析也要设防Python内置的json.loads()在解析极端复杂的对象时也可能消耗大量CPU和内存。可以考虑使用ijson这样的流式JSON解析库或者为json.loads()设置object_hook和解析深度限制尽管标准库的json模块对深度有默认限制但了解它很重要。6. 细节四速率限制与防抖动策略速率限制用于防止滥用无论是恶意的DoS攻击还是客户端的bug导致的请求风暴。防抖动则用于处理在短时间内因重试、错误恢复等产生的突发请求。分层速率限制策略一个完善的系统可能需要多层限流IP层限流在网关或负载均衡器如Nginx层面实现是最有效和轻量的第一道防线。用户/令牌层限流基于API Key、用户ID等进行限制保证公平使用。端点层限流对某些特别耗资源的端点如模型推理实施更严格的限制。在Python应用层实现令牌桶算法虽然网关层做更好但有时需要在应用层补充。我们可以使用asyncio和内存结构或Redis用于分布式实现。import asyncio import time from collections import defaultdict from aiohttp import web class TokenBucketLimiter: 一个简单的内存令牌桶限流器 def __init__(self, rate, capacity): rate: 每秒补充的令牌数 capacity: 桶容量 self.rate rate self.capacity capacity self.tokens capacity self.last_update time.monotonic() self._lock asyncio.Lock() async def acquire(self, tokens1): async with self._lock: now time.monotonic() # 计算自上次更新以来应补充的令牌 elapsed now - self.last_update refill elapsed * self.rate self.tokens min(self.capacity, self.tokens refill) self.last_update now if self.tokens tokens: self.tokens - tokens return True else: return False # 按客户端IP限流 ip_limiters defaultdict(lambda: TokenBucketLimiter(rate10, capacity20)) # 每秒10个突发20个 web.middleware async def rate_limit_middleware(request, handler): # 获取客户端IP注意如果服务前有代理需要从X-Forwarded-For读取 client_ip request.remote limiter ip_limiters[client_ip] if await limiter.acquire(): return await handler(request) else: # 返回429 Too Many Requests raise web.HTTPTooManyRequests(textRate limit exceeded. Please try again later.) app.middlewares.append(rate_limit_middleware)防抖动策略防抖动通常用于按钮提交、搜索框输入等场景在MCP服务中可能用于处理客户端的重试逻辑。核心是“一段时间内只执行一次”。import asyncio from functools import wraps def debounce(wait): 防抖动装饰器简易版适用于单实例。 wait: 防抖等待时间秒 def decorator(fn): last_call_task None wraps(fn) async def wrapped(*args, **kwargs): nonlocal last_call_task if last_call_task and not last_call_task.done(): last_call_task.cancel() # 取消上一次未完成的调用 # 创建一个新的任务但延迟执行 async def call_it(): await asyncio.sleep(wait) return await fn(*args, **kwargs) last_call_task asyncio.create_task(call_it()) return await last_call_task return wrapped return decorator app.post(/process) debounce(wait1.0) # 1秒内重复调用只执行最后一次 async def process_request(request): data await request.json() # ... 处理逻辑 return web.json_response({result: ok})注意事项内存限流器的局限上述内存限流器在单进程多实例部署时会失效因为每个进程有自己的字典。生产环境需要使用Redis等分布式存储来实现全局限流。限流响应返回429 Too Many Requests状态码并可以携带Retry-After头部告知客户端多久后重试。白名单机制为内部服务、管理接口或可信客户端设置限流白名单。监控与告警记录被限流的请求并设置告警。如果某个IP或用户频繁被限流可能是攻击迹象或客户端有bug。防抖动的取消逻辑上面的简易防抖动装饰器使用了task.cancel()需要确保被装饰的函数能正确处理asyncio.CancelledError。7. 细节五输入验证与输出净化永远不要信任客户端输入。这是安全的第一信条。对于MCP服务输入可能包括URL参数、JSON body、文件上传等。分层验证策略结构验证数据格式是否符合预期是对象、数组、字符串等使用JSON Schema或Pydantic进行声明式验证。类型与范围验证数字是否在合理范围内字符串长度是否有限制枚举值是否合法业务逻辑验证提供的ID在数据库中是否存在用户是否有权限执行此操作净化对用于拼接命令、文件路径、HTML渲染的字符串进行转义或过滤。使用Pydantic进行强类型验证Pydantic能极大地简化输入验证并自动生成清晰的错误信息。from pydantic import BaseModel, Field, validator, HttpUrl from typing import List, Optional import re class ToolInput(BaseModel): name: str Field(..., min_length1, max_length50, regexr^[a-zA-Z0-9_-]$) description: Optional[str] Field(None, max_length500) parameters: List[dict] Field(default_factorylist) endpoint: HttpUrl # Pydantic会自动验证URL格式 validator(parameters) def validate_parameters(cls, v): for param in v: if name not in param or not isinstance(param[name], str): raise ValueError(Each parameter must have a name string field) # 可以添加更多参数规则验证 return v app.post(/register_tool) async def register_tool(request): try: # 1. 解析并验证JSON raw_data await request.json() tool_input ToolInput(**raw_data) # 这里会自动触发验证失败会抛ValidationError # 2. 业务逻辑验证例如名称是否已存在 # if await tool_exists_in_db(tool_input.name): # raise web.HTTPConflict(textfTool {tool_input.name} already exists) # 3. 处理数据此时tool_input是经过验证的Pydantic模型实例 # await save_tool_to_db(tool_input.dict()) return web.json_response({status: success, tool: tool_input.dict()}) except json.JSONDecodeError: raise web.HTTPBadRequest(textInvalid JSON) except ValidationError as e: # 将Pydantic的详细错误信息返回给客户端生产环境可能只返回概括性错误 raise web.HTTPBadRequest(textfValidation error: {e.errors()})输出净化当你的MCP服务需要将数据返回给前端或生成日志、错误消息时要小心信息泄露。错误信息生产环境不应返回详细的堆栈跟踪、内部文件路径、数据库错误信息给客户端。使用全局异常处理中间件来捕获异常并返回通用的错误消息同时将详细错误记录到服务器日志。web.middleware async def error_middleware(request, handler): try: return await handler(request) except web.HTTPException as ex: # 传递已知的HTTP异常 raise except Exception as e: # 记录完整的异常信息到日志系统 request.app.logger.exception(fUnhandled exception: {e}) # 向客户端返回通用错误 raise web.HTTPInternalServerError(textAn internal server error occurred.)日志脱敏确保日志中不会记录密码、API密钥、令牌等敏感信息。在记录请求/响应体之前先过滤掉敏感字段。CORS配置如果MCP服务需要被浏览器端调用正确配置CORS跨源资源共享头至关重要避免未授权的网站访问你的API。使用aiohttp_cors库可以方便管理。8. 细节六连接池管理与资源清理异步编程中连接数据库、Redis、外部API是稀缺资源。不恰当的管理会导致连接泄漏最终耗尽资源。连接池的最佳实践使用库内置的连接池大多数成熟的异步客户端如asyncpg,aioredis,aiohttp.ClientSession都内置了连接池。关键是要正确配置和复用它们。应用生命周期管理在aiohttp应用中通常在on_startup信号中创建全局连接池在on_cleanup信号中关闭。from aiohttp import web import asyncpg import aioredis async def init_db_pool(app): 初始化数据库连接池 # 注意生产环境应从环境变量或配置中心读取连接信息 app[db_pool] await asyncpg.create_pool( hostlocalhost, port5432, useruser, passwordpassword, databasemcp_db, min_size5, # 连接池最小连接数 max_size20, # 连接池最大连接数 max_queries50000, # 连接回收前执行的最大查询数 max_inactive_connection_lifetime300.0, # 空闲连接存活时间秒 ) app.logger.info(Database connection pool initialized.) async def close_db_pool(app): 关闭数据库连接池 await app[db_pool].close() app.logger.info(Database connection pool closed.) async def init_redis(app): 初始化Redis连接 # aioredis 2.x 版本用法 app[redis] await aioredis.from_url( redis://localhost, encodingutf-8, decode_responsesTrue, max_connections50 ) app.logger.info(Redis connection initialized.) async def close_redis(app): 关闭Redis连接 await app[redis].close() app.logger.info(Redis connection closed.) app web.Application() app.on_startup.append(init_db_pool) app.on_startup.append(init_redis) app.on_cleanup.append(close_db_pool) app.on_cleanup.append(close_redis) app.get(/data) async def get_data(request): # 从应用上下文中获取连接池并使用 pool request.app[db_pool] async with pool.acquire() as connection: # 使用connection执行查询 result await connection.fetch(SELECT * FROM tools) return web.json_response({data: [dict(r) for r in result]})异步任务的生命周期管理对于手动创建的asyncio.Task务必确保它们有结束的时候。async def background_cleanup_task(app): 一个后台清理任务示例 try: while True: await asyncio.sleep(3600) # 每小时运行一次 # 执行一些清理逻辑比如清理临时文件、过期缓存等 app.logger.info(Performing background cleanup...) except asyncio.CancelledError: # 优雅地处理取消请求 app.logger.info(Background cleanup task cancelled.) raise async def start_background_tasks(app): app[cleanup_task] asyncio.create_task(background_cleanup_task(app)) async def cleanup_background_tasks(app): app[cleanup_task].cancel() await app[cleanup_task] app.on_startup.append(start_background_tasks) app.on_cleanup.append(cleanup_background_tasks)实操心得连接池参数调优min_size和max_size需要根据实际负载调整。设置太小会影响性能设置太大会浪费资源。监控数据库的活跃连接数来辅助决策。连接泄漏检测定期检查应用持有的连接数是否异常增长。一些客户端库提供了诊断工具。使用async with管理资源确保在async with块内使用连接这样即使发生异常连接也能正确返回池中。避免在全局作用域创建客户端像aiohttp.ClientSession这样的对象如果在全局创建且未关闭会阻止事件循环正常结束。应该遵循“创建-使用-关闭”的模式或将其绑定到应用生命周期。9. 细节七全面的日志、监控与告警安全加固不是一劳永逸的。你需要眼睛日志来观察需要仪表盘监控来度量需要警报告警来叫醒你。结构化日志记录使用structlog或python-json-logger记录结构化的JSON日志便于后续的集中分析和检索。import structlog import logging from aiohttp import web # 配置structlog structlog.configure( processors[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmtiso), structlog.processors.StackInfoRenderer(), structqlog.processors.format_exc_info, structlog.processors.JSONRenderer() # 输出为JSON ], context_classdict, logger_factorystructlog.stdlib.LoggerFactory(), cache_logger_on_first_useTrue, ) logger structlog.get_logger() async def logging_middleware(request, handler): start_time asyncio.get_event_loop().time() log logger.bind( methodrequest.method, pathrequest.path, remoterequest.remote, request_idrequest.headers.get(X-Request-ID, unknown) ) request[log] log try: response await handler(request) processing_time asyncio.get_event_loop().time() - start_time log.info( request_finished, statusresponse.status, processing_timeprocessing_time, content_lengthresponse.content_length ) return response except Exception as e: processing_time asyncio.get_event_loop().time() - start_time log.error( request_failed, exception_typetype(e).__name__, error_messagestr(e), processing_timeprocessing_time ) raise app.middlewares.append(logging_middleware)关键监控指标基础设施层CPU、内存、磁盘、网络使用率。应用层请求率与延迟总QPS各端点P50/P95/P99延迟。错误率4xx和5xx状态码的比例。资源使用数据库连接池使用率、Redis内存使用、外部API调用成功率与延迟。业务指标特定工具调用次数、平均处理时长。安全相关被速率限制的请求数、认证失败次数、异常大的请求体数量。告警策略基于阈值错误率1%持续5分钟P99延迟2秒内存使用率85%。基于变化请求量突然暴跌可能服务挂了或暴增可能被攻击。基于事件检测到登录失败风暴、特定的恶意payload模式。配置示例使用Prometheus客户端from aiohttp import web from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST REQUEST_COUNT Counter(http_requests_total, Total HTTP Requests, [method, endpoint, status]) REQUEST_LATENCY Histogram(http_request_duration_seconds, HTTP request latency, [method, endpoint]) web.middleware async def metrics_middleware(request, handler): endpoint request.path method request.method start_time asyncio.get_event_loop().time() try: response await handler(request) status response.status return response finally: processing_time asyncio.get_event_loop().time() - start_time REQUEST_COUNT.labels(methodmethod, endpointendpoint, statusstatus).inc() REQUEST_LATENCY.labels(methodmethod, endpointendpoint).observe(processing_time) app.route(/metrics) async def metrics(request): return web.Response(bodygenerate_latest(), content_typeCONTENT_TYPE_LATEST)最后的小技巧将你的MCP服务容器化Docker并使用healthcheck端点。这样编排工具如K8s可以自动重启不健康的实例。一个简单的健康检查端点应该检查核心依赖数据库、Redis的连接状态。