Model Context Protocol (MCP) 这个协议简单说就是给大语言模型接入外部数据和工具提供了一套标准化方案。MCP 统一了模型和各种数据源、工具服务之间的交互方式。
FastMCP 是目前用 Python 构建 MCP 服务器最顺手的框架,把底层那些复杂的协议实现全都封装好了,开发者只需要关注业务逻辑就行。
这篇文章会讲清楚 MCP 的基本概念,FastMCP 的工作原理,以及怎么从零开始写一个能跑的 MCP 服务器。
MCP 本质上是个标准协议,让语言模型能安全、一致地和外部系统对话。和传统 API 不太一样的是MCP 是专门为大语言模型设计的。
一个 MCP 服务器主要提供三类能力:
把数据暴露成资源(resources),功能上类似 REST API 的 GET 请求。提供可执行的工具(tools),对标 POST 类的操作接口。定义提示模板(prompts),用来规范模型和数据、用户的交互方式。
如果你有开发经验可以理解为MCP的每一个“能力”其实就是一个 可远程调用的函数(tool)。只要你能和 MCP Server 建立通信(通常通过 WebSocket、Stdio 或 HTTP RPC),你就可以像调用 REST API 一样调用 MCP 的函数。只不过MCP 的「函数」并不是像 Python RPC 那样“自动调用”,它还需要定义 Schema(输入输出结构)+ 元信息(描述),告诉大语言模型 怎么用、什么时候用、怎么传参数。
官方 SDK 当然也能实现 MCP 服务器,但 FastMCP 提供的东西更多。可以说是一个生产级框架,内置了企业身份认证、客户端库、测试工具,还能自动生成 API 文档。
FastMCP 可以很方便地和 Google、GitHub、Azure 这些平台集成,支持部署到云端或者私有化环境。更重要的是开发体验相当友好,几行代码就能跑起来一个完整服务。
先装好 FastMCP,推荐用 uv 这个工具,依赖管理比 pip 效率高:
uv pip install fastmcp
每个 MCP 服务器都从
FastMCP
类开始初始化。这个类管理着工具、资源和提示模板。我们先写个最简单的加法服务器。
新建
server.py
文件:
from fastmcp import FastMCP
mcp = FastMCP("Demo Server 🚀")
@mcp.tool def add(a: int, b: int) -> int:
"""Add two numbers and return the result"""
return a + b
if __name__ == "__main__":
mcp.run()
就这么多代码,一个功能完整的 MCP 服务器就写好了。客户端调用
add
工具时,服务器会执行加法运算并返回结果。
本地运行很简单
fastmcp run server.py
这个命令会启动 MCP 服务器。如果要做 Web 部署,可以用 HTTP 或 SSE 传输:
mcp.run(transport="http", host="127.0.0.1", port=8000, path="/mcp")
服务跑起来以后,客户端就能远程调用工具了。
FastMCP 的工具就是普通 Python 函数加个
@mcp.tool
装饰器。想加多少加多少,比如再来个乘法:
@mcp.tool def multiply(a: float, b: float) ->float: """Multiply two numbers""" returna*b
重启服务器后客户端就能同时用
add
和
multiply
了。
FastMCP 会根据函数签名和 docstring 自动生成 schema,客户端可以直接理解你的 API 结构。
资源在 MCP 里代表只读数据,可以是静态的也可以带参数动态生成。比如暴露个版本号或者用户资料:
@mcp.resource("config://version")
def get_version():
return "1.0.0"
@mcp.resource("user://{user_id}/profile")
def get_profile(user_id: int):
return {"name": f"User {user_id}", "status": "active"}
第一个资源返回固定版本号,第二个根据传入的 user_id 动态返回用户信息。
FastMCP 允许在工具、资源或提示里通过
ctx: Context
参数访问会话上下文。这个上下文提供了日志、LLM 采样、进度追踪、资源访问等能力。
看个实际例子:
from fastmcp import Context
@mcp.tool async def summarize(uri: str, ctx: Context):
await ctx.info(f"Reading resource from {uri}")
data = await ctx.read_resource(uri)
summary = await ctx.sample(f"Summarize this: {data.content[:500]}")
return summary.text
上面代码先记录一条日志然后读取资源内容,最后让客户端的 LLM 生成摘要。有了 Context工具的交互性和智能程度会高很多。
服务器跑起来后,用
fastmcp.Client
就能连接。支持 STDIO、HTTP、SSE 等方式,测试时甚至可以跑内存模式。
from fastmcp import Client
import asyncio
async def main():
async with Client("server.py") as client:
tools = await client.list_tools()
print("Available tools:", tools)
result = await client.call_tool("add", {"a": 5, "b": 7})
print("Result:", result.content[0].text)
asyncio.run(main())
还可以用标准 MCP 配置文件连多个服务器,构建复杂的多服务交互系统。
生产环境肯定要考虑安全问题。FastMCP 内置支持 Google、GitHub、Azure、Auth0、WorkOS 这些企业级认证服务商。启用 OAuth 认证只需要几行配置:
from fastmcp.server.auth.providers.google import GoogleProvider
from fastmcp import FastMCP
auth = GoogleProvider(client_id="...", client_secret="...", base_url="https://myserver.com")
mcp = FastMCP("Secure Server", auth=auth)
这样就只有通过认证的用户能访问服务了。客户端侧用 OAuth 流程连接:
async with Client("https://secure-server.com/mcp", auth="oauth") as client:
result = await client.call_tool("protected_tool")
Token 管理、刷新、错误处理这些 FastMCP 都自动搞定了,我们不用写复杂的代码,只考虑实现我们的工具功能就可以了
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70047311/viewspace-3101443/ ,如需转载,请注明出处,否则将追究法律责任。