MCP
基本介绍
官方地址:
- https://modelcontextprotocol.io/introduction
“MCP 是一种开放协议,旨在标准化应用程序向大型语言模型(LLM)提供上下文的方式。可以把 MCP 想象成 AI 应用程序的 USB-C 接口。就像 USB-C 提供了一种标准化的方式,让你的设备能够连接各种外设和配件一样,MCP 也提供了一种标准化的方式,让 AI 模型能够连接不同的数据源和工具。”
● MCP 主机(MCP Hosts):像 Claude Desktop、IDE 或 AI 工具等程序,它们希望通过 MCP 访问数据。
● MCP 客户端(MCP Clients):维护与服务器 1:1 连接的协议客户端。
● MCP 服务器(MCP Servers):轻量级程序,它们通过标准化的模型上下文协议(Model Context Protocol)公开特定的功能。
● 本地数据源(Local Data Sources):你的计算机上的文件、数据库和服务,MCP 服务器可以安全地访问这些数据。
● 远程服务(Remote Services):通过互联网可用的外部系统(例如 API),MCP 服务器可以与其连接。
https://www.anthropic.com/news/model-context-protocol
MCP
之前已经介绍过了,这里就不做复杂介绍了。
重点要提一下,MCP的 Server 和 Client 有两种交互方式:
● 标准输入输出(stdio)通信:在本地环境中,MCP 客户端通常将服务器程序作为子进程启动,双方通过标准输入(stdin)和标准输出(stdout)进行数据交换,采用 JSON-RPC 2.0 格式的消息进行通信。
● 服务器发送事件(SSE)通信:在远程环境中,MCP 客户端与服务器通过 HTTP 协议进行通信,利用 SSE 实现服务器向客户端的实时数据推送,同样使用 JSON-RPC 2.0 格式的消息进行交互。
平常通过标准的输入输出通信,但是可能有些环境本地是没有的,就算有我们也需要自己实现。
所以如果有别人封装好的 SSE 的 Server,我们就可以直接调用了。
FastAPI
FastAPI 是一个用于构建 API 的现代化、高性能的 Python Web 框架。
https://fastapi.tiangolo.com/
该项目的特性:
● 高性能:得益于 Starlette 和 Pydantic 的支持,FastAPI 的性能可与 NodeJS 和 Go 相媲美,是最快的 Python Web 框架之一。
● 高效编码:通过自动生成交互式 API 文档、数据验证和序列化等功能,开发效率可提高约 200% 至 300%。
● 减少错误:利用标准的 Python 类型提示,减少约 40% 的人为错误,并提供出色的编辑器支持,如自动补全和类型检查,降低调试时间。
● 自动生成文档:FastAPI 自动为您的 API 生成交互式文档,支持 Swagger UI 和 ReDoc,方便测试和使用。
FastAPI-MCP
将 FastAPI + MCP 组合起来,可以快速实现 SSE:
- https://github.com/tadata-org/fastapi_mcp
两这结合后:
● 直接集成:直接将 MCP 服务器挂载到您的 FastAPI 应用中。
● 无需配置:只需指向您的 FastAPI 应用即可运行,无需额外配置。
● 自动发现并转换:自动发现所有 FastAPI 的端点,并将其转换为 MCP 工具。
● 保留模式:保留您的请求模型和响应模型的模式。
● 保留文档:保留所有端点的文档,与 Swagger 中的保持一致。
● 扩展功能:除了自动生成的工具外,还可添加自定义的 MCP 工具。
安装配置
默认你已经配置好了 uv,如果没有用pip也可以。
uv init fastapimcp
cd fastapimcpuv venv
source .venv/bin/activateuv add fastapi-mcp
对应的结果如下所示:
测试服务端
回顾之前
我们上次在 MCP 中写了一个 Client 和 Server,这里进行改写。
放一段上次的 Server 代码,帮助你回忆,这里的代码我们不动:
编写代码
我们修改 main.py,写入如下的内容:
from fastapi import FastAPI
from fastapi_mcp import add_mcp_serverapp = FastAPI()mcp_server = add_mcp_server(app,# 路径mount_path="/now",# 名字name="wzk-icu API MCP",
)# 编写一个获取时间戳的服务
@mcp_server.tool()
async def get_server_time() -> str:"""获取服务器当前时间"""from datetime import datetimereturn datetime.now().isoformat()if __name__ == "__main__":import uvicornuvicorn.run(app, host="127.0.0.1", port=8000)
如下所示:
启动测试
# 当前在 fastapimcp 目录下
python main.py
效果如下:
我们直接访问:
http://127.0.0.1:8000/now
可以看到标准的访问格式: