skillhub-013-mcp-builder/reference/python_mcp_server.md

Python MCP 服务器实现指南

概述

本文档提供了使用 MCP Python SDK 实现 MCP 服务器的 Python 特定最佳实践和示例。内容涵盖服务器设置、工具注册模式、基于 Pydantic 的输入验证、错误处理以及完整的工作示例。

---

快速参考

关键导入

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx

服务器初始化

mcp = FastMCP("service_mcp")

工具注册模式

@mcp.tool(name="tool_name", annotations={...})
async def tool_function(params: InputModel) -> str:
    # 实现代码
    pass

---

MCP Python SDK 与 FastMCP

官方的 MCP Python SDK 提供了 FastMCP，这是一个用于构建 MCP 服务器的高级框架。它提供了以下能力：

• 从函数签名和文档字符串自动生成描述和 inputSchema
• 集成 Pydantic 模型进行输入验证
• 基于装饰器的工具注册，使用 @mcp.tool

如需完整的 SDK 文档，请使用 WebFetch 加载： https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md

服务器命名规范

Python MCP 服务器必须遵循以下命名模式：

• 格式：{service}_mcp（小写，使用下划线）
• 示例：github_mcp、jira_mcp、stripe_mcp

名称应符合以下要求：

• 通用性（不与特定功能绑定）
• 能够描述所集成的服务/API
• 易于从任务描述中推断
• 不包含版本号或日期

工具实现

工具命名

工具名称使用蛇形命名法（snake_case）（例如 "search_users"、"create_project"、"get_channel_info"），名称应清晰且具有动作导向。

避免命名冲突：加入服务上下文以防止重叠：

• 使用 "slack_send_message" 而非仅用 "send_message"
• 使用 "github_create_issue" 而非仅用 "create_issue"
• 使用 "asana_list_tasks" 而非仅用 "list_tasks"

使用 FastMCP 的工具结构

工具通过 @mcp.tool 装饰器定义，并使用 Pydantic 模型进行输入验证：

from pydantic import BaseModel, Field, ConfigDict
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("example_mcp")

# 定义用于输入验证的 Pydantic 模型
class ServiceToolInput(BaseModel):
    '''服务工具操作的输入模型。'''
    model_config = ConfigDict(
        str_strip_whitespace=True,  # 自动去除字符串首尾空白
        validate_assignment=True,    # 赋值时进行验证
        extra='forbid'              # 禁止额外字段
    )

    param1: str = Field(..., description="第一个参数描述（例如 'user123'、'project-abc'）", min_length=1, max_length=100)
    param2: Optional[int] = Field(default=None, description="带约束的可选整数参数", ge=0, le=1000)
    tags: Optional[List[str]] = Field(default_factory=list, description="要应用的标签列表", max_items=10)

@mcp.tool(
    name="service_tool_name",
    annotations={
        "title": "人类可读的工具标题",
        "readOnlyHint": True,     # 工具不会修改环境
        "destructiveHint": False,  # 工具不会执行破坏性操作
        "idempotentHint": True,    # 重复调用不会产生额外效果
        "openWorldHint": False     # 工具不与外部实体交互
    }
)
async def service_tool_name(params: ServiceToolInput) -> str:
    '''工具描述自动成为 'description' 字段。

    该工具对服务执行特定操作。在处理之前，它会使用 ServiceToolInput
    Pydantic 模型验证所有输入。

    Args:
        params (ServiceToolInput): 经过验证的输入参数，包含：
            - param1 (str): 第一个参数描述
            - param2 (Optional[int]): 带默认值的可选参数
            - tags (Optional[List[str]]): 标签列表

    Returns:
        str: 包含操作结果的 JSON 格式响应
    '''
    # 实现代码
    pass

Pydantic v2 关键特性

• 使用 model_config 代替嵌套的 Config 类
• 使用 field_validator 代替已弃用的 validator
• 使用 model_dump() 代替已弃用的 dict()
• 验证器需要 @classmethod 装饰器
• 验证器方法必须包含类型提示

from pydantic import BaseModel, Field, field_validator, ConfigDict

class CreateUserInput(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    name: str = Field(..., description="用户全名", min_length=1, max_length=100)
    email: str = Field(..., description="用户电子邮件地址", pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(..., description="用户年龄", ge=0, le=150)

    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("电子邮件不能为空")
        return v.lower()

响应格式选项

支持多种输出格式以提高灵活性：

from enum import Enum

class ResponseFormat(str, Enum):
    '''工具响应的输出格式。'''
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    query: str = Field(..., description="搜索查询")
    response_format: ResponseFormat = Field(
        default=ResponseFormat.MARKDOWN,
        description="输出格式：'markdown' 用于人类可读，'json' 用于机器可读"
    )

Markdown 格式：

• 使用标题、列表和格式提高可读性
• 将时间戳转换为人类可读格式（例如使用 "2024-01-15 10:30:00 UTC" 而非时间戳）
• 显示名称并附带括号内的 ID（例如 "@john.doe (U123456)"）
• 省略冗长的元数据（例如只显示一个个人资料图片 URL，而非所有尺寸）
• 按逻辑对相关信息进行分组

JSON 格式：

• 返回适合程序化处理的完整、结构化数据
• 包含所有可用字段和元数据
• 使用一致的字段名称和类型

分页实现

对于列出资源的工具：

class ListInput(BaseModel):
    limit: Optional[int] = Field(default=20, description="要返回的最大结果数", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="分页时要跳过的结果数", ge=0)

async def list_items(params: ListInput) -> str:
    # 使用分页参数发起 API 请求
    data = await api_request(limit=params.limit, offset=params.offset)

    # 返回分页信息
    response = {
        "total": data["total"],
        "count": len(data["items"]),
        "offset": params.offset,
        "items": data["items"],
        "has_more": data["total"] > params.offset + len(data["items"]),
        "next_offset": params.offset + len(data["items"]) if data["total"] > params.offset + len(data["items"]) else None
    }
    return json.dumps(response, indent=2)

错误处理

提供清晰、可操作的错误消息：

def _handle_api_error(e: Exception) -> str:
    '''在所有工具中保持一致的错误格式化。'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "错误：资源未找到。请检查 ID 是否正确。"
        elif e.response.status_code == 403:
            return "错误：权限被拒绝。您没有访问此资源的权限。"
        elif e.response.status_code == 429:
            return "错误：超出速率限制。请等待后再发起更多请求。"
        return f"错误：API 请求失败，状态码 {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "错误：请求超时。请重试。"
    return f"错误：发生意外错误：{type(e).__name__}"

共享工具函数

将通用功能提取为可复用的函数：

# 共享的 API 请求函数
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''所有 API 调用的可复用函数。'''
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs
        )
        response.raise_for_status()
        return response.json()

Async/Await 最佳实践

始终对网络请求和 I/O 操作使用 async/await：

# 推荐：异步网络请求
async def fetch_data(resource_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{API_URL}/resource/{resource_id}")
        response.raise_for_status()
        return response.json()

# 不推荐：同步请求
def fetch_data(resource_id: str) -> dict:
    response = requests.get(f"{API_URL}/resource/{resource_id}")  # 阻塞
    return response.json()

类型提示

全程使用类型提示：

from typing import Optional, List, Dict, Any

async def get_user(user_id: str) -> Dict[str, Any]:
    data = await fetch_user(user_id)
    return {"id": data["id"], "name": data["name"]}

工具文档字符串

每个工具必须包含带有显式类型信息的全面文档字符串：

async def search_users(params: UserSearchInput) -> str:
    '''
    通过名称、电子邮件或团队在 Example 系统中搜索用户。

    该工具搜索 Example 平台中的所有用户资料，支持部分匹配和各种
    搜索筛选条件。它不会创建或修改用户，仅搜索现有用户。

    Args:
        params (UserSearchInput): 经过验证的输入参数，包含：
            - query (str): 用于匹配名称/电子邮件的搜索字符串（例如 "john"、"@example.com"、"team:marketing"）
            - limit (Optional[int]): 要返回的最大结果数，范围 1-100（默认：20）
            - offset (Optional[int]): 分页时要跳过的结果数（默认：0）

    Returns:
        str: 包含搜索结果的 JSON 格式字符串，schema 如下：

        成功响应：
        {
            "total": int,           # 找到的匹配总数
            "count": int,           # 当前响应中的结果数
            "offset": int,          # 当前分页偏移量
            "users": [
                {
                    "id": str,      # 用户 ID（例如 "U123456789"）
                    "name": str,    # 全名（例如 "John Doe"）
                    "email": str,   # 电子邮件地址（例如 "john@example.com"）
                    "team": str     # 团队名称（例如 "Marketing"）- 可选
                }
            ]
        }

        错误响应：
        "错误：<错误消息>" 或 "未找到匹配 '<query>' 的用户"

    Examples:
        - 何时使用："查找所有市场团队成员" -> 参数 query="team:marketing"
        - 何时使用："搜索 John 的账号" -> 参数 query="john"
        - 何时不使用：需要创建用户时（应使用 example_create_user）
        - 何时不使用：已有用户 ID 并需要完整详情时（应使用 example_get_user）

    错误处理：
        - 输入验证错误由 Pydantic 模型处理
        - 如果请求过多（429 状态），返回"错误：超出速率限制"
        - 如果 API 密钥无效（401 状态），返回"错误：API 身份验证无效"
        - 返回格式化的结果列表，或"未找到匹配 'query' 的用户"
    '''

完整示例

参见以下完整的 Python MCP 服务器示例：

#!/usr/bin/env python3
'''
Example 服务的 MCP 服务器。

该服务器提供与 Example API 交互的工具，包括用户搜索、
项目管理和数据导出功能。
'''

from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("example_mcp")

# 常量
API_BASE_URL = "https://api.example.com/v1"

# 枚举
class ResponseFormat(str, Enum):
    '''工具响应的输出格式。'''
    MARKDOWN = "markdown"
    JSON = "json"

# 用于输入验证的 Pydantic 模型
class UserSearchInput(BaseModel):
    '''用户搜索操作的输入模型。'''
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    query: str = Field(..., description="用于匹配名称/电子邮件的搜索字符串", min_length=2, max_length=200)
    limit: Optional[int] = Field(default=20, description="要返回的最大结果数", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="分页时要跳过的结果数", ge=0)
    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="输出格式")

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("查询字符串不能为空或仅包含空白字符")
        return v.strip()

# 共享工具函数
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''所有 API 调用的可复用函数。'''
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs
        )
        response.raise_for_status()
        return response.json()

def _handle_api_error(e: Exception) -> str:
    '''在所有工具中保持一致的错误格式化。'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "错误：资源未找到。请检查 ID 是否正确。"
        elif e.response.status_code == 403:
            return "错误：权限被拒绝。您没有访问此资源的权限。"
        elif e.response.status_code == 429:
            return "错误：超出速率限制。请等待后再发起更多请求。"
        return f"错误：API 请求失败，状态码 {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "错误：请求超时。请重试。"
    return f"错误：发生意外错误：{type(e).__name__}"

# 工具定义
@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "搜索 Example 用户",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True
    }
)
async def example_search_users(params: UserSearchInput) -> str:
    '''通过名称、电子邮件或团队在 Example 系统中搜索用户。

    [完整文档字符串如上所示]
    '''
    try:
        # 使用验证后的参数发起 API 请求
        data = await _make_api_request(
            "users/search",
            params={
                "q": params.query,
                "limit": params.limit,
                "offset": params.offset
            }
        )

        users = data.get("users", [])
        total = data.get("total", 0)

        if not users:
            return f"未找到匹配 '{params.query}' 的用户"

        # 根据请求的格式格式化响应
        if params.response_format == ResponseFormat.MARKDOWN:
            lines = [f"# 用户搜索结果：'{params.query}'", ""]
            lines.append(f"找到 {total} 个用户（显示 {len(users)} 个）")
            lines.append("")

            for user in users:
                lines.append(f"## {user['name']} ({user['id']})")
                lines.append(f"- **电子邮件**：{user['email']}")
                if user.get('team'):
                    lines.append(f"- **团队**：{user['team']}")
                lines.append("")

            return "\n".join(lines)

        else:
            # 机器可读的 JSON 格式
            import json
            response = {
                "total": total,
                "count": len(users),
                "offset": params.offset,
                "users": users
            }
            return json.dumps(response, indent=2)

    except Exception as e:
        return _handle_api_error(e)

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

---

FastMCP 高级特性

上下文参数注入

FastMCP 可以自动将 Context 参数注入工具，以实现日志记录、进度报告、资源读取和用户交互等高级功能：

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("example_mcp")

@mcp.tool()
async def advanced_search(query: str, ctx: Context) -> str:
    '''具有上下文访问权限的高级工具，支持日志记录和进度报告。'''

    # 报告长时间操作的进度
    await ctx.report_progress(0.25, "开始搜索...")

    # 记录调试信息
    await ctx.log_info("正在处理查询", {"query": query, "timestamp": datetime.now()})

    # 执行搜索
    results = await search_api(query)
    await ctx.report_progress(0.75, "正在格式化结果...")

    # 访问服务器配置
    server_name = ctx.fastmcp.name

    return format_results(results)

@mcp.tool()
async def interactive_tool(resource_id: str, ctx: Context) -> str:
    '''可以从用户处请求额外输入的工具。'''

    # 在需要时请求敏感信息
    api_key = await ctx.elicit(
        prompt="请提供您的 API 密钥：",
        input_type="password"
    )

    # 使用提供的密钥
    return await api_call(resource_id, api_key)

Context 功能：

• ctx.report_progress(progress, message) - 报告长时间操作的进度
• ctx.log_info(message, data) / ctx.log_error() / ctx.log_debug() - 日志记录
• ctx.elicit(prompt, input_type) - 向用户请求输入
• ctx.fastmcp.name - 访问服务器配置
• ctx.read_resource(uri) - 读取 MCP 资源

资源注册

将数据暴露为资源，实现高效的基于模板的访问：

@mcp.resource("file://documents/{name}")
async def get_document(name: str) -> str:
    '''将文档暴露为 MCP 资源。

    资源适用于不需要复杂参数的静态或半静态数据。
    它们使用 URI 模板实现灵活的访问。
    '''
    document_path = f"./docs/{name}"
    with open(document_path, "r") as f:
        return f.read()

@mcp.resource("config://settings/{key}")
async def get_setting(key: str, ctx: Context) -> str:
    '''将配置暴露为带上下文的资源。'''
    settings = await load_settings()
    return json.dumps(settings.get(key, {}))

何时使用资源与工具：

• 资源：适用于带简单参数的数据访问（URI 模板）
• 工具：适用于需要验证和业务逻辑的复杂操作

结构化输出类型

FastMCP 支持多种返回类型，不仅限于字符串：

from typing import TypedDict
from dataclasses import dataclass
from pydantic import BaseModel

# 用于结构化返回的 TypedDict
class UserData(TypedDict):
    id: str
    name: str
    email: str

@mcp.tool()
async def get_user_typed(user_id: str) -> UserData:
    '''返回结构化数据 - FastMCP 处理序列化。'''
    return {"id": user_id, "name": "John Doe", "email": "john@example.com"}

# 用于复杂验证的 Pydantic 模型
class DetailedUser(BaseModel):
    id: str
    name: str
    email: str
    created_at: datetime
    metadata: Dict[str, Any]

@mcp.tool()
async def get_user_detailed(user_id: str) -> DetailedUser:
    '''返回 Pydantic 模型 - 自动生成 schema。'''
    user = await fetch_user(user_id)
    return DetailedUser(**user)

生命周期管理

初始化跨请求持久化的资源：

from contextlib import asynccontextmanager

@asynccontextmanager
async def app_lifespan():
    '''管理在服务器整个生命周期内存在的资源。'''
    # 初始化连接、加载配置等
    db = await connect_to_database()
    config = load_configuration()

    # 使对所有工具可用
    yield {"db": db, "config": config}

    # 关闭时清理
    await db.close()

mcp = FastMCP("example_mcp", lifespan=app_lifespan)

@mcp.tool()
async def query_data(query: str, ctx: Context) -> str:
    '''通过上下文访问生命周期资源。'''
    db = ctx.request_context.lifespan_state["db"]
    results = await db.query(query)
    return format_results(results)

传输选项

FastMCP 支持两种主要的传输机制：

# stdio 传输（适用于本地工具）- 默认
if __name__ == "__main__":
    mcp.run()

# Streamable HTTP 传输（适用于远程服务器）
if __name__ == "__main__":
    mcp.run(transport="streamable_http", port=8000)

传输选择：

• stdio：命令行工具、本地集成、子进程执行
• Streamable HTTP：Web 服务、远程访问、多客户端

---

代码最佳实践

代码可组合性与可复用性

您的实现必须优先考虑可组合性和代码复用：

1. 提取通用功能：

   • 为跨多个工具使用的操作创建可复用的辅助函数
   • 构建共享的 API 客户端用于 HTTP 请求，而不是重复代码
   • 将错误处理逻辑集中到工具函数中
   • 将业务逻辑提取到可组合的专用函数中
   • 提取共享的 markdown 或 JSON 字段选择与格式化功能

2. 避免重复：

   • 切勿在工具之间复制粘贴相似的代码
   • 如果发现自己编写了两次相似的逻辑，将其提取为函数
   • 分页、过滤、字段选择和格式化等通用操作应共享
   • 身份验证/授权逻辑应集中处理

Python 特定最佳实践

1. 使用类型提示：始终包含函数参数和返回值的类型注解
2. Pydantic 模型：为所有输入验证定义清晰的 Pydantic 模型
3. 避免手动验证：让 Pydantic 使用约束来处理输入验证
4. 正确的导入：对导入进行分组（标准库、第三方、本地）
5. 错误处理：使用特定的异常类型（httpx.HTTPStatusError，而非通用的 Exception）
6. 异步上下文管理器：对需要清理的资源使用 async with
7. 常量：使用大写字母定义模块级常量

质量检查清单

在最终确定您的 Python MCP 服务器实现之前，请确保：

策略设计

• 工具支持完整的工作流程，而不仅仅是 API 端点封装
• 工具名称反映自然的任务划分
• 响应格式针对 agent 上下文效率进行优化
• 在适当之处使用人类可读的标识符
• 错误消息引导 agent 正确使用

实现质量

• 聚焦实现：实现最重要和最具有价值的工具
• 所有工具都有描述性名称和文档
• 相似操作的返回类型保持一致
• 所有外部调用都实现了错误处理
• 服务器名称遵循格式：{service}_mcp
• 所有网络操作使用 async/await
• 通用功能被提取为可复用的函数
• 错误消息清晰、可操作且具有教育意义
• 输出经过正确验证和格式化

工具配置

• 所有工具在装饰器中实现了 'name' 和 'annotations'
• 注解设置正确（readOnlyHint、destructiveHint、idempotentHint、openWorldHint）
• 所有工具使用 Pydantic BaseModel 进行输入验证，并附带 Field() 定义
• 所有 Pydantic Field 具有显式类型、描述和约束
• 所有工具具有包含显式输入/输出类型的全面文档字符串
• 文档字符串包含 dict/JSON 返回的完整 schema 结构
• Pydantic 模型处理输入验证（无需手动验证）

高级特性（如适用）

• 使用上下文注入进行日志记录、进度报告或信息征求
• 为适当的数据端点注册了资源
• 为持久连接实现了生命周期管理
• 使用了结构化输出类型（TypedDict、Pydantic 模型）
• 配置了合适的传输方式（stdio 或 Streamable HTTP）

代码质量

• 文件包含正确的导入，包括 Pydantic 导入
• 在适用之处正确实现了分页
• 为可能较大的结果集提供了过滤选项
• 所有异步函数使用 async def 正确定义
• HTTP 客户端使用遵循异步模式并配有正确的上下文管理器
• 整个代码中使用了类型提示
• 常量使用大写字母在模块级别定义

测试

• 服务器成功运行：python your_server.py --help
• 所有导入正确解析
• 示例工具调用按预期工作
• 错误场景得到优雅处理