定义MCP的工具（Tools）

工具允许LLM通过您的服务器执行操作。与资源不同，工具需要执行计算并具有副作用。

MCP工具的定义规范：

必须要定义工具服务的，文档注释（告诉大模型或Agent，该工具的作用是什么）
工具必须定义参数及声明的参数类型（告诉大模型或Agent，该工具传什么样的参数）
对于工具返回的数据，类型和数据结构要声明（告诉大模型或Agent，该工具执行完之后的结果是什么样的，方便后面去进行解析和使用）

示例：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="Tool Example")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算数字相加"""
    return a + b

定义MCP的资源（Resources）

资源是向LLM公开数据的方式。它们类似于REST API中的GET端点 - 它们提供数据，但不应执行大量计算或产生副作用。

示例

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="Resource Example")

@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
    """按名称读取文档。"""
    # 这通常会从磁盘读取
    return f"{name}的内容"

@mcp.resource("config://settings")
def get_settings() -> str:
    """获取应用程序设置。"""
    return """{
  "theme": "dark",
  "language": "en",
  "debug": false
}"""

工具结构化输出

默认情况下，工具将返回结构化结果，如果其返回类型注释是兼容的。否则，它们将返回非结构化结果。

结构化输出支持以下返回类型：

Pydantic模型（BaseModel子类）
类型字典
数据类和其他具有类型提示的类
dict[str, T]（其中T是任何JSON可序列化类型）
原始类型（str、int、float、bool、bytes、None）- 包装在{“result”: value}
泛型类型（列表、元组、联合、可选等）- 包装在{“result”: value}

from typing import TypedDict
from pydantic import BaseModel, Field

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Structured Output Example")

# 使用Pydantic模型获得丰富的结构化数据
class WeatherData(BaseModel):
    """天气信息结构。"""

    temperature: float = Field(description="摄氏温度")
    humidity: float = Field(description="湿度百分比")
    condition: str
    wind_speed: float

@mcp.tool()
def get_weather(city: str) -> WeatherData:
    """获取城市天气 - 返回结构化数据。"""
    # 模拟天气数据
    return WeatherData(
        temperature=72.5,
        humidity=45.0,
        condition="sunny",
        wind_speed=5.2,
    )

# 使用TypedDict获得更简单的结构
class LocationInfo(TypedDict):
    latitude: float
    longitude: float
    name: str

@mcp.tool()
def get_location(address: str) -> LocationInfo:
    """获取位置坐标"""
    return LocationInfo(latitude=51.5074, longitude=-0.1278, name="英国伦敦")

# 使用dict[str, Any]获得灵活的模式
@mcp.tool()
def get_statistics(data_type: str) -> dict[str, float]:
    """获取各种统计数据"""
    return {"mean": 42.5, "median": 40.0, "std_dev": 5.2}

# 具有类型提示的普通类也可以用于结构化输出
class UserProfile:
    name: str
    age: int
    email: str | None = None

    def __init__(self, name: str, age: int, email: str | None = None):
        self.name = name
        self.age = age
        self.email = email

@mcp.tool()
def get_user(user_id: str) -> UserProfile:
    """获取用户资料 - 返回结构化数据"""
    return UserProfile(name="Alice", age=30, email="alice@example.com")

# 没有类型提示的类无法用于结构化输出
class UntypedConfig:
    def __init__(self, setting1, setting2):
        self.setting1 = setting1
        self.setting2 = setting2

@mcp.tool()
def get_config() -> UntypedConfig:
    """这返回非结构化输出 - 没有生成模式"""
    return UntypedConfig("value1", "value2")

# 列表和其他类型会自动包装
@mcp.tool()
def list_cities() -> list[str]:
    """获取城市列表"""
    return ["伦敦", "巴黎", "东京"]
    # 返回: {"result": ["伦敦", "巴黎", "东京"]}

@mcp.tool()
def get_temperature(city: str) -> float:
    """获取温度作为简单浮点数"""
    return 22.5
    # 返回: {"result": 22.5}

定义MCP的提示词（Prompts）

提示是可重用的模板，可帮助LLM有效地与您的服务器交互。

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

mcp = FastMCP(name="Prompt Example")

@mcp.prompt(title="代码审查")
def review_code(code: str) -> str:
    return f"请审查此代码:\n\n{code}"

@mcp.prompt(title="调试助手")
def debug_error(error: str) -> list[base.Message]:
    return [
        base.UserMessage("我看到了这个错误:"),
        base.UserMessage(error),
        base.AssistantMessage("我会帮你调试。你试过什么方法？"),
    ]

定义MCP的上下文（Context）

Context对象使您的工具和资源能够访问MCP功能。

from mcp.server.fastmcp import Context, FastMCP

mcp = FastMCP(name="Progress Example")

@mcp.tool()
async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:
    """执行带进度更新的任务。"""
    await ctx.info(f"开始: {task_name}")

    for i in range(steps):
        progress = (i + 1) / steps
        await ctx.report_progress(
            progress=progress,
            total=1.0,
            message=f"步骤 {i + 1}/{steps}",
        )
        await ctx.debug(f"完成步骤 {i + 1}")

    return f"任务 '{task_name}' 完成"

MCP 添加认证

认证架构：

授权服务器（AS）：处理OAuth流、用户身份验证和令牌颁发
资源服务器（RS）：验证令牌并为受保护资源提供服务的MCP服务器
客户端：获取令牌，提交并将其与MCP服务器进行验证

MCP服务器可以通过提供协议的TokenVerifier实现来使用身份验证。

from pydantic import AnyHttpUrl

from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp import FastMCP

class SimpleTokenVerifier(TokenVerifier):
    """用于演示的简单令牌验证器。"""

    async def verify_token(self, token: str) -> AccessToken | None:
        pass  # 这里您需要实现实际的令牌验证

# 创建FastMCP实例作为资源服务器
mcp = FastMCP(
    "Weather Service",
    # 用于身份验证的令牌验证器
    token_verifier=SimpleTokenVerifier(),
    # 认证设置
    auth=AuthSettings(
        issuer_url=AnyHttpUrl("https://auth.example.com"),  # 授权服务器URL
        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # 此服务器的URL
        required_scopes=["user"],
    ),
)

@mcp.tool()
async def get_weather(city: str = "London") -> dict[str, str]:
    """获取城市天气数据"""
    return {
        "city": city,
        "temperature": "22",
        "condition": "Partly cloudy",
        "humidity": "65%",
    }

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

带有认证的MCP client示例

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

client = MultiServerMCPClient(
    {
        "weather": {
            "transport": "streamable_http",
            "url": "http://localhost:8000/mcp",
            # 设置请求头，配置用于身份验证的token
            "headers": {
                "Authorization": "Bearer YOUR_TOKEN",
                "X-Custom-Header": "custom-value"
            },
        }
    }
)

Jean's Blog

MCP服务开发的核心概念和使用