Template MCP Server

🚀 Python MCP 服务器模板

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

🚀 快速开始

项目结构

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

✨ 主要特性

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

📦 安装指南

安装依赖的命令如下：

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 服务器，满足各种业务需求。