使用基于约定的自动模块发现机制，为Claude Desktop打造的灵活MCP服务器框架。通过简洁的模块接口和声明式配置，轻松扩展AI应用功能，无需修改核心代码。支持FastMCP标准，包含完整工具、资源和提示模板注册能力。

特性

  • 🧩 模块化设计：功能以自包含模块组织，便于扩展
  • 🔍 自动发现：约定优于配置的模块加载方式
  • ⚙️ 声明式配置：通过config.json灵活配置模块和参数
  • 🔌 即插即用：新功能只需添加符合接口的模块目录
  • 🔒 安全稳定：分层架构确保核心系统稳定可靠
  • 📝 详细日志：完善的日志系统便于调试与监控
  • 🖥️ Claude Desktop集成：与Claude深度集成，提供AI增强体验

技术栈

Python FastMCP Model Context Protocol Claude Desktop JSON 模块化设计 微内核架构

1. 架构概述

基于对MCP（Model Context Protocol）和FastMCP的深入分析，我设计了一个模块化、可扩展的MCP Server框架。该框架旨在简化MCP服务器的开发和配置，使开发者能够轻松添加新功能并根据需要配置服务器。

1.1 设计理念

  • 模块化：功能以模块为单位组织，每个模块都是自包含的，具有明确的接口
  • 可发现性：自动发现和加载modules目录下的模块
  • 可配置性：通过config.json文件配置模块加载和全局参数
  • 扩展性：定义清晰的模块接口，方便开发者添加新功能
  • 兼容性：与Claude Desktop和FastMCP完全兼容

2. 架构设计图

  1. ┌─────────────────────────────────────────────────────────────────┐
  2. │                          MCP-BOS Server                             │
  3. ├─────────────────────────────────────────────────────────────────┤
  4. │                                                                 │
  5. │  ┌─────────────┐   ┌─────────────┐   ┌─────────────┐            │
  6. │  │   核心系统   │   │  模块注册表  │   │  配置管理器  │            │
  7. │  └─────────────┘   └─────────────┘   └─────────────┘            │
  8. │         │                 │                 │                    │
  9. │         └────────────────┬─────────────────┘                    │
  10. │                          │                                       │
  11. │  ┌─────────────────────────────────────────────────────────┐    │
  12. │  │                     模块加载器                            │    │
  13. │  └─────────────────────────────────────────────────────────┘    │
  14. │                          │                                       │
  15. │                          ▼                                       │
  16. │  ┌─────────────────────────────────────────────────────────┐    │
  17. │  │                       模块池                             │    │
  18. │  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐     │    │
  19. │  │  │ 模块 A   │  │ 模块 B   │  │ 模块 C   │  │ 模块 D   │...  │    │
  20. │  │  └─────────┘  └─────────┘  └─────────┘  └─────────┘     │    │
  21. │  └─────────────────────────────────────────────────────────┘    │
  22. │                          │                                       │
  23. │                          ▼                                       │
  24. │  ┌─────────────────────────────────────────────────────────┐    │
  25. │  │                    FastMCP适配器                         │    │
  26. │  └─────────────────────────────────────────────────────────┘    │
  27. │                          │                                       │
  28. └────────────────────────────────────────────────────────────────┘
  29.                            │
  30.                            ▼
  31. ┌────────────────────────────────────────────────────────────────┐
  32. │                      Claude Desktop                            │
  33. └────────────────────────────────────────────────────────────────┘

3. 目录结构

  1. mcp_server/
  2. ├── config.json             # 全局配置文件
  3. ├── main.py                 # 主入口文件
  4. ├── core/                   # 核心系统
  5. │   ├── __init__.py
  6. │   ├── module_registry.py  # 模块注册表
  7. │   ├── module_loader.py    # 模块加载器
  8. │   ├── config_manager.py   # 配置管理器
  9. │   └── server.py           # FastMCP服务器适配
  10. ├── modules/                # 功能模块目录
  11. │   ├── __init__.py
  12. │   ├── hello_world/        # Hello World示例模块
  13. │   │   ├── __init__.py
  14. │   │   └── hello.py
  15. │   ├── file_system/        # 文件系统模块示例
  16. │   │   ├── __init__.py
  17. │   │   └── fs.py
  18. │   └── ...                 # 其他功能模块
  19. ├── utils/                  # 工具函数
  20. │   ├── __init__.py
  21. │   └── helpers.py
  22. ├── tests/                  # 测试
  23. │   ├── __init__.py
  24. │   └── test_core.py
  25. └── README.md               # 项目说明文档

4. 核心组件设计

4.1 模块接口 (Module Interface)

所有功能模块都需要实现以下接口：

  1. class ModuleInterface:
  2.     """
  3.     模块接口定义，所有模块都需要实现这个接口
  4.     """
  6.     def __init__(self, config=None):
  7.         """
  8.         初始化模块
  10.         Args:
  11.             config: 模块配置参数
  12.         """
  13.         self.config = config or {}
  15.     def get_info(self):
  16.         """
  17.         获取模块信息
  19.         Returns:
  20.             dict: 包含模块名称、版本、作者等信息的字典
  21.         """
  22.         raise NotImplementedError
  24.     def register(self, server):
  25.         """
  26.         向MCP服务器注册功能
  28.         Args:
  29.             server: FastMCP服务器实例
  30.         """
  31.         raise NotImplementedError

4.2 模块注册表 (Module Registry)

负责跟踪所有已加载的模块及其状态：

  1. class ModuleRegistry:
  2.     """
  3.     模块注册表，负责跟踪所有已加载的模块
  4.     """
  6.     def __init__(self):
  7.         """
  8.         初始化模块注册表
  9.         """
  10.         self.modules = {}
  12.     def register(self, module_name, module_instance):
  13.         """
  14.         注册模块实例
  16.         Args:
  17.             module_name: 模块名称
  18.             module_instance: 模块实例
  19.         """
  20.         self.modules[module_name] = module_instance
  22.     def get_module(self, module_name):
  23.         """
  24.         获取模块实例
  26.         Args:
  27.             module_name: 模块名称
  29.         Returns:
  30.             ModuleInterface: 模块实例
  31.         """
  32.         return self.modules.get(module_name)
  34.     def get_all_modules(self):
  35.         """
  36.         获取所有已注册模块
  38.         Returns:
  39.             dict: 模块名称和实例的映射
  40.         """
  41.         return self.modules

4.3 模块加载器 (Module Loader)

负责自动发现和加载模块：

  1. class ModuleLoader:
  2.     """
  3.     模块加载器，负责自动发现和加载模块
  4.     """
  6.     def __init__(self, modules_dir, registry, config_manager):
  7.         """
  8.         初始化模块加载器
  10.         Args:
  11.             modules_dir: 模块目录
  12.             registry: 模块注册表实例
  13.             config_manager: 配置管理器实例
  14.         """
  15.         self.modules_dir = modules_dir
  16.         self.registry = registry
  17.         self.config_manager = config_manager
  19.     def discover_modules(self):
  20.         """
  21.         发现可用模块
  23.         Returns:
  24.             list: 可用模块列表
  25.         """
  26.         # 实现模块发现逻辑
  27.         pass
  29.     def load_module(self, module_name):
  30.         """
  31.         加载指定模块
  33.         Args:
  34.             module_name: 模块名称
  36.         Returns:
  37.             ModuleInterface: 加载的模块实例
  38.         """
  39.         # 实现模块加载逻辑
  40.         pass
  42.     def load_enabled_modules(self):
  43.         """
  44.         加载配置中启用的所有模块
  46.         Returns:
  47.             list: 已加载模块列表
  48.         """
  49.         # 根据配置加载启用的模块
  50.         pass

4.4 配置管理器 (Config Manager)

负责加载和管理全局配置：

  1. class ConfigManager:
  2.     """
  3.     配置管理器，负责加载和管理全局配置
  4.     """
  6.     def __init__(self, config_path):
  7.         """
  8.         初始化配置管理器
  10.         Args:
  11.             config_path: 配置文件路径
  12.         """
  13.         self.config_path = config_path
  14.         self.config = self._load_config()
  16.     def _load_config(self):
  17.         """
  18.         加载配置文件
  20.         Returns:
  21.             dict: 配置字典
  22.         """
  23.         # 实现配置加载逻辑
  24.         pass
  26.     def get_global_config(self):
  27.         """
  28.         获取全局配置
  30.         Returns:
  31.             dict: 全局配置字典
  32.         """
  33.         return self.config.get('global', {})
  35.     def get_modules_config(self):
  36.         """
  37.         获取模块配置
  39.         Returns:
  40.             dict: 模块配置字典
  41.         """
  42.         return self.config.get('modules', {})
  44.     def get_module_config(self, module_name):
  45.         """
  46.         获取指定模块的配置
  48.         Args:
  49.             module_name: 模块名称
  51.         Returns:
  52.             dict: 模块配置
  53.         """
  54.         modules_config = self.get_modules_config()
  55.         return modules_config.get(module_name, {})
  57.     def is_module_enabled(self, module_name):
  58.         """
  59.         检查模块是否启用
  61.         Args:
  62.             module_name: 模块名称
  64.         Returns:
  65.             bool: 模块是否启用
  66.         """
  67.         module_config = self.get_module_config(module_name)
  68.         return module_config.get('enabled', False)

4.5 服务器适配器 (Server Adapter)

将模块功能适配到FastMCP服务器：

  1. class MCPServer:
  2.     """
  3.     MCP Server适配器，整合核心系统和FastMCP
  4.     """
  6.     def __init__(self, config_path='config.json', modules_dir='modules'):
  7.         """
  8.         初始化MCP服务器
  10.         Args:
  11.             config_path: 配置文件路径
  12.             modules_dir: 模块目录
  13.         """
  14.         self.config_manager = ConfigManager(config_path)
  15.         self.registry = ModuleRegistry()
  16.         self.module_loader = ModuleLoader(modules_dir, self.registry, self.config_manager)
  18.         # 创建FastMCP服务器实例
  19.         global_config = self.config_manager.get_global_config()
  20.         server_name = global_config.get('server_name', 'MCP Server')
  21.         self.server = FastMCP(server_name)
  23.     def initialize(self):
  24.         """
  25.         初始化服务器，加载模块
  26.         """
  27.         # 加载启用的模块
  28.         modules = self.module_loader.load_enabled_modules()
  30.         # 注册模块功能
  31.         for module_name, module in self.registry.get_all_modules().items():
  32.             module.register(self.server)
  34.         return self
  36.     def run(self, transport='stdio'):
  37.         """
  38.         运行MCP服务器
  40.         Args:
  41.             transport: 传输协议，默认为stdio
  42.         """
  43.         self.server.run(transport=transport)

5. 配置文件结构

  1. {
  2.   "global": {
  3.     "server_name": "My MCP Server",
  4.     "debug": true,
  5.     "log_level": "INFO"
  6.   },
  7.   "modules": {
  8.     "hello_world": {
  9.       "enabled": true,
  10.       "message": "Hello from MCP Server!"
  11.     },
  12.     "file_system": {
  13.       "enabled": false,
  14.       "allowed_paths": ["/home/user/Documents", "/tmp"]
  15.     }
  16.   }
  17. }

6. 模块示例：Hello World模块

  1. # modules/hello_world/hello.py
  3. from core.module_interface import ModuleInterface
  5. class HelloWorldModule(ModuleInterface):
  6.     """
  7.     Hello World示例模块
  9.     作者：kinbos 严富坤
  10.     邮箱：fookinbos@gmail.com
  11.     个人网站：htttps://www.yanfukun.com
  12.     """
  14.     def get_info(self):
  15.         """
  16.         获取模块信息
  18.         Returns:
  19.             dict: 模块信息
  20.         """
  21.         return {
  22.             "name": "hello_world",
  23.             "version": "1.0.0",
  24.             "description": "Hello World示例模块",
  25.             "author": "kinbos 严富坤",
  26.             "email": "fookinbos@gmail.com",
  27.             "website": "htttps://www.yanfukun.com"
  28.         }
  30.     def register(self, server):
  31.         """
  32.         向MCP服务器注册功能
  34.         Args:
  35.             server: FastMCP服务器实例
  36.         """
  37.         # 注册一个简单的工具
  38.         @server.tool()
  39.         def hello(name: str = "World") -> str:
  40.             """
  41.             返回问候消息
  43.             Args:
  44.                 name: 要问候的名称
  46.             Returns:
  47.                 str: 问候消息
  48.             """
  49.             message = self.config.get('message', 'Hello, {}!')
  50.             return message.format(name)
  52.         # 注册一个资源
  53.         @server.resource("hello://greeting")
  54.         def get_greeting() -> str:
  55.             """
  56.             获取问候消息
  58.             Returns:
  59.                 str: 问候消息
  60.             """
  61.             return "Welcome to MCP Server!"

7. 主入口文件

  1. # main.py
  3. import os
  4. import sys
  5. from pathlib import Path
  7. # 添加当前目录到Python路径
  8. current_dir = Path(__file__).parent
  9. sys.path.append(str(current_dir))
  11. from core.server import MCPServer
  13. def main():
  14.     """
  15.     主入口函数
  17.     作者：kinbos 严富坤
  18.     邮箱：fookinbos@gmail.com
  19.     个人网站：htttps://www.yanfukun.com
  20.     """
  21.     # 创建并初始化MCP服务器
  22.     server = MCPServer().initialize()
  24.     # 运行服务器
  25.     server.run()
  27. if __name__ == "__main__":
  28.     main()

8. 完整代码实现

接下来，我将为每个核心文件提供完整的代码实现。

8.1 核心模块接口 (core/module_interface.py)

  1. """
  2. 模块接口定义
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. class ModuleInterface:
  10.     """
  11.     模块接口定义，所有模块都需要实现这个接口
  12.     """
  14.     def __init__(self, config=None):
  15.         """
  16.         初始化模块
  18.         Args:
  19.             config: 模块配置参数
  20.         """
  21.         self.config = config or {}
  23.     def get_info(self):
  24.         """
  25.         获取模块信息
  27.         Returns:
  28.             dict: 包含模块名称、版本、作者等信息的字典
  29.         """
  30.         raise NotImplementedError
  32.     def register(self, server):
  33.         """
  34.         向MCP服务器注册功能
  36.         Args:
  37.             server: FastMCP服务器实例
  38.         """
  39.         raise NotImplementedError

8.2 模块注册表 (core/module_registry.py)

  1. """
  2. 模块注册表，负责跟踪所有已加载的模块
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. class ModuleRegistry:
  10.     """
  11.     模块注册表，负责跟踪所有已加载的模块
  12.     """
  14.     def __init__(self):
  15.         """
  16.         初始化模块注册表
  17.         """
  18.         self.modules = {}
  20.     def register(self, module_name, module_instance):
  21.         """
  22.         注册模块实例
  24.         Args:
  25.             module_name: 模块名称
  26.             module_instance: 模块实例
  27.         """
  28.         self.modules[module_name] = module_instance
  29.         print(f"已注册模块: {module_name}")
  31.     def get_module(self, module_name):
  32.         """
  33.         获取模块实例
  35.         Args:
  36.             module_name: 模块名称
  38.         Returns:
  39.             ModuleInterface: 模块实例
  40.         """
  41.         return self.modules.get(module_name)
  43.     def get_all_modules(self):
  44.         """
  45.         获取所有已注册模块
  47.         Returns:
  48.             dict: 模块名称和实例的映射
  49.         """
  50.         return self.modules
  52.     def list_modules(self):
  53.         """
  54.         列出所有已注册模块
  56.         Returns:
  57.             list: 模块信息列表
  58.         """
  59.         return [
  60.             {
  61.                 "name": name,
  62.                 "info": module.get_info()
  63.             }
  64.             for name, module in self.modules.items()
  65.         ]

8.3 配置管理器 (core/config_manager.py)

  1. """
  2. 配置管理器，负责加载和管理全局配置
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. import json
  10. import os
  11. from pathlib import Path
  13. class ConfigManager:
  14.     """
  15.     配置管理器，负责加载和管理全局配置
  16.     """
  18.     def __init__(self, config_path):
  19.         """
  20.         初始化配置管理器
  22.         Args:
  23.             config_path: 配置文件路径
  24.         """
  25.         self.config_path = Path(config_path)
  26.         self.config = self._load_config()
  28.     def _load_config(self):
  29.         """
  30.         加载配置文件
  32.         Returns:
  33.             dict: 配置字典
  34.         """
  35.         if not self.config_path.exists():
  36.             print(f"警告: 配置文件 {self.config_path} 不存在，使用默认配置")
  37.             return self._default_config()
  39.         try:
  40.             with open(self.config_path, 'r', encoding='utf-8') as f:
  41.                 return json.load(f)
  42.         except Exception as e:
  43.             print(f"错误: 加载配置文件失败: {e}")
  44.             return self._default_config()
  46.     def _default_config(self):
  47.         """
  48.         返回默认配置
  50.         Returns:
  51.             dict: 默认配置字典
  52.         """
  53.         return {
  54.             "global": {
  55.                 "server_name": "MCP Server",
  56.                 "debug": False,
  57.                 "log_level": "INFO"
  58.             },
  59.             "modules": {
  60.                 "hello_world": {
  61.                     "enabled": True,
  62.                     "message": "Hello, {}!"
  63.                 }
  64.             }
  65.         }
  67.     def get_global_config(self):
  68.         """
  69.         获取全局配置
  71.         Returns:
  72.             dict: 全局配置字典
  73.         """
  74.         return self.config.get('global', {})
  76.     def get_modules_config(self):
  77.         """
  78.         获取模块配置
  80.         Returns:
  81.             dict: 模块配置字典
  82.         """
  83.         return self.config.get('modules', {})
  85.     def get_module_config(self, module_name):
  86.         """
  87.         获取指定模块的配置
  89.         Args:
  90.             module_name: 模块名称
  92.         Returns:
  93.             dict: 模块配置
  94.         """
  95.         modules_config = self.get_modules_config()
  96.         return modules_config.get(module_name, {})
  98.     def is_module_enabled(self, module_name):
  99.         """
  100.         检查模块是否启用
  102.         Args:
  103.             module_name: 模块名称
  105.         Returns:
  106.             bool: 模块是否启用
  107.         """
  108.         module_config = self.get_module_config(module_name)
  109.         return module_config.get('enabled', False)
  111.     def save_config(self):
  112.         """
  113.         保存配置到文件
  115.         Returns:
  116.             bool: 是否保存成功
  117.         """
  118.         try:
  119.             with open(self.config_path, 'w', encoding='utf-8') as f:
  120.                 json.dump(self.config, f, indent=2)
  121.             return True
  122.         except Exception as e:
  123.             print(f"错误: 保存配置文件失败: {e}")
  124.             return False

8.4 模块加载器 (core/module_loader.py)

  1. """
  2. 模块加载器，负责自动发现和加载模块
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. import importlib
  10. import importlib.util
  11. import inspect
  12. import os
  13. import sys
  14. from pathlib import Path
  16. from core.module_interface import ModuleInterface
  18. class ModuleLoader:
  19.     """
  20.     模块加载器，负责自动发现和加载模块
  21.     """
  23.     def __init__(self, modules_dir, registry, config_manager):
  24.         """
  25.         初始化模块加载器
  27.         Args:
  28.             modules_dir: 模块目录
  29.             registry: 模块注册表实例
  30.             config_manager: 配置管理器实例
  31.         """
  32.         self.modules_dir = Path(modules_dir)
  33.         self.registry = registry
  34.         self.config_manager = config_manager
  36.         # 确保模块目录存在
  37.         if not self.modules_dir.exists():
  38.             raise FileNotFoundError(f"模块目录不存在: {modules_dir}")
  40.         # 添加模块目录到Python路径
  41.         sys.path.append(str(self.modules_dir))
  43.     def discover_modules(self):
  44.         """
  45.         发现可用模块
  47.         Returns:
  48.             list: 可用模块名称列表
  49.         """
  50.         modules = []
  52.         # 遍历模块目录
  53.         for item in self.modules_dir.iterdir():
  54.             if not item.is_dir() or item.name.startswith('_') or item.name.startswith('.'):
  55.                 continue
  57.             # 检查是否是有效的Python包
  58.             init_file = item / "__init__.py"
  59.             if not init_file.exists():
  60.                 continue
  62.             modules.append(item.name)
  64.         return modules
  66.     def _find_module_class(self, module_name):
  67.         """
  68.         在模块中查找实现ModuleInterface的类
  70.         Args:
  71.             module_name: 模块名称
  73.         Returns:
  74.             type: 模块类
  75.         """
  76.         # 尝试导入模块
  77.         try:
  78.             # 先尝试直接导入主模块
  79.             module = importlib.import_module(f"modules.{module_name}")
  80.         except ImportError:
  81.             try:
  82.                 # 尝试导入与模块同名的子模块
  83.                 module = importlib.import_module(f"modules.{module_name}.{module_name}")
  84.             except ImportError:
  85.                 print(f"错误: 无法导入模块 {module_name}")
  86.                 return None
  88.         # 查找实现ModuleInterface的类
  89.         for name, obj in inspect.getmembers(module):
  90.             if (inspect.isclass(obj) and 
  91.                 issubclass(obj, ModuleInterface) and 
  92.                 obj is not ModuleInterface):
  93.                 return obj
  95.         # 查找所有子模块
  96.         for item in (self.modules_dir / module_name).iterdir():
  97.             if not item.is_file() or not item.name.endswith('.py') or item.name.startswith('_'):
  98.                 continue
  100.             sub_module_name = item.stem
  101.             try:
  102.                 sub_module = importlib.import_module(f"modules.{module_name}.{sub_module_name}")
  104.                 for name, obj in inspect.getmembers(sub_module):
  105.                     if (inspect.isclass(obj) and 
  106.                         issubclass(obj, ModuleInterface) and 
  107.                         obj is not ModuleInterface):
  108.                         return obj
  109.             except ImportError:
  110.                 continue
  112.         return None
  114.     def load_module(self, module_name):
  115.         """
  116.         加载指定模块
  118.         Args:
  119.             module_name: 模块名称
  121.         Returns:
  122.             ModuleInterface: 加载的模块实例
  123.         """
  124.         # 获取模块配置
  125.         module_config = self.config_manager.get_module_config(module_name)
  127.         # 查找模块类
  128.         module_class = self._find_module_class(module_name)
  129.         if not module_class:
  130.             print(f"错误: 模块 {module_name} 中未找到有效的模块类")
  131.             return None
  133.         # 实例化模块类
  134.         try:
  135.             module_instance = module_class(module_config)
  136.             self.registry.register(module_name, module_instance)
  137.             return module_instance
  138.         except Exception as e:
  139.             print(f"错误: 加载模块 {module_name} 失败: {e}")
  140.             return None
  142.     def load_enabled_modules(self):
  143.         """
  144.         加载配置中启用的所有模块
  146.         Returns:
  147.             list: 已加载模块列表
  148.         """
  149.         loaded_modules = []
  151.         # 发现可用模块
  152.         available_modules = self.discover_modules()
  154.         # 加载启用的模块
  155.         for module_name in available_modules:
  156.             if self.config_manager.is_module_enabled(module_name):
  157.                 module = self.load_module(module_name)
  158.                 if module:
  159.                     loaded_modules.append(module)
  161.         return loaded_modules

8.5 服务器适配器 (core/server.py)

  1. """
  2. MCP Server适配器，整合核心系统和FastMCP
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. import os
  10. from pathlib import Path
  12. from fastmcp import FastMCP, Context
  14. from core.config_manager import ConfigManager
  15. from core.module_loader import ModuleLoader
  16. from core.module_registry import ModuleRegistry
  18. class MCPServer:
  19.     """
  20.     MCP Server适配器，整合核心系统和FastMCP
  21.     """
  23.     def __init__(self, config_path='config.json', modules_dir='modules'):
  24.         """
  25.         初始化MCP服务器
  27.         Args:
  28.             config_path: 配置文件路径
  29.             modules_dir: 模块目录
  30.         """
  31.         # 初始化核心组件
  32.         self.config_manager = ConfigManager(config_path)
  33.         self.registry = ModuleRegistry()
  34.         self.module_loader = ModuleLoader(modules_dir, self.registry, self.config_manager)
  36.         # 创建FastMCP服务器实例
  37.         global_config = self.config_manager.get_global_config()
  38.         server_name = global_config.get('server_name', 'MCP Server')
  40.         # 获取配置的依赖
  41.         dependencies = global_config.get('dependencies', [])
  43.         # 创建FastMCP实例
  44.         self.server = FastMCP(
  45.             server_name,
  46.             dependencies=dependencies,
  47.             log_level=global_config.get('log_level', 'INFO'),
  48.             debug=global_config.get('debug', False)
  49.         )
  51.         # 注册核心工具
  52.         self._register_core_tools()
  54.     def _register_core_tools(self):
  55.         """
  56.         注册核心工具
  57.         """
  58.         # 注册服务器信息工具
  59.         @self.server.tool()
  60.         def server_info(ctx: Context) -> dict:
  61.             """
  62.             获取服务器信息
  64.             Args:
  65.                 ctx: MCP上下文
  67.             Returns:
  68.                 dict: 服务器信息
  69.             """
  70.             global_config = self.config_manager.get_global_config()
  72.             return {
  73.                 "name": global_config.get('server_name', 'MCP Server'),
  74.                 "version": "1.0.0",
  75.                 "modules": self.registry.list_modules()
  76.             }
  78.     def initialize(self):
  79.         """
  80.         初始化服务器，加载模块
  82.         Returns:
  83.             self: 服务器实例，用于链式调用
  84.         """
  85.         # 加载启用的模块
  86.         modules = self.module_loader.load_enabled_modules()
  88.         # 注册模块功能
  89.         for module_name, module in self.registry.get_all_modules().items():
  90.             try:
  91.                 module.register(self.server)
  92.                 print(f"已注册模块功能: {module_name}")
  93.             except Exception as e:
  94.                 print(f"错误: 注册模块 {module_name} 功能失败: {e}")
  96.         return self
  98.     def run(self, transport='stdio'):
  99.         """
  100.         运行MCP服务器
  102.         Args:
  103.             transport: 传输协议，默认为stdio
  104.         """
  105.         global_config = self.config_manager.get_global_config()
  106.         transport = global_config.get('transport', transport)
  108.         # 运行服务器
  109.         print(f"启动MCP服务器，传输协议: {transport}")
  110.         self.server.run(transport=transport)

8.6 核心包初始化 (core/init.py)

  1. """
  2. MCP Server核心包
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. from core.module_interface import ModuleInterface
  10. from core.module_registry import ModuleRegistry
  11. from core.module_loader import ModuleLoader
  12. from core.config_manager import ConfigManager
  13. from core.server import MCPServer
  15. # 导出核心类
  16. __all__ = [
  17.     'ModuleInterface',
  18.     'ModuleRegistry',
  19.     'ModuleLoader',
  20.     'ConfigManager',
  21.     'MCPServer',
  22. ]

8.7 模块包初始化 (modules/init.py)

  1. """
  2. MCP Server模块包
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. # 这个文件使modules目录成为一个Python包
  10. # 模块发现和加载由模块加载器自动完成

8.8 Hello World模块 (modules/hello_world/hello.py)

  1. """
  2. Hello World示例模块
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. from core.module_interface import ModuleInterface
  11. class HelloWorldModule(ModuleInterface):
  12.     """
  13.     Hello World示例模块，展示了一个简单的MCP模块实现
  14.     """
  16.     def get_info(self):
  17.         """
  18.         获取模块信息
  20.         Returns:
  21.             dict: 模块信息
  22.         """
  23.         return {
  24.             "name": "hello_world",
  25.             "version": "1.0.0",
  26.             "description": "Hello World示例模块",
  27.             "author": "kinbos 严富坤",
  28.             "email": "fookinbos@gmail.com",
  29.             "website": "htttps://www.yanfukun.com"
  30.         }
  32.     def register(self, server):
  33.         """
  34.         向MCP服务器注册功能
  36.         Args:
  37.             server: FastMCP服务器实例
  38.         """
  39.         # 注册一个简单的工具
  40.         @server.tool()
  41.         def hello(name: str = "World") -> str:
  42.             """
  43.             返回问候消息
  45.             Args:
  46.                 name: 要问候的名称
  48.             Returns:
  49.                 str: 问候消息
  50.             """
  51.             message = self.config.get('message', 'Hello, {}!')
  52.             return message.format(name)
  54.         # 注册一个资源
  55.         @server.resource("hello://greeting")
  56.         def get_greeting() -> str:
  57.             """
  58.             获取问候消息
  60.             Returns:
  61.                 str: 问候消息
  62.             """
  63.             return "Welcome to MCP Server!"
  65.         # 注册一个提示模板
  66.         @server.prompt()
  67.         def hello_prompt(name: str = "User") -> str:
  68.             """
  69.             创建一个问候提示
  71.             Args:
  72.                 name: 要问候的名称
  74.             Returns:
  75.                 str: 问候提示
  76.             """
  77.             return f"Welcome, {name}! How can I assist you today?"

8.9 Hello World模块初始化 (modules/helloworld/_init.py)

  1. """
  2. Hello World模块初始化
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. from modules.hello_world.hello import HelloWorldModule
  11. # 导出模块类
  12. __all__ = ['HelloWorldModule']

8.10 工具函数 (utils/helpers.py)

  1. """
  2. 工具函数
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. import importlib
  10. import os
  11. import sys
  12. from pathlib import Path
  14. def ensure_directory(path):
  15.     """
  16.     确保目录存在，如果不存在则创建
  18.     Args:
  19.         path: 目录路径
  21.     Returns:
  22.         Path: 目录路径对象
  23.     """
  24.     path = Path(path)
  25.     path.mkdir(parents=True, exist_ok=True)
  26.     return path
  28. def import_module_from_path(module_name, module_path):
  29.     """
  30.     从路径导入模块
  32.     Args:
  33.         module_name: 模块名称
  34.         module_path: 模块路径
  36.     Returns:
  37.         module: 导入的模块
  38.     """
  39.     spec = importlib.util.spec_from_file_location(module_name, module_path)
  40.     if spec is None:
  41.         return None
  43.     module = importlib.util.module_from_spec(spec)
  44.     sys.modules[module_name] = module
  45.     spec.loader.exec_module(module)
  46.     return module
  48. def format_error(e):
  49.     """
  50.     格式化异常信息
  52.     Args:
  53.         e: 异常对象
  55.     Returns:
  56.         str: 格式化后的异常信息
  57.     """
  58.     return f"{type(e).__name__}: {str(e)}"

8.11 工具包初始化 (utils/init.py)

  1. """
  2. 工具函数包
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. from utils.helpers import ensure_directory, import_module_from_path, format_error
  11. # 导出工具函数
  12. __all__ = [
  13.     'ensure_directory',
  14.     'import_module_from_path',
  15.     'format_error',
  16. ]

8.12 主入口文件 (main.py)

  1. """
  2. MCP Server主入口文件
  4. 作者：kinbos 严富坤
  5. 邮箱：fookinbos@gmail.com
  6. 个人网站：htttps://www.yanfukun.com
  7. """
  9. import os
  10. import sys
  11. from pathlib import Path
  13. # 添加当前目录到Python路径
  14. current_dir = Path(__file__).parent
  15. sys.path.append(str(current_dir))
  17. # 导入核心服务器类
  18. from core.server import MCPServer
  20. def main():
  21.     """
  22.     主入口函数
  23.     """
  24.     # 默认配置文件路径
  25.     config_path = current_dir / 'config.json'
  27.     # 默认模块目录
  28.     modules_dir = current_dir / 'modules'
  30.     # 创建并初始化MCP服务器
  31.     try:
  32.         server = MCPServer(
  33.             config_path=config_path,
  34.             modules_dir=modules_dir
  35.         ).initialize()
  37.         # 运行服务器
  38.         server.run()
  39.     except Exception as e:
  40.         print(f"错误: 启动MCP服务器失败: {e}")
  41.         sys.exit(1)
  43. if __name__ == "__main__":
  44.     main()

8.13 默认配置文件 (config.json)

  1. {
  2.   "global": {
  3.     "server_name": "MCP Server",
  4.     "debug": true,
  5.     "log_level": "INFO",
  6.     "transport": "stdio",
  7.     "dependencies": []
  8.   },
  9.   "modules": {
  10.     "hello_world": {
  11.       "enabled": true,
  12.       "message": "Hello, {}!"
  13.     }
  14.   }
  15. }

9. 配置和部署说明

9.1 配置MCP Server到Claude Desktop

要将框架配置到Claude Desktop中使用，可以使用以下步骤:

  1. 确保已安装Claude Desktop和uv:

    1. # 安装uv (如果未安装)
    2. brew install uv  # macOS
    3. # 或者
    4. pip install uv   # 其他平台

  2. 安装框架:

    1. # 克隆仓库
    2. git clone https://github.com/kinbos/mcp-server.git
    3. cd mcp-server
    5. # 安装依赖
    6. uv pip install fastmcp

  3. 将框架安装到Claude Desktop:

    1. # 方法一: 使用FastMCP CLI
    2. fastmcp install main.py
    4. # 方法二: 直接运行
    5. uv run main.py

  4. 或者手动编辑Claude Desktop配置文件:

    1. {
    2.   "mcpServers": {
    3.     "mcp_server": {
    4.       "command": "uv",
    5.       "args": ["run", "/path/to/mcp-server/main.py"]
    6.     }
    7.   }
    8. }

9.2 扩展框架

要扩展框架功能，只需按照以下步骤添加新模块:

  1. 在modules目录下创建新的模块目录
  2. 创建__init__.py文件和模块实现文件
  3. 实现ModuleInterface接口
  4. 在config.json中启用模块

框架会自动发现并加载新模块。

10. 测试和验证

要测试框架是否正常工作，可以使用:

  1. 启动Claude Desktop并连接到MCP Server
  2. 在Claude对话中测试Hello World模块功能:
    • 输入: “使用hello工具向John问好”
    • 预期输出: “Hello, John!”
    • 输入: “获取问候资源”
    • 预期输出: 访问”hello://greeting”资源，显示”Welcome to MCP Server!”
    • 输入: “列出服务器信息”
    • 预期输出: 显示服务器名称、版本和已加载模块列表

这个MCP Server框架提供了一个灵活、可扩展的基础，用于构建与Claude Desktop集成的MCP服务器。框架的模块化设计使得开发者可以轻松添加新功能，而无需修改核心代码。通过自动发现和配置驱动的方法，框架实现了高度的可配置性和可扩展性，适用于各种MCP应用场景。