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": "dev@example.com"}
)

if __name__ == "__main__":
    mcp.run()

2.2 Tools：为LLM提供功能

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

from fastmcp import FastMCP

mcp = 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 * b

if __name__ == "__main__":
    mcp.run()

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

import asyncio
from fastmcp import FastMCP

mcp = FastMCP(name="AsyncServer")

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

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

from fastmcp import FastMCP
from fastmcp.server import Context

mcp = 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 FastMCP

mcp = 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: 用户ID
        
    Returns:
        用户个人资料
    """
    # 在实际应用中，这里会从数据库获取用户信息
    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 Message

mcp = 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 Context

mcp = 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 = complexity
    for 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 FastMCP

mcp = 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 Client

async 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 Client

async 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 StreamableHttpTransport

async 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 FastMCP

mcp = 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)
    
    # 另一种方式：工厂函数
    @classmethod
    def 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 Request

mcp = 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

# 创建要测试的服务器
@pytest.fixture
def mcp_server():
    mcp = FastMCP(name="TestServer")
    
    @mcp.tool()
    def add(a: float, b: float) -> float:
        return a + b
    
    @mcp.resource()
    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):
        # 实际应用中，这里会调用真实的天气API
        await 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 30
        
        for 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_data
        
        ctx.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: int
    title: str
    description: Optional[str] = None
    completed: bool = False

class TaskCreate(BaseModel):
    title: str
    description: 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 1
    new_task = Task(id=task_id, title=task.title, description=task.description)
    tasks_db[task_id] = new_task
    return new_task

@app.put("/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 = True
    return 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 - completed
    
    return {
        "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应用和服务。现在正是开始探索这一强大工具的好时机！