前言 所有示例代码我都上传到 Git Repo 了，有需要的话可以直接 clone。https://github.com/rainuxhe/mcp-examples 简介 MCP（全称为Model Context Protocol，模型上下文协议）是一种面向大模型交互过程的通用上下文协议标准。其核心目标在于为模型构建一个结构化、可控、可扩展的语义执行环境，使语言模型能够在统一的上下文管理体系下进行任务调度、工具调用、资源协作与状态保持，从而突破传统Prompt Engineering在多轮交互、指令组合与行为稳定性方面的瓶颈。 在传统的大模型应用中，模型本身只能被动地接收输入、产生输出，要让它调用外部工具或访问自定义的上下文，就需要在代码里逐条写好 API 调用、认证、错误处理的逻辑，既繁琐又难以维护。MCP的初衷，就是将这些"上下文管理"和"工具调用"能力抽象成一个标准化的通信协议，让大模型应用只需关注"我想用什么资源"，由专门的 MCP 服务端来真正执行调用、管理状态、返回结果。 MCP 官方GitHub 有一种说法是，传统的大模型应用叫做Prompt Engineering，而MCP出现后，大模型应用开发应该叫做Context Engineering。传统的提示工程常常依赖于简单的字符串拼接，这种方式有几个问题： 歧义性：模型可能难以区分哪些是指令，哪些是用户输入，哪些是检索到的数据。 提示注入风险：如果提示中包含恶意指令，例如ignore all previous instructions，模型可能被欺骗。 脆弱性：格式的微小变化（比如多一个换行符）都可能导致模型性能下降。 难以维护：当上下文变得更复杂时（例如，多个数据源、工具定义、历史消息），这种拼接方式会变得一团糟（亲身体验，塞了一堆历史消息后，模型的回答越拐越远） 核心概念 Tools（工具） 工具是AI模型可以调用以执行特定操作的函数。它们允许模型与外部系统交互，执行有副作用的操作，如： 调用API获取实时数据 查询或修改数据库 执行代码或脚本 发送邮件或消息 文件操作 工具由模型控制，这意味着AI决定是否以及何时使用它们。工具调用可能会产生副作用，其结果可以反馈到对话中。 Resources（资源） 资源是提供给模型的只读上下文单元（数据源）。它们可以是： 文件内容 数据库记录 API响应 知识库内容 资源由应用程序控制，托管方或开发人员决定公开哪些数据以及如何公开。读取资源没有副作用，类似于仅获取数据的GET请求。资源提供可在需要时注入模型上下文的内容（例如，在问答场景中检索到的文档）。 Prompts（提示模板） 提示模板是可重复使用的提示模板或指令，可以根据需要调用。它们由用户控制或由开发人员预定义。提示可能包含常见任务或指导性工作流程的模板（例如，代码审查模板或问答格式）。 提示模板的关键特性包括： 参数化：支持动态参数输入 资源整合：可嵌入资源上下文供模型参考 多轮交互：支持构建多轮对话流程 统一发现：通过标准接口注册和调用 Sampling（采样） 采样是工具与LLM交互以生成文本的机制。通过采样，工具可以请求LLM生成文本内容，例如生成诗歌、文章或其他文本内容。采样允许工具利用LLM的能力来创建内容，而不仅限于执行预定义的操作。 Elicitation（征询） 征询是一种允许工具向用户请求额外信息或确认的机制。当工具执行过程中需要更多信息才能继续执行时，可以使用征询功能与用户交互。这在处理需要用户确认或提供额外参数的操作时特别有用。 例如，在预订系统中，如果用户请求的日期已满，工具可以征询用户是否愿意选择其他日期。征询机制确保了工具可以在必要时暂停执行，等待用户输入，从而提供更好的用户体验。 征询的关键特性包括： 交互性：允许工具与用户进行双向沟通 验证：可以对用户输入进行验证，确保数据的正确性 可选性：用户可以选择接受、拒绝或取消征询请求 结构化：支持结构化数据输入，便于处理复杂信息 Roots（根） Root表示一次语义执行的起点，携带资源引用、执行目标、响应格式等信息，支持多并发执行流。它作为语义执行的基础输入结构，可以包含多个Prompt和工具，为模型提供完整的上下文环境。 Logging（日志记录） 日志记录是MCP中的一个重要功能，允许服务器和工具向客户端发送日志信息。通过日志记录，开发者可以跟踪工具执行过程、调试问题以及监控系统状态。MCP支持多种日志级别，包括调试(debug)、信息(info)、警告(warning)和错误(error)等。 Notifications（通知） 通知机制允许服务器向客户端发送实时更新信息，例如资源变更、工具列表更新等。通过通知，客户端可以及时了解服务器状态的变化，并相应地更新用户界面或执行其他操作。常见的通知类型包括资源更新通知、工具列表变更通知、提示列表变更通知等。 组件 MCP Server Server 是一个独立的程序或服务，它通过 MCP 协议向 MCP 客户端暴露特定的功能、工具或数据资源。 ...

前言 在实际业务中，根据 tracking_id 追查日志中一条请求的完整处理路径是一个比较常见的需求。不过 FastAPI 官方并没有提供相对应的功能，因此需要开发者自行实现。本文介绍如何基于 contextvars，为每次请求的完整流程都添加一个 tracking_id，并在日志中自动记录。 什么是 contextvars Python 在 3.7 版本的标准库中加入了一个模块 contextvars，顾名思义就是 “(Context Variables) 上下文变量”，通常用来隐式地传递一些环境信息的变量，其作用跟 threading.local() 比较相似。不过 threading.local() 是针对线程的，隔离线程之间的数据状态，而 contextvars 可以用在 asyncio 生态的异步协程中。PS: contextvars 不仅可以用在异步协程中，也可以替代 threading.local() 用在多线程函数中。 基本使用 首先编写 context.py import contextvars from typing import Optional TRACKING_ID: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar( 'tracking_id', default=None ) def get_tracking_id() -> Optional[str]: """用于依赖注入""" return TRACKING_ID.get() 编写中间件 middlewares.py，在请求头和响应头中添加 tracking_id 的信息。常见场景就是客户拿着 tracking_id 找碴。 import uuid from starlette.middleware.base import (BaseHTTPMiddleware, RequestResponseEndpoint) from starlette.requests import Request from starlette.responses import Response from context import TRACKING_ID class TrackingIDMiddleware(BaseHTTPMiddleware): async def dispatch( self, request: Request, call_next: RequestResponseEndpoint ) -> Response: tracking_id = str(uuid.uuid4()) token = TRACKING_ID.set(tracking_id) # HTTP 请求头习惯于使用 latin-1 编码 request.scope["headers"].append((b"x-request-id", tracking_id.encode("latin-1"))) try: resp = await call_next(request) finally: # 无论是否成功，每次请求结束时重置 tracking_id，避免泄露到下一次的请求中 TRACKING_ID.reset(token) # 可选, 在响应中设置跟踪 ID 头 resp.headers["X-Tracking-ID"] = tracking_id return resp 编写 handler 函数 handlers.py，测试在 handler 函数中获取 tracking_id。 import asyncio from context import TRACKING_ID async def mock_db_query(): await asyncio.sleep(1) current_id = TRACKING_ID.get() print(f"This is mock_db_query. Current tracking ID: {current_id}") await asyncio.sleep(1) 编写主函数 main.py import uvicorn from fastapi import Depends, FastAPI from fastapi.responses import PlainTextResponse from starlette.background import BackgroundTasks from context import TRACKING_ID, get_tracking_id from handlers import mock_db_query from middlewares import TrackingIDMiddleware app = FastAPI() app.add_middleware(TrackingIDMiddleware) @app.get("/qwer") async def get_qwer(): """测试上下文变量传递""" current_id = TRACKING_ID.get() print(f"This is get qwer. Current tracking ID: {current_id}") return PlainTextResponse(f"Current tracking ID: {current_id}") @app.get("/asdf") async def get_asdf(tracking_id: str = Depends(get_tracking_id)): """测试依赖注入""" print(f"This is get asdf. tracking ID: {tracking_id}") await mock_db_query() return PlainTextResponse(f"Get request, tracking ID: {tracking_id}") if __name__ == "__main__": uvicorn.run("main:app", host="127.0.0.1", port=8000, workers=4) 启动服务后用 curl 测试 api，在控制台可以看到 tracking_id 在请求中都能捕获到。 This is get qwer. Current tracking ID: 01b0153f-4877-4ca0-ac35-ed88ab406452 INFO: 127.0.0.1:55708 - "GET /qwer HTTP/1.1" 200 OK This is get asdf. tracking ID: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639 This is mock_db_query. Current tracking ID: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639 INFO: 127.0.0.1:55722 - "GET /asdf HTTP/1.1" 200 OK 使用 curl 的控制台输出 ...