1. 集成与部署 #

本章目标：

• 理解 MCP 服务器的三种运行模式：开发模式、直接执行、Claude Desktop 集成
• 使用 mcp dev 进行开发调试与实时重载
• 配置 Claude Desktop 集成并测试端到端功能
• 掌握生产环境部署的最佳实践

2. 功能完整的示例服务器 #

新建 C:\mcp-project\integration_server.py：

# 导入异步IO模块
import asyncio

# 导入datetime用于时间处理
from datetime import datetime

# 导入Any类型用于类型注解
from typing import Any

# 导入FastMCP框架
from mcp.server.fastmcp import FastMCP, Context

# 1) 创建 FastMCP 服务器实例
# 创建FastMCP服务器对象，设置名称和说明
server = FastMCP(
    name="integration-server",  # 服务器名称
    instructions="这是一个用于演示集成与部署的服务器，提供文件操作、时间查询等功能。",  # 服务器说明
)


# 2) 文件操作工具
# 注册一个工具用于读取文件内容
@server.tool()
async def read_file(ctx: Context, file_path: str) -> str:
    """读取指定文件的内容。"""
    try:
        # 记录操作日志
        ctx.info(f"正在读取文件：{file_path}")

        # 模拟文件读取，根据文件名返回不同内容
        if file_path == "config.txt":
            content = "服务器配置：\n- 端口：8080\n- 模式：生产\n- 日志级别：INFO"
        elif file_path == "status.log":
            content = f"状态日志：\n- 启动时间：{datetime.now()}\n- 运行状态：正常\n- 内存使用：128MB"
        else:
            content = f"文件 {file_path} 不存在或无法访问"

        # 报告进度
        ctx.report_progress(1.0, 1.0, "文件读取完成")

        # 返回文件内容
        return content
    except Exception as e:
        # 记录错误日志
        ctx.error(f"读取文件时发生错误：{e}")
        # 抛出异常
        raise


# 注册一个工具用于写入文件内容
@server.tool()
async def write_file(ctx: Context, file_path: str, content: str) -> dict[str, Any]:
    """写入内容到指定文件。"""
    try:
        # 记录操作日志
        ctx.info(f"正在写入文件：{file_path}")

        # 模拟文件写入，仅允许写入以"temp_"开头的文件
        if file_path.startswith("temp_"):
            # 模拟成功写入，返回写入结果
            result = {
                "status": "success",
                "file_path": file_path,
                "bytes_written": len(content.encode("utf-8")),
                "timestamp": datetime.now().isoformat(),
            }
            # 记录写入成功日志
            ctx.info(f"文件写入成功：{result['bytes_written']} 字节")
        else:
            # 模拟权限错误，抛出异常
            raise PermissionError(f"没有权限写入文件：{file_path}")

        # 返回写入结果
        return result
    except Exception as e:
        # 记录错误日志
        ctx.error(f"写入文件时发生错误：{e}")
        # 抛出异常
        raise


# 3) 系统信息工具
# 注册一个工具用于获取系统信息
@server.tool()
async def get_system_info(ctx: Context) -> dict[str, Any]:
    """获取系统信息。"""
    # 记录日志
    ctx.info("正在收集系统信息")

    # 导入平台和psutil模块
    import platform
    import psutil

    try:
        # 收集系统信息
        system_info = {
            "platform": platform.system(),
            "platform_version": platform.version(),
            "python_version": platform.python_version(),
            "cpu_count": psutil.cpu_count(),
            "memory_total": f"{psutil.virtual_memory().total // (1024**3)}GB",
            "memory_available": f"{psutil.virtual_memory().available // (1024**3)}GB",
            "timestamp": datetime.now().isoformat(),
        }

        # 记录收集完成日志
        ctx.info("系统信息收集完成")
        # 返回系统信息
        return system_info
    except ImportError:
        # psutil未安装时，返回基础信息
        ctx.warning("psutil 未安装，返回基本系统信息")
        return {
            "platform": platform.system(),
            "platform_version": platform.version(),
            "python_version": platform.python_version(),
            "note": "psutil 未安装，无法获取详细硬件信息",
            "timestamp": datetime.now().isoformat(),
        }


# 4) 时间相关工具
# 注册一个工具用于获取当前时间
@server.tool()
async def get_current_time(ctx: Context, timezone: str = "UTC") -> dict[str, Any]:
    """获取当前时间信息。"""
    # 记录日志
    ctx.info(f"正在获取 {timezone} 时区的当前时间")

    # 导入时区相关模块
    from datetime import timezone as tz
    import pytz

    try:
        # 获取指定时区的当前时间
        if timezone != "UTC":
            tz_obj = pytz.timezone(timezone)
            current_time = datetime.now(tz_obj)
        else:
            current_time = datetime.now(tz.utc)

        # 构造时间信息结果
        result = {
            "timezone": timezone,
            "current_time": current_time.isoformat(),
            "formatted_time": current_time.strftime("%Y-%m-%d %H:%M:%S"),
            "timestamp": current_time.timestamp(),
        }

        # 记录获取完成日志
        ctx.info(f"时间信息获取完成：{result['formatted_time']}")
        # 返回时间信息
        return result
    except Exception as e:
        # 记录错误日志
        ctx.error(f"获取时间信息时发生错误：{e}")
        # 抛出异常
        raise


# 5) 资源定义
# 注册一个资源用于获取服务器配置信息
@server.resource("config://settings")
async def get_settings() -> str:
    """获取服务器配置信息。"""
    # 返回配置信息字符串
    return """# 服务器配置
    server_name: integration-server
    version: 1.0.0
    debug: true
    log_level: INFO
    max_connections: 100
    timeout: 30s
    """


# 注册一个资源用于获取服务器健康状态
@server.resource("status://health")
async def get_health_status() -> str:
    """获取服务器健康状态。"""
    # 返回健康状态字符串
    return f"""# 服务器健康状态
    status: healthy
    uptime: {datetime.now().isoformat()}
    version: 1.0.0
    last_check: {datetime.now().isoformat()}
    """


# 6) 提示模板
# 注册一个提示模板用于系统状态分析
@server.prompt("system_status")
async def get_system_status_prompt() -> str:
    """获取系统状态提示模板。"""
    # 返回提示模板内容
    return """请分析当前系统状态并提供以下信息：
    1. 系统运行状态
    2. 资源使用情况
    3. 潜在问题或建议
    4. 性能优化建议

    请使用 markdown 格式输出。"""


# 7) 主运行函数（用于直接执行）
# 定义主运行函数，用于直接运行服务器
def run_server():
    """直接运行服务器（不通过 CLI）。"""
    # 使用 stdio 传输方式启动服务器
    server.run(transport="stdio")


# 主入口
# 判断是否为主程序入口
if __name__ == "__main__":
    # 直接运行服务器，不需要 asyncio.run()
    run_server()

要点：

• 服务器包含多种工具类型：文件操作、系统信息、时间查询。
• 使用 ctx.info()、ctx.error() 等记录详细日志。
• 支持结构化输出与错误处理。
• 提供资源与提示模板。

3. Claude Desktop 集成配置 #

新建 C:\mcp-project\claude_desktop_config.json：

{
  "mcpServers": {
    "integration-server": {
      "command": "python",
      "args": ["C:/mcp-project/integration_server.py"],
      "env": {
        "PYTHONPATH": "C:/Users/Administrator/AppData/Local/Programs/Python/Python313"
      }
    }
  }
}

说明：

• 将此配置添加到 Claude Desktop 的 MCP 设置中。
• 路径使用绝对路径确保正确启动。
• 设置 PYTHONPATH 确保模块导入正常。

4. 开发模式测试客户端 #

新建 C:\mcp-project\dev_mode_client.py：

# 导入异步IO模块
import asyncio

# 导入操作系统相关模块
import os

# 从mcp包导入客户端会话、stdio服务器参数和类型定义
from mcp import ClientSession, StdioServerParameters, types

# 导入stdio客户端适配器
from mcp.client.stdio import stdio_client


# 定义异步函数，测试服务器的各项功能
async def test_server_functionality():
    """测试服务器的各项功能。"""
    # 1) 获取当前文件的绝对路径所在目录
    base_dir = os.path.dirname(os.path.abspath(__file__))
    # 2) 拼接得到 integration_server.py 的绝对路径
    server_path = os.path.join(base_dir, "integration_server.py")

    # 3) 配置以 stdio 方式启动服务器的参数
    server_params = StdioServerParameters(
        command="python",  # 使用 python 运行
        args=[server_path],  # 传入服务器脚本路径
        env={},  # 环境变量为空
    )

    # 4) 建立 stdio 连接并创建客户端会话
    async with stdio_client(server_params) as (read, write):
        # 基于读写流创建 MCP 客户端会话
        async with ClientSession(read, write) as session:
            # 5) 初始化 MCP 会话
            await session.initialize()
            print("✓ 会话初始化成功")

            # 6) 列出所有可用工具
            tools = await session.list_tools()
            print(f"✓ 可用工具：{[t.name for t in tools.tools]}")

            # 7) 列出所有可用资源
            resources = await session.list_resources()
            print(f"✓ 可用资源：{[r.uri for r in resources.resources]}")

            # 8) 列出所有可用提示模板
            prompts = await session.list_prompts()
            print(f"✓ 可用提示：{[p.name for p in prompts.prompts]}")

            # 9) 测试文件读取工具
            print("\n=== 测试文件读取 ===")
            try:
                # 调用 read_file 工具，读取 config.txt 文件内容
                result = await session.call_tool(
                    "read_file", {"file_path": "config.txt"}
                )
                # 打印读取结果的前50个字符
                print(f"✓ 文件读取成功：{result.content[0].text[:50]}...")
            except Exception as e:
                # 捕获异常并打印错误信息
                print(f"✗ 文件读取失败：{e}")

            # 10) 测试文件写入工具
            print("\n=== 测试文件写入 ===")
            try:
                # 调用 write_file 工具，写入内容到 temp_test.txt
                result = await session.call_tool(
                    "write_file",
                    {"file_path": "temp_test.txt", "content": "这是一个测试文件"},
                )
                # 打印写入结果内容
                print(f"✓ 文件写入成功：{result.content[0].text}")
            except Exception as e:
                # 捕获异常并打印错误信息
                print(f"✗ 文件写入失败：{e}")

            # 11) 测试系统信息工具
            print("\n=== 测试系统信息 ===")
            try:
                # 调用 get_system_info 工具，获取系统信息
                result = await session.call_tool("get_system_info", {})
                # 打印系统信息的前100个字符
                print(f"✓ 系统信息获取成功：{result.content[0].text[:100]}...")
            except Exception as e:
                # 捕获异常并打印错误信息
                print(f"✗ 系统信息获取失败：{e}")

            # 12) 测试时间工具
            print("\n=== 测试时间工具 ===")
            try:
                # 调用 get_current_time 工具，获取UTC时间
                result = await session.call_tool(
                    "get_current_time", {"timezone": "UTC"}
                )
                # 打印时间信息内容
                print(f"✓ 时间信息获取成功：{result.content[0].text}")
            except Exception as e:
                # 捕获异常并打印错误信息
                print(f"✗ 时间信息获取失败：{e}")

            # 13) 测试资源读取
            print("\n=== 测试资源读取 ===")
            try:
                # 导入 AnyUrl 类型
                from pydantic import AnyUrl

                # 读取 config://settings 资源内容
                result = await session.read_resource(AnyUrl("config://settings"))
                # 打印资源内容的前50个字符
                print(f"✓ 配置资源读取成功：{result.contents[0].text[:50]}...")
            except Exception as e:
                # 捕获异常并打印错误信息
                print(f"✗ 配置资源读取失败：{e}")

            # 14) 测试提示模板
            print("\n=== 测试提示模板 ===")
            try:
                # 获取名为 system_status 的提示模板内容
                result = await session.get_prompt("system_status")
                texts = []
                for message in result.messages:
                    texts.append(message.content.text)
                # 打印提示模板内容的前100个字符
                print(f"✓ 提示模板获取成功：{"".join(texts)}...")
            except Exception as e:
                # 捕获异常并打印错误信息
                print(f"✗ 提示模板获取失败：{e}")

            # 所有测试完成
            print("\n=== 所有测试完成 ===")


# 定义主异步函数
async def main():
    """主函数。"""
    # 打印测试开始信息
    print("开始测试集成服务器功能...")
    # 调用测试函数
    await test_server_functionality()


# 判断是否为主程序入口
if __name__ == "__main__":
    # 运行主异步函数
    asyncio.run(main())

5. 运行与验证 #

4.1 开发模式运行 #

cd C:\mcp-project

# 方式1：使用 mcp dev（推荐开发调试）
mcp dev integration_server.py

# 方式2：直接运行 Python 脚本
python integration_server.py

# 方式3：使用 mcp run
mcp run integration_server.py

4.2 测试客户端 #

python dev_mode_client.py

预期输出：

开始测试集成服务器功能...
✓ 会话初始化成功
✓ 可用工具：['read_file', 'write_file', 'get_system_info', 'get_current_time']
✓ 可用资源：[AnyUrl('config://settings'), AnyUrl('status://health')]
✓ 可用提示：['system_status']

=== 测试文件读取 ===
✓ 文件读取成功：服务器配置：
- 端口：8080
- 模式：生产
- 日志级别：INFO...

=== 测试文件写入 ===
✓ 文件写入成功：{
  "status": "success",
  "file_path": "temp_test.txt",
  "bytes_written": 24,
  "timestamp": "2025-08-21T22:41:51.834457"
}

=== 测试系统信息 ===
✓ 系统信息获取成功：{
  "platform": "Windows",
  "platform_version": "10.0.19045",
  "python_version": "3.13.5",
  "cpu_...

=== 测试时间工具 ===
✓ 时间信息获取成功：{
  "timezone": "UTC",
  "current_time": "2025-08-21T14:41:51.904560+00:00",
  "formatted_time": "2025-08-21 14:41:51",
  "timestamp": 1755787311.90456
}

=== 测试资源读取 ===
✓ 配置资源读取成功：# 服务器配置
    server_name: integration-server
    ve...

=== 测试提示模板 ===
✓ 提示模板获取成功：请分析当前系统状态并提供以下信息：
    1. 系统运行状态
    2. 资源使用情况
    3. 潜在问题或建议
    4. 性能优化建议

    请使用 markdown 格式输出。...

=== 所有测试完成 ===

6. 生产环境部署配置 #

6.1 环境变量配置 #

新建 C:\mcp-project\.env：

# 生产环境配置
# 设置服务器名称
MCP_SERVER_NAME=integration-server
# 设置服务器版本号
MCP_SERVER_VERSION=1.0.0
# 是否开启调试模式（生产环境应为 false）
MCP_DEBUG=false
# 设置日志级别
MCP_LOG_LEVEL=INFO
# 最大连接数限制
MCP_MAX_CONNECTIONS=100
# 超时时间（单位：秒）
MCP_TIMEOUT=30

6.2 日志配置 #

新建 C:\mcp-project\logging_config.py：

# logging_config.py
# 导入标准日志库
import logging
# 导入日志处理器模块
import logging.handlers
# 导入操作系统相关模块
import os
# 从pathlib导入Path用于路径操作
from pathlib import Path

# 定义日志配置函数
def setup_logging():
    """设置生产环境日志配置。"""
    # 创建日志目录（如果不存在则自动创建）
    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)

    # 配置根日志记录器
    logging.basicConfig(
        # 设置日志级别为INFO
        level=logging.INFO,
        # 设置日志输出格式
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        # 配置日志处理器列表
        handlers=[
            # 控制台输出日志
            logging.StreamHandler(),
            # 文件输出（按日期轮转，每天生成一个新日志文件，保留30天）
            logging.handlers.TimedRotatingFileHandler(
                log_dir / "mcp_server.log",
                when="midnight",         # 每天午夜轮转
                interval=1,              # 间隔1天
                backupCount=30,          # 最多保留30个备份
                encoding="utf-8"         # 日志文件编码为utf-8
            ),
            # 错误日志单独记录（文件大小达到10MB轮转，最多保留5个备份）
            logging.handlers.RotatingFileHandler(
                log_dir / "error.log",
                maxBytes=10*1024*1024,   # 最大10MB
                backupCount=5,           # 最多保留5个备份
                encoding="utf-8"         # 日志文件编码为utf-8
            )
        ]
    )

    # 设置第三方库uvicorn的日志级别为WARNING，减少无关日志输出
    logging.getLogger("uvicorn").setLevel(logging.WARNING)
    # 设置第三方库httpx的日志级别为WARNING，减少无关日志输出
    logging.getLogger("httpx").setLevel(logging.WARNING)