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

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": "[e-mail]"}
)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 + [e-mail]()
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": "[e-mail]", "role": "管理员"},2: {"name": "李四", "email": "[e-mail]", "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支持两种组合方式：

1. 静态导入：使用import_server方法一次性复制组件
2. 动态挂载：使用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)# 正确方式：使用静态方法或类方法@[e-mail]()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": "[e-mail]"},2: {"id": 2, "name": "李四", "email": "[e-mail]"}}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# 创建要测试的服务器
@pytest.fixture
def mcp_server():mcp = FastMCP(name="TestServer")@mcp.tool()def add(a: float, b: float) -> float:return a + [e-mail]()def config() -> dict:return {"version": "1.0.0"}return mcp# 测试工具功能
async def test_add_tool(mcp_server):async with Client(mcp_server) as client:result = await client.tools.call("add", a=5, b=3)assert result == 8# 测试资源功能
async def test_config_resource(mcp_server):async with Client(mcp_server) as client:config = await client.resources.read("config")assert config["version"] == "1.0.0"

6. 实际应用场景

6.1 天气查询服务

下面是一个完整的天气查询服务示例，展示了FastMCP的多种功能：

import asyncio
import httpx
from fastmcp import FastMCP
from fastmcp.server import Context# 创建天气服务器
mcp = FastMCP(name="WeatherService",description="提供全球城市的天气预报和历史天气数据"
)# 模拟天气API客户端
class WeatherAPI:async def get_current_weather(self, city: str):# 实际应用中，这里会调用真实的天气APIawait asyncio.sleep(1)  # 模拟网络延迟# 返回模拟数据weather_data = {"北京": {"temp": 25, "condition": "晴朗", "humidity": 40},"上海": {"temp": 28, "condition": "多云", "humidity": 65},"广州": {"temp": 30, "condition": "雨", "humidity": 80},}return weather_data.get(city, {"error": "城市不存在"})async def get_forecast(self, city: str, days: int = 3):# 模拟天气预报await asyncio.sleep(1.5)if city not in ["北京", "上海", "广州"]:return {"error": "城市不存在"}forecast = []base_temp = 25 if city == "北京" else 28 if city == "上海" else 30for day in range(1, days + 1):forecast.append({"day": day,"temp": base_temp + day - 1,"condition": "晴朗" if day % 3 == 0 else "多云" if day % 3 == 1 else "雨"})return forecast# 创建API客户端实例
weather_api = WeatherAPI()@mcp.tool()
async def current_weather(ctx: Context, city: str) -> dict:"""获取指定城市的当前天气。Args:ctx: FastMCP上下文city: 城市名称（如：北京、上海、广州）Returns:当前天气信息"""ctx.info(f"正在获取{city}的天气信息...")try:weather_data = await weather_api.get_current_weather(city)if "error" in weather_data:ctx.warning(f"无法获取{city}的天气信息")return weather_datactx.info(f"成功获取{city}的天气信息")return {"city": city,"temperature": f"{weather_data['temp']}°C","condition": weather_data['condition'],"humidity": f"{weather_data['humidity']}%"}except Exception as e:ctx.error(f"获取天气信息时出错: {str(e)}")return {"error": f"服务错误: {str(e)}"}@mcp.tool()
async def weather_forecast(ctx: Context, city: str, days: int = 3) -> dict:"""获取指定城市的天气预报。Args:ctx: FastMCP上下文city: 城市名称（如：北京、上海、广州）days: 预报天数（1-7）Returns:天气预报信息"""# 验证参数if days < 1 or days > 7:ctx.warning("预报天数必须在1-7之间")days = max(1, min(days, 7))ctx.info(f"正在获取{city}未来{days}天的天气预报...")try:# 显示进度ctx.progress(0.3, "连接天气服务...")forecast_data = await weather_api.get_forecast(city, days)ctx.progress(0.7, "处理预报数据...")if "error" in forecast_data:ctx.warning(f"无法获取{city}的天气预报")return {"city": city, "error": forecast_data["error"]}# 格式化预报数据formatted_forecast = []for day in forecast_data:formatted_forecast.append({"day": f"第{day['day']}天","temperature": f"{day['temp']}°C","condition": day['condition']})ctx.progress(1.0, "预报数据准备完成")ctx.info(f"成功获取{city}的{days}天天气预报")return {"city": city,"forecast_days": days,"forecast": formatted_forecast}except Exception as e:ctx.error(f"获取天气预报时出错: {str(e)}")return {"error": f"服务错误: {str(e)}"}@mcp.resource()
def supported_cities() -> list:"""返回支持的城市列表。"""return ["北京", "上海", "广州"]@mcp.resource()
def weather_tips() -> dict:"""提供不同天气条件下的建议。"""return {"晴朗": "天气晴朗，适合户外活动，记得防晒。","多云": "天气多云，温度适宜，适合各类活动。","雨": "有雨，外出请携带雨具，注意路滑。"}@mcp.prompt()
def weather_report_template(city: str) -> list:"""创建天气报告提示模板。"""return [{"role": "system", "content": "你是一位专业的天气播报员，提供准确、简洁的天气信息。"},{"role": "user", "content": f"请为{city}提供一份简短的天气报告，包括温度、天气状况和建议。"}]if __name__ == "__main__":mcp.run()

6.2 与FastAPI集成的Web应用

下面是一个将FastMCP与FastAPI集成的任务管理系统示例：

from fastapi import FastAPI, Depends, HTTPException, Path, Query
from fastmcp import FastMCP
from pydantic import BaseModel
from typing import List, Optional
import uvicorn# 创建FastAPI应用
app = FastAPI(title="任务管理API",description="使用FastAPI和FastMCP构建的任务管理系统"
)# 数据模型
class Task(BaseModel):id: inttitle: strdescription: Optional[str] = Nonecompleted: bool = Falseclass TaskCreate(BaseModel):title: strdescription: Optional[str] = None# 模拟数据库
tasks_db = {1: Task(id=1, title="学习FastMCP", description="阅读文档并尝试示例", completed=False),2: Task(id=2, title="构建MCP服务器", description="创建一个简单的工具服务器", completed=True),
}# FastAPI路由
@app.get("/tasks", response_model=List[Task], tags=["任务"])
async def get_tasks(completed: Optional[bool] = Query(None, description="按完成状态筛选")):"""获取所有任务或按完成状态筛选任务"""if completed is None:return list(tasks_db.values())return [task for task in tasks_db.values() if task.completed == completed]@app.get("/tasks/{task_id}", response_model=Task, tags=["任务"])
async def get_task(task_id: int = Path(..., description="任务ID")):"""获取特定任务详情"""if task_id not in tasks_db:raise HTTPException(status_code=404, detail="任务不存在")return tasks_db[task_id]@app.post("/tasks", response_model=Task, status_code=201, tags=["任务"])
async def create_task(task: TaskCreate):"""创建新任务"""task_id = max(tasks_db.keys()) + 1 if tasks_db else 1new_task = Task(id=task_id, title=task.title, description=task.description)tasks_db[task_id] = new_taskreturn [e-mail]("/tasks/{task_id}/complete", response_model=Task, tags=["任务"])
async def complete_task(task_id: int = Path(..., description="任务ID")):"""将任务标记为已完成"""if task_id not in tasks_db:raise HTTPException(status_code=404, detail="任务不存在")tasks_db[task_id].completed = Truereturn tasks_db[task_id]@app.delete("/tasks/{task_id}", status_code=204, tags=["任务"])
async def delete_task(task_id: int = Path(..., description="任务ID")):"""删除任务"""if task_id not in tasks_db:raise HTTPException(status_code=404, detail="任务不存在")del tasks_db[task_id]return None# 创建FastMCP服务器
mcp = FastMCP.from_fastapi(app)# 添加额外的MCP功能
@mcp.tool()
async def task_statistics() -> dict:"""获取任务统计信息。Returns:任务统计数据"""total = len(tasks_db)completed = sum(1 for task in tasks_db.values() if task.completed)pending = total - completedreturn {"total_tasks": total,"completed_tasks": completed,"pending_tasks": pending,"completion_rate": f"{(completed / total * 100) if total > 0 else 0:.1f}%"}@mcp.resource()
def task_categories() -> list:"""获取任务分类建议。"""return ["工作", "学习", "个人", "健康", "家庭", "娱乐", "财务", "其他"]@mcp.prompt()
def task_reminder_template(task_id: int) -> list:"""创建任务提醒提示模板。"""if task_id not in tasks_db:return [{"role": "system", "content": "你是一位任务助手。"},{"role": "user", "content": f"我想获取ID为{task_id}的任务提醒，但该任务不存在。"}]task = tasks_db[task_id]return [{"role": "system", "content": "你是一位友好的任务提醒助手，提供简短、积极的提醒。"},{"role": "user", "content": f"请为以下任务创建一个提醒消息：\n\n标题：{task.title}\n描述：{task.description or '无'}\n状态：{'已完成' if task.completed else '未完成'}"}]# 挂载MCP到FastAPI
app.mount("/mcp", mcp.http_app())# 运行服务器
if __name__ == "__main__":uvicorn.run(app, host="0.0.0.0", port=8000)

7. 总结

FastMCP是一个强大而灵活的Python库，它大大简化了MCP服务器和客户端的创建过程。通过提供高级、Pythonic的接口，FastMCP使开发者能够专注于业务逻辑，而不是底层协议细节。

主要优势包括：

• 简洁的API设计：最小化样板代码，提高开发效率
• 完整的功能集：支持工具、资源、提示和上下文等MCP核心功能
• 灵活的部署选项：支持多种运行方式和与现有框架的集成
• 强大的客户端功能：提供与MCP服务器交互的简单方式
• 丰富的高级模式：支持服务器组合、代理和各种集成方案

无论您是想为大语言模型提供数据访问、功能执行，还是创建复杂的AI应用，FastMCP都是一个值得考虑的工具。它不仅简化了开发过程，还提供了丰富的功能和灵活的集成选项，使您能够充分发挥大语言模型的潜力。

随着MCP协议的不断发展和FastMCP的持续改进，我们可以期待看到更多创新的AI应用和服务。现在正是开始探索这一强大工具的好时机！