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支持两种组合方式：

• 静态导入：使用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 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
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 + b
    
    return 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的持续发展，我们可以期待更多强大功能的加入，进一步扩展大语言模型的能力边界。