Python MCP 服务器模板

简介

这是一个用于创建和管理 MCP (Model Context Protocol) 服务器的通用 Python 模板。该模板遵循生产环境中 MCP 服务器的最佳实践，并提供了一个结构化的开发框架，方便快速启动项目。

特性

• 结构清晰: 提供了完整的代码目录结构和文件示例
• 日志系统: 集成了灵活的日志记录机制，默认支持 INFO 级别日志
• 配置管理: 使用 Pydantic 模型进行配置管理和环境变量处理
• 错误处理: 实现了统一的错误捕捉和响应机制
• 服务状态: 提供了服务初始化和清理流程，确保资源管理正确

快速开始

项目结构

template-mcp-server/
├── src/
│   ├── server.py                 # 主程序入口
│   └── mcp_server/              # 核心功能包
│       ├── config.py            # 配置加载与管理
│       ├── logger.py            # 日志模块
│       ├── service_client.py    # 服务客户端实现
│       └── errors.py           # 定义自定义错误类
├── requirements.txt          # 项目依赖管理
└── .env.example              # 环境变量示例文件

安装依赖

pip install -r requirements.txt

使用说明

环境变量配置

通过设置以下环境变量来配置服务器行为：

• SERVICE_API_KEY: 你的服务 API 密钥（必填）
• SERVICE_ENDPOINT: 服务端点 URL（默认: https://api.example.com）
• SERVICE_TIMEOUT: 请求超时时间（单位：秒，默认: 30）
• LOG_LEVEL: 日志级别（默认: INFO）

启动服务器

# 基本启动方式
python src/server.py

# 设置自定义配置
SERVICE_API_KEY=test python src/server.py

开发指南

配置管理

使用 Pydantic 模型加载和验证配置：

from pydantic import BaseSettings

class Settings(BaseSettings):
    api_key: str
    endpoint: str = "https://api.example.com"
    timeout: int = 30
    log_level: str = "INFO"

settings = Settings(_env_file=["./.env", ""])

日志系统

实现一个简单的日志记录类：

import logging

class Logger:
    def __init__(self, name: str, level: int = logging.INFO):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(level)
        handler = logging.StreamHandler()
        formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def info(self, message: str):
        self.logger.info(message)

    def error(self, message: str):
        self.logger.error(message)

    def debug(self, message: str):
        self.logger.debug(message)

服务客户端

实现一个通用的服务客户端：

import httpx

class ServiceClient:
    def __init__(self, endpoint: str, timeout: int):
        self.client = httpx.Client(base_url=endpoint, timeout=timeout)

    async def fetch_data(self, path: str) -> dict:
        try:
            response = await self.client.get(path)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            raise ServiceError(f"Failed to fetch data from {path}: {str(e)}")

测试指南

手动测试

# 基本运行
python src/server.py

# 以 DEBUG 级别启动
LOG_LEVEL=DEBUG python src/server.py

集成测试

编写集成测试用例：

import pytest
from fastapi.testclient import TestClient

from server import app

client = TestClient(app)

def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert "status" in response.json()

资源

• FastAPI 官方文档
• Pydantic 官方文档
• HttpX 官方文档

最佳实践

代码风格

遵循 PEP8 和 Google Python 代码风格指南。

文档编写

为所有公共 API 编写详细的注释和使用说明。

版本控制

采用 Semantic Versioning (MAJOR.MINOR.PATCH) 规范进行版本管理。

许可证

MIT License

通过遵循此模板，你可以快速搭建一个稳定、可靠的 MCP 服务器，满足各种业务需求。