构建强大的工具与资源

理解 MCP 组件

FastMCP 服务器向 LLM 客户端公开三种主要的组件类型：

工具（Tools）：执行操作或计算的函数。
资源（Resources）：客户端可读取的数据源。
提示（Prompts）：用于 LLM 交互的可复用消息模板。

接下来，我们将探讨如何有效地构建每种组件。

创建工具

工具是 MCP 服务器的强大核心。它们使 LLM 能够执行操作、进行计算、调用 API 并与外部系统交互。

基本工具

最简单的工具是使用装饰器修饰的函数：

from fastmcp import FastMCP

mcp = FastMCP("Utility Server")

@mcp.tool
def calculate_area(length: float, width: float) -> float:
    """计算矩形面积。"""
    return length * width

@mcp.tool
def generate_password(length: int = 12, include_symbols: bool = True) -> str:
    """生成随机密码。"""
    import random
    import string

    chars = string.ascii_letters + string.digits
    if include_symbols:
        chars += "!@#$%^&*"

    return ''.join(random.choice(chars) for _ in range(length))

带上下文的高级工具

工具可以访问 MCP 上下文以执行高级操作：

from fastmcp import FastMCP, Context
import aiohttp
import json

mcp = FastMCP("API Tools")

@mcp.tool
async def fetch_weather(city: str, ctx: Context) -> dict:
    """获取指定城市当前天气。"""
    await ctx.info(f"正在获取 {city} 的天气信息...")

    # 使用上下文进行 HTTP 请求
    response = await ctx.http_request(
        "GET", 
        f"https://api.openweathermap.org/data/2.5/weather",
        params={"q": city, "appid": "your_api_key"}
    )

    weather_data = response.json()

    return {
        "city": city,
        "temperature": weather_data["main"]["temp"],
        "description": weather_data["weather"][0]["description"],
        "humidity": weather_data["main"]["humidity"]
    }

@mcp.tool
async def analyze_text(text: str, ctx: Context) -> str:
    """使用客户端的 LLM 分析文本。"""
    prompt = f"""
    请分析以下文本并提供以下方面的见解：
    1. 情感
    2. 关键主题
    3. 写作风格

    文本: {text}
    """

    # 使用客户端的 LLM 进行分析
    response = await ctx.sample(prompt)
    return response.text

具有复杂返回类型的工具

工具可以返回各种数据类型，包括媒体：

from fastmcp import FastMCP
from fastmcp.types import Image, BlobResourceContents
import matplotlib.pyplot as plt
import io
import base64

mcp = FastMCP("Data Visualization")

@mcp.tool
def create_chart(data: list[float], chart_type: str = "line") -> Image:
    """根据数据创建图表。"""
    plt.figure(figsize=(10, 6))

    if chart_type == "line":
        plt.plot(data)
    elif chart_type == "bar":
        plt.bar(range(len(data)), data)

    plt.title(f"{chart_type.title()} Chart")
    plt.xlabel("Index")
    plt.ylabel("Value")

    # 保存为字节流
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png')
    buffer.seek(0)

    # 转换为 base64
    image_base64 = base64.b64encode(buffer.read()).decode()

    return Image(
        data=image_base64,
        mimeType="image/png"
    )

@mcp.tool
def process_data(numbers: list[float]) -> dict:
    """处理数字列表并返回统计信息。"""
    import statistics

    if not numbers:
        return {"error": "未提供数据"}

    return {
        "count": len(numbers),
        "sum": sum(numbers),
        "mean": statistics.mean(numbers),
        "median": statistics.median(numbers),
        "std_dev": statistics.stdev(numbers) if len(numbers) > 1 else 0,
        "min": min(numbers),
        "max": max(numbers)
    }

构建资源

资源公开了 LLM 可以读取并融入其上下文的数据。它们非常适合配置数据、文档或任何只读信息。

静态资源

具有固定 URI 的简单数据源：

@mcp.resource("config://database")
def get_database_config() -> dict:
    """数据库配置设置。"""
    return {
        "host": "localhost",
        "port": 5432,
        "database": "myapp",
        "ssl_enabled": True
    }

@mcp.resource("docs://api-guide")
def get_api_documentation() -> str:
    """API 使用文档。"""
    return """
    # API 使用指南

    ## 认证
    所有请求均需在 Authorization 头部中提供 API 密钥。

    ## 速率限制
    - 基础版：每小时 1000 次请求
    - 高级版：每小时 10000 次请求

    ## 可用端点
    - GET /users - 列出所有用户
    - POST /users - 创建新用户
    - GET /users/{id} - 获取用户详情
    """

动态资源模板

模板允许客户端使用参数请求特定数据：

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """获取用户档案信息。"""
    # 在真实应用中，这里会查询数据库
    users_db = {
        1: {"name": "Alice Johnson", "role": "Admin", "email": "[email protected]"},
        2: {"name": "Bob Smith", "role": "User", "email": "[email protected]"},
        3: {"name": "Carol Davis", "role": "Manager", "email": "[email protected]"}
    }

    user = users_db.get(user_id)
    if not user:
        raise ValueError(f"未找到用户 {user_id}")

    return {
        "id": user_id,
        **user,
        "last_login": "2024-01-15T10:30:00Z",
        "status": "active"
    }

@mcp.resource("data://{dataset}/stats")
def get_dataset_stats(dataset: str) -> dict:
    """获取特定数据集的统计信息。"""
    # 模拟不同的数据集
    datasets = {
        "sales": {"records": 10000, "last_updated": "2024-01-15", "size_mb": 45.2},
        "users": {"records": 5000, "last_updated": "2024-01-14", "size_mb": 12.8},
        "products": {"records": 2500, "last_updated": "2024-01-13", "size_mb": 8.1}
    }

    if dataset not in datasets:
        raise ValueError(f"未找到数据集 '{dataset}'")

    return {
        "dataset": dataset,
        **datasets[dataset],
        "access_level": "public"
    }

基于文件的资源

资源也可以提供文件内容：

import os
from pathlib import Path

@mcp.resource("files://{file_path}")
def read_file_content(file_path: str) -> str:
    """读取项目目录中的文件内容。"""
    # 安全措施：限制在项目目录内
    project_root = Path(__file__).parent
    full_path = project_root / file_path

    # 确保路径在项目目录内
    if not str(full_path.resolve()).startswith(str(project_root.resolve())):
        raise ValueError("访问被拒绝：路径超出项目目录")

    if not full_path.exists():
        raise FileNotFoundError(f"文件未找到: {file_path}")

    return full_path.read_text(encoding='utf-8')

@mcp.resource("logs://{date}")
def get_daily_logs(date: str) -> str:
    """获取特定日期的应用程序日志。"""
    import datetime

    try:
        # 验证日期格式
        datetime.datetime.strptime(date, "%Y-%m-%d")
    except ValueError:
        raise ValueError("日期格式必须为 YYYY-MM-DD")

    log_file = f"logs/{date}.log"

    if not os.path.exists(log_file):
        return f"在 {date} 未找到日志"

    with open(log_file, 'r') as f:
        return f.read()

创建提示

提示为 LLM 交互提供了可复用的模板：

@mcp.prompt
def code_review_prompt(code: str, language: str = "python") -> str:
    """生成用于代码审查的提示。"""
    return f"""
    请审查以下 {language} 代码并提供以下方面的反馈：

    1. 代码质量和可读性
    2. 性能考量
    3. 安全顾虑
    4. 最佳实践遵循情况
    5. 改进建议

    代码:
    ```{language}
    {code}
    ```

    请提供建设性反馈，并给出具体示例。
    """

@mcp.prompt
def data_analysis_prompt(data_description: str, questions: list[str]) -> str:
    """生成用于数据分析的提示。"""
    questions_text = "\\n".join(f"- {q}" for q in questions)

    return f"""
    我需要帮助分析具有以下特征的数据：
    {data_description}

    我希望探索的具体问题：
    {questions_text}

    请提供结构化的分析方法，并建议
    适当的统计方法或可视化方式。
    """

错误处理与验证

健壮的工具应包含适当的错误处理：

@mcp.tool
def divide_numbers(a: float, b: float) -> float:
    """除以两个数字，并进行错误处理。"""
    if b == 0:
        raise ValueError("不能除以零")

    if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
        raise TypeError("两个参数都必须是数字类型")

    return a / b

@mcp.tool
def process_email(email: str) -> dict:
    """验证并处理电子邮件地址。"""
    import re

    # 基础电子邮件验证
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'

    if not re.match(email_pattern, email):
        raise ValueError("电子邮件格式无效")

    username, domain = email.split('@')

    return {
        "email": email,
        "username": username,
        "domain": domain,
        "is_valid": True
    }

测试组件

始终测试你的工具和资源：

import asyncio
from fastmcp import Client

async def test_server():
    async with Client(mcp) as client:
        # 测试工具
        result = await client.call_tool("calculate_area", {"length": 5, "width": 3})
        print(f"面积: {result.text}")

        # 测试资源
        config = await client.read_resource("config://database")
        print(f"配置: {config.content}")

        # 测试资源模板
        user = await client.read_resource("users://1/profile")
        print(f"用户: {user.content}")

if __name__ == "__main__":
    asyncio.run(test_server())

最佳实践

清晰文档：始终包含描述性的文档字符串。
类型提示：使用适当的类型注解以自动生成模式。
错误处理：提供有意义的错误消息。
安全性：适当验证输入并限制文件访问。
性能：考虑将异步操作用于 I/O 密集型任务。
测试：彻底测试所有组件。

在下一节中，我们将探讨如何将 FastMCP 服务器与 Claude Code 集成，以实现无缝的开发工作流程。

FastMCP：构建MCP服务器与客户端的完整指南

打造强大的工具与资源