FastMCP：为大语言模型构建强大的上下文和工具服务

在人工智能快速发展的今天，大语言模型（LLM）已经成为许多应用的核心。然而，如何让这些模型更好地与外部世界交互，获取实时信息，执行特定任务，一直是开发者面临的挑战。模型上下文协议（Model Context Protocol，简称MCP）应运而生，它为LLM提供了一种标准化的方式来访问外部数据和功能。而FastMCP则是一个高效、简洁的Python库，让开发者能够轻松构建符合MCP规范的服务器和客户端。

本文将深入介绍FastMCP的核心功能、架构设计和实际应用，帮助开发者快速掌握这一强大工具，为大语言模型构建更智能、更实用的应用。

1. FastMCP简介

1.1 什么是MCP协议

模型上下文协议（Model Context Protocol）是一种专为大语言模型设计的标准化协议，它允许LLM以安全、一致的方式与外部系统交互。MCP协议常被描述为"AI的USB-C接口"，提供了一种统一的方式连接LLM与它们可以使用的资源。

MCP协议的核心功能包括：

• 资源（Resources）：类似于GET端点，用于将信息加载到LLM的上下文中
• 工具（Tools）：类似于POST端点，用于执行代码或产生副作用
• 提示（Prompts）：可重用的LLM交互模板
• 上下文（Context）：提供额外的交互功能，如日志记录和进度报告

1.2 FastMCP的优势

FastMCP是MCP协议的高级Python实现，它大大简化了创建MCP服务器和客户端的过程。与直接使用底层MCP SDK相比，FastMCP具有以下优势：

• 高效开发：高级接口意味着更少的代码和更快的开发速度
• 简洁设计：最小化样板代码，专注于业务逻辑
• Python风格：API设计符合Python开发者的习惯
• 完整实现：提供MCP规范的全面实现

1.3 FastMCP 2.0的新特性

FastMCP 2.0是该项目的活跃开发版本，相比于贡献给官方MCP SDK的1.0版本，它增加了许多强大的新功能：

• 强大的客户端功能
• 服务器代理和组合
• OpenAPI/FastAPI集成
• 更多高级特性

2. Servers模块：构建MCP服务器

Servers模块是FastMCP的核心，提供了创建和管理MCP服务器的基础设施。

2.1 FastMCP服务器基础

创建一个基本的FastMCP服务器非常简单：

from fastmcp import FastMCP# 创建一个FastMCP服务器实例
mcp = FastMCP(name="MyServer")# 启动服务器
if __name__ == "__main__":mcp.run()

您还可以添加更多元数据来描述您的服务器：

from fastmcp import FastMCP# 创建带有元数据的服务器
mcp = FastMCP(name="WeatherServer",description="提供天气预报和历史天气数据",version="1.0.0",contact={"name": "开发团队", "email": "dev@example.com"}
)if __name__ == "__main__":mcp.run()

2.2 Tools：为LLM提供功能

Tools（工具）是FastMCP中最常用的功能之一，它允许LLM执行特定的操作或函数。使用@mcp.tool()装饰器可以轻松将Python函数转换为LLM可调用的工具：

from fastmcp import FastMCPmcp = FastMCP(name="CalculatorServer")@mcp.tool()
def add(a: float, b: float) -> float:"""将两个数字相加。Args:a: 第一个数字b: 第二个数字Returns:两个数字的和"""return a + b@mcp.tool()
def multiply(a: float, b: float) -> float:"""将两个数字相乘。Args:a: 第一个数字b: 第二个数字Returns:两个数字的乘积"""return a * bif __name__ == "__main__":mcp.run()

FastMCP支持异步工具函数，这对于需要I/O操作的任务特别有用：

import asyncio
from fastmcp import FastMCPmcp = FastMCP(name="AsyncServer")@mcp.tool()
async def fetch_data(url: str) -> dict:"""从指定URL异步获取数据。Args:url: 要获取数据的URLReturns:获取的数据字典"""# 模拟异步网络请求await asyncio.sleep(1)return {"url": url, "status": "success", "data": "示例数据"}

工具函数还可以使用上下文对象来访问额外功能：

from fastmcp import FastMCP
from fastmcp.server import Contextmcp = FastMCP(name="ContextServer")@mcp.tool()
async def log_message(ctx: Context, message: str) -> bool:"""记录消息并返回成功状态。Args:ctx: FastMCP上下文对象message: 要记录的消息Returns:操作是否成功"""# 使用上下文发送日志ctx.info(f"收到消息: {message}")return True

2.3 Resources：提供上下文数据

Resources（资源）用于向LLM提供数据和上下文信息。与工具不同，资源主要用于读取数据而非执行操作：

from fastmcp import FastMCPmcp = FastMCP(name="DocumentServer")@mcp.resource()
def company_info() -> str:"""提供公司基本信息。"""return """公司名称: 示例科技有限公司成立时间: 2020年主营业务: 人工智能解决方案员工人数: 100+"""@mcp.resource()
def pricing_policy() -> dict:"""返回产品定价策略。"""return {"basic_plan": {"price": 99, "features": ["基础功能", "电子邮件支持"]},"pro_plan": {"price": 199, "features": ["所有基础功能", "高级功能", "优先支持"]},"enterprise": {"price": "联系销售", "features": ["定制解决方案", "专属支持团队"]}}

资源也可以接受参数，这使得它们更加灵活：

@mcp.resource()
def user_profile(user_id: int) -> dict:"""获取指定用户的个人资料。Args:user_id: 用户IDReturns:用户个人资料"""# 在实际应用中，这里会从数据库获取用户信息users = {1: {"name": "张三", "email": "zhang@example.com", "role": "管理员"},2: {"name": "李四", "email": "li@example.com", "role": "用户"},}if user_id in users:return users[user_id]else:return {"error": "用户不存在"}

2.4 Prompts：创建提示模板

Prompts（提示）允许您创建可重用的提示模板，这些模板可以被参数化并用于标准化LLM交互：

from fastmcp import FastMCP
from fastmcp.types import Messagemcp = FastMCP(name="CustomerServiceBot")@mcp.prompt()
def greeting(customer_name: str) -> list[Message]:"""创建一个个性化的问候提示。Args:customer_name: 客户名称Returns:包含问候消息的提示"""return [{"role": "system", "content": "你是一位专业的客户服务代表，语气友好且乐于助人。"},{"role": "user", "content": f"你好，我是{customer_name}，我需要一些帮助。"}]@mcp.prompt()
def product_inquiry(product_name: str, specific_question: str = None) -> list[Message]:"""创建产品咨询提示。Args:product_name: 产品名称specific_question: 具体问题（可选）Returns:包含产品咨询的提示"""content = f"我想了解关于{product_name}的信息。"if specific_question:content += f" 具体来说，{specific_question}"return [{"role": "system", "content": "你是产品专家，提供准确、简洁的产品信息。"},{"role": "user", "content": content}]

2.5 Context：访问服务器功能

Context（上下文）对象提供了在工具和资源函数内部访问FastMCP服务器功能的方法：

from fastmcp import FastMCP
from fastmcp.server import Contextmcp = FastMCP(name="ContextDemoServer")@mcp.tool()
async def process_task(ctx: Context, task_name: str, complexity: int) -> dict:"""处理一个任务并报告进度。Args:ctx: FastMCP上下文对象task_name: 任务名称complexity: 任务复杂度(1-10)Returns:任务处理结果"""# 发送信息日志ctx.info(f"开始处理任务: {task_name}")# 报告进度total_steps = complexityfor step in range(1, total_steps + 1):# 更新进度ctx.progress(step / total_steps, f"完成步骤 {step}/{total_steps}")# 如果任务复杂，发送详细日志if complexity > 5:ctx.debug(f"步骤 {step} 详细信息: 正在处理中...")# 任务完成，发送成功日志ctx.info(f"任务 {task_name} 已完成")return {"task": task_name,"status": "completed","complexity_level": complexity}

上下文对象提供的主要功能包括：

• 日志记录（debug、info、warning、error）
• 进度报告
• 资源访问
• 请求元数据访问

2.6 Proxy Servers：代理远程服务器

Proxy Servers（代理服务器）允许一个FastMCP服务器实例作为另一个MCP服务器的前端：

from fastmcp import FastMCP, Client# 假设我们有一个远程MCP服务器运行在这个URL
remote_url = "https://api.example.com/mcp"# 创建一个客户端连接到远程服务器
client = Client(remote_url)# 从客户端创建代理服务器
proxy_server = FastMCP.from_client(client=client,name="LocalProxy",description="本地代理到远程MCP服务"
)if __name__ == "__main__":# 在本地运行代理服务器proxy_server.run()

代理服务器的主要用途包括：

• 为远程MCP服务器提供不同的传输方式
• 集成不同实现的MCP服务器
• 创建服务器网关

2.7 Composition：组合多个服务器

Composition（组合）功能允许您将多个FastMCP服务器组合成一个更大的应用：

from fastmcp import FastMCP# 创建主服务器
main_server = FastMCP(name="MainApplication")# 创建工具服务器
tools_server = FastMCP(name="ToolsServer")@tools_server.tool()
def utility_function(text: str) -> str:return f"处理结果: {text.upper()}"# 创建数据服务器
data_server = FastMCP(name="DataServer")@data_server.resource()
def get_config() -> dict:return {"api_version": "v2", "timeout": 30}# 将工具服务器和数据服务器导入到主服务器
main_server.import_server(tools_server, prefix="tools")
main_server.import_server(data_server, prefix="data")if __name__ == "__main__":main_server.run()

FastMCP支持两种组合方式：

• 静态导入：使用import_server方法一次性复制组件
• 动态挂载：使用mount方法创建实时委托链接

# 动态挂载服务器
main_server.mount(auth_server, prefix="auth")
main_server.mount(content_server, prefix="content")

3. Deployment模块：部署MCP服务器

Deployment模块提供了多种方式来运行和部署FastMCP服务器。

3.1 Running the Server：运行服务器

最基本的运行方式是使用run()方法：

from fastmcp import FastMCPmcp = FastMCP(name="SimpleServer")@mcp.tool()
def hello(name: str) -> str:return f"你好，{name}！"if __name__ == "__main__":# 使用默认设置运行服务器mcp.run()

您可以配置多种运行选项：

if __name__ == "__main__":# 使用HTTP传输并指定端口mcp.run(transport="http",host="0.0.0.0",port=8080,log_level="info")

FastMCP支持多种传输协议：

• HTTP：标准HTTP传输
• SSE：服务器发送事件传输
• STDIO：标准输入/输出传输（用于命令行工具）

3.2 ASGI Integration：与ASGI框架集成

FastMCP可以轻松集成到现有的ASGI应用中，如Starlette和FastAPI：

from fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.responses import JSONResponse# 创建FastMCP服务器
mcp = FastMCP(name="APIServer")@mcp.tool()
def greet(name: str) -> str:return f"你好，{name}！"# 创建Starlette路由
async def homepage(request):return JSONResponse({"message": "欢迎访问API服务器"})# 获取MCP的ASGI应用
mcp_app = mcp.http_app()# 创建Starlette应用并挂载MCP
app = Starlette(routes=[Route("/", homepage),Mount("/mcp", app=mcp_app)
])# 使用uvicorn运行
# uvicorn myapp:app

与FastAPI集成：

from fastmcp import FastMCP
from fastapi import FastAPI# 创建FastMCP服务器
mcp = FastMCP(name="FastAPIIntegration")@mcp.tool()
def analyze_text(text: str) -> dict:word_count = len(text.split())char_count = len(text)return {"word_count": word_count,"character_count": char_count,"summary": f"文本包含{word_count}个单词，{char_count}个字符"}# 创建FastAPI应用
app = FastAPI(title="集成示例")@app.get("/")
def read_root():return {"message": "欢迎访问API"}# 挂载MCP到FastAPI
app.mount("/mcp", mcp.http_app())

ASGI集成的主要用途包括：

• 向现有网站或API添加MCP功能
• 在特定URL路径下挂载MCP服务器
• 结合多种服务在单一应用中

3.3 Authentication：服务器认证

FastMCP利用底层MCP SDK的OAuth 2.0支持提供认证功能：

from fastmcp import FastMCP# 创建带有认证的服务器
mcp = FastMCP(name="SecureServer",# 配置认证（这里使用简化示例，实际使用时需参考MCP认证文档）auth={"type": "oauth2","flows": {"clientCredentials": {"tokenUrl": "https://auth.example.com/token","scopes": {"read": "读取权限","write": "写入权限"}}}}
)@mcp.tool()
def secure_operation(data: str) -> str:return f"安全处理: {data}"

3.4 CLI：命令行界面

FastMCP提供了一个命令行界面，使您可以轻松运行和管理服务器：

# 直接运行服务器
fastmcp run server.py# 以开发模式运行（带MCP检查器）
fastmcp dev server.py# 安装到Claude桌面应用
fastmcp install server.py# 查看版本信息
fastmcp version

CLI还支持依赖管理：

# 使用指定依赖运行
fastmcp run server.py --with pandas,numpy# 使用可编辑依赖运行
fastmcp run server.py --with-editable ./my_package

4. Clients模块：与MCP服务器交互

Clients模块提供了与MCP服务器交互的客户端功能。

4.1 Client Overview：客户端概述

基本的客户端使用方式：

import asyncio
from fastmcp import Clientasync def main():# 连接到MCP服务器async with Client("http://localhost:8080") as client:# 调用工具result = await client.tools.call("add", a=5, b=3)print(f"计算结果: {result}")# 读取资源weather = await client.resources.read("current_weather", city="北京")print(f"天气信息: {weather}")if __name__ == "__main__":asyncio.run(main())

客户端还支持回调处理：

import asyncio
from fastmcp import Clientasync def progress_callback(progress: float, message: str):print(f"进度: {progress*100:.1f}% - {message}")async def main():async with Client("http://localhost:8080") as client:# 注册回调client.register_progress_callback(progress_callback)# 调用可能产生进度更新的工具result = await client.tools.call("process_large_file", filename="data.csv")print(f"处理结果: {result}")

4.2 Transports：客户端传输

FastMCP客户端支持多种传输方式：

import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransportasync def main():# 创建HTTP传输transport = StreamableHttpTransport(base_url="http://localhost:8080")# 使用指定传输创建客户端async with Client(transport=transport) as client:result = await client.tools.call("hello", name="世界")print(result)

对于测试，可以使用内存传输：

import asyncio
from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport# 创建服务器
mcp = FastMCP(name="TestServer")@mcp.tool()
def echo(message: str) -> str:return f"回声: {message}"async def main():# 创建内存传输transport = FastMCPTransport(mcp=mcp)# 使用内存传输创建客户端async with Client(transport=transport) as client:result = await client.tools.call("echo", message="测试消息")print(result)  # 输出: 回声: 测试消息

FastMCP支持的主要传输类型包括：

• StreamableHttpTransport：用于HTTP连接（推荐用于远程服务器）
• SSETransport：用于SSE连接（旧版选项）
• FastMCPTransport：用于内存中测试
• UvxStdioTransport/NpxStdioTransport：用于本地服务器

5. Patterns模块：高级使用模式

Patterns模块提供了一系列高级使用模式和集成方案。

5.1 Decorating Methods：装饰方法模式

在类中使用FastMCP装饰器需要特别注意：

from fastmcp import FastMCPmcp = FastMCP(name="ClassMethodsDemo")class Calculator:def __init__(self, precision: int = 2):self.precision = precision# 不要直接装饰实例方法def add_wrong(self, a: float, b: float) -> float:"""这种方式会导致错误，因为装饰器在实例存在前捕获方法"""return round(a + b, self.precision)# 正确方式：使用静态方法或类方法@staticmethod@mcp.tool()def add(a: float, b: float, precision: int = 2) -> float:"""正确方式：使用静态方法并将实例属性作为参数传递"""return round(a + b, precision)# 另一种方式：工厂函数@classmethoddef register_methods(cls, server: FastMCP):"""使用工厂方法注册实例方法"""calculator = cls()@server.tool()def multiply(a: float, b: float) -> float:return round(a * b, calculator.precision)return calculator# 注册方法
calc = Calculator.register_methods(mcp)

5.2 HTTP Requests：访问HTTP请求信息

在FastMCP服务器中访问HTTP请求信息：

from fastmcp import FastMCP
from fastmcp.server.dependencies import get_http_request
from starlette.requests import Requestmcp = FastMCP(name="HTTPRequestDemo")@mcp.tool()
async def user_agent_info(request: Request = get_http_request()) -> dict:"""返回有关客户端的信息。Args:request: HTTP请求对象（自动注入）Returns:客户端信息"""return {"user_agent": request.headers.get("user-agent", "未知"),"client_ip": request.client.host if request.client else "未知","request_path": request.url.path,"query_params": dict(request.query_params)}

5.3 OpenAPI：从OpenAPI生成服务器

FastMCP可以从OpenAPI规范自动生成MCP服务器：

import httpx
from fastmcp import FastMCP# 创建API客户端
api_client = httpx.AsyncClient(base_url="https://api.example.com")# 加载OpenAPI规范
openapi_spec = {"openapi": "3.0.0","info": {"title": "示例API", "version": "1.0.0"},"paths": {"/users": {"get": {"summary": "获取用户列表","operationId": "getUsers","responses": {"200": {"description": "成功响应","content": {"application/json": {}}}}}},"/users/{userId}": {"get": {"summary": "获取用户详情","operationId": "getUserById","parameters": [{"name": "userId","in": "path","required": True,"schema": {"type": "integer"}}],"responses": {"200": {"description": "成功响应","content": {"application/json": {}}}}}}}
}# 从OpenAPI规范创建MCP服务器
mcp = FastMCP.from_openapi(openapi_spec=openapi_spec,client=api_client
)

5.4 FastAPI：从FastAPI应用生成服务器

FastMCP可以从FastAPI应用自动生成MCP服务器：

from fastapi import FastAPI, Path
from fastmcp import FastMCP# 创建FastAPI应用
app = FastAPI(title="用户API")@app.get("/users")
async def get_users():"""获取所有用户列表"""return [{"id": 1, "name": "张三"},{"id": 2, "name": "李四"}]@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., description="用户ID")):"""获取特定用户信息"""users = {1: {"id": 1, "name": "张三", "email": "zhang@example.com"},2: {"id": 2, "name": "李四", "email": "li@example.com"}}if user_id in users:return users[user_id]return {"error": "用户不存在"}# 从FastAPI应用创建MCP服务器
mcp = FastMCP.from_fastapi(app)

5.5 Contrib Modules：贡献模块

FastMCP包含一个contrib包，其中包含社区贡献的扩展模块：

from fastmcp import FastMCP
from fastmcp.contrib import my_module  # 假设存在这样的贡献模块mcp = FastMCP(name="ContribDemo")# 使用贡献模块的功能
my_module.setup(mcp)

5.6 Testing：测试MCP服务器

FastMCP提供了测试MCP服务器的工具和模式：

import pytest
from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport# 创建要测试的服务器
@pytest.fixture
def mcp_server():mcp = FastMCP(name="TestServer")@mcp.tool()def add(a: float, b: float) -> float:return a + breturn mcp# 创建客户端测试夹具
@pytest.fixture
async def client(mcp_server):transport = FastMCPTransport(mcp=mcp_server)async with Client(transport=transport) as client:yield client# 编写测试
async def test_add_tool(client):result = await client.tools.call("add", a=2, b=3)assert result == 5

总结

FastMCP是一个强大而灵活的Python库，为开发者提供了构建符合MCP规范的服务器和客户端的简便方法。通过其简洁的API和丰富的功能，FastMCP使得为大语言模型创建上下文和工具服务变得前所未有的简单。

无论您是想为LLM提供数据访问能力，还是想让它们能够执行特定操作，FastMCP都能满足您的需求。随着AI应用的不断发展，FastMCP将成为连接大语言模型与外部世界的重要桥梁。

通过本文的介绍，希望您已经对FastMCP有了全面的了解，并能够开始使用它来构建自己的MCP服务。随着FastMCP 2.0的持续发展，我们可以期待更多强大功能的加入，进一步扩展大语言模型的能力边界。