使用基于约定的自动模块发现机制,为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. 模块 A 模块 B 模块 C 模块 D │...
  15. └─────────┘ └─────────┘ └─────────┘ └─────────┘
  16. └─────────────────────────────────────────────────────────┘
  17. ┌─────────────────────────────────────────────────────────┐
  18. FastMCP适配器
  19. └─────────────────────────────────────────────────────────┘
  20. └────────────────────────────────────────────────────────────────┘
  21. ┌────────────────────────────────────────────────────────────────┐
  22. Claude Desktop
  23. └────────────────────────────────────────────────────────────────┘

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. """
  5. def __init__(self, config=None):
  6. """
  7. 初始化模块
  8. Args:
  9. config: 模块配置参数
  10. """
  11. self.config = config or {}
  12. def get_info(self):
  13. """
  14. 获取模块信息
  15. Returns:
  16. dict: 包含模块名称、版本、作者等信息的字典
  17. """
  18. raise NotImplementedError
  19. def register(self, server):
  20. """
  21. 向MCP服务器注册功能
  22. Args:
  23. server: FastMCP服务器实例
  24. """
  25. raise NotImplementedError

4.2 模块注册表 (Module Registry)

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

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

4.3 模块加载器 (Module Loader)

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

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

4.4 配置管理器 (Config Manager)

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

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

4.5 服务器适配器 (Server Adapter)

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

  1. class MCPServer:
  2. """
  3. MCP Server适配器,整合核心系统和FastMCP
  4. """
  5. def __init__(self, config_path='config.json', modules_dir='modules'):
  6. """
  7. 初始化MCP服务器
  8. Args:
  9. config_path: 配置文件路径
  10. modules_dir: 模块目录
  11. """
  12. self.config_manager = ConfigManager(config_path)
  13. self.registry = ModuleRegistry()
  14. self.module_loader = ModuleLoader(modules_dir, self.registry, self.config_manager)
  15. # 创建FastMCP服务器实例
  16. global_config = self.config_manager.get_global_config()
  17. server_name = global_config.get('server_name', 'MCP Server')
  18. self.server = FastMCP(server_name)
  19. def initialize(self):
  20. """
  21. 初始化服务器,加载模块
  22. """
  23. # 加载启用的模块
  24. modules = self.module_loader.load_enabled_modules()
  25. # 注册模块功能
  26. for module_name, module in self.registry.get_all_modules().items():
  27. module.register(self.server)
  28. return self
  29. def run(self, transport='stdio'):
  30. """
  31. 运行MCP服务器
  32. Args:
  33. transport: 传输协议,默认为stdio
  34. """
  35. 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
  2. from core.module_interface import ModuleInterface
  3. class HelloWorldModule(ModuleInterface):
  4. """
  5. Hello World示例模块
  6. 作者:kinbos 严富坤
  7. 邮箱:fookinbos@gmail.com
  8. 个人网站:htttps://www.yanfukun.com
  9. """
  10. def get_info(self):
  11. """
  12. 获取模块信息
  13. Returns:
  14. dict: 模块信息
  15. """
  16. return {
  17. "name": "hello_world",
  18. "version": "1.0.0",
  19. "description": "Hello World示例模块",
  20. "author": "kinbos 严富坤",
  21. "email": "fookinbos@gmail.com",
  22. "website": "htttps://www.yanfukun.com"
  23. }
  24. def register(self, server):
  25. """
  26. 向MCP服务器注册功能
  27. Args:
  28. server: FastMCP服务器实例
  29. """
  30. # 注册一个简单的工具
  31. @server.tool()
  32. def hello(name: str = "World") -> str:
  33. """
  34. 返回问候消息
  35. Args:
  36. name: 要问候的名称
  37. Returns:
  38. str: 问候消息
  39. """
  40. message = self.config.get('message', 'Hello, {}!')
  41. return message.format(name)
  42. # 注册一个资源
  43. @server.resource("hello://greeting")
  44. def get_greeting() -> str:
  45. """
  46. 获取问候消息
  47. Returns:
  48. str: 问候消息
  49. """
  50. return "Welcome to MCP Server!"

7. 主入口文件

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

8. 完整代码实现

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

8.12 主入口文件 (main.py)

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

    1. # 方法一: 使用FastMCP CLI
    2. fastmcp install main.py
    3. # 方法二: 直接运行
    4. 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应用场景。