- 特性
- 技术栈
- 1. 架构概述
- 2. 架构设计图
- 3. 目录结构
- 4. 核心组件设计
- 5. 配置文件结构
- 6. 模块示例:Hello World模块
- 7. 主入口文件
- 8. 完整代码实现
- 8.1 核心模块接口 (core/module_interface.py)
- 8.2 模块注册表 (core/module_registry.py)
- 8.3 配置管理器 (core/config_manager.py)
- 8.4 模块加载器 (core/module_loader.py)
- 8.5 服务器适配器 (core/server.py)
- 8.6 核心包初始化 (core/init.py)
- 8.7 模块包初始化 (modules/init.py)
- 8.8 Hello World模块 (modules/hello_world/hello.py)
- 8.9 Hello World模块初始化 (modules/helloworld/_init.py)
- 8.10 工具函数 (utils/helpers.py)
- 8.11 工具包初始化 (utils/init.py)
- 8.12 主入口文件 (main.py)
- 8.13 默认配置文件 (config.json)
- 9. 配置和部署说明
- 10. 测试和验证
使用基于约定的自动模块发现机制,为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. 架构设计图
┌─────────────────────────────────────────────────────────────────┐
│ MCP-BOS Server │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 核心系统 │ │ 模块注册表 │ │ 配置管理器 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │ │
│ └────────────────┬─────────────────┘ │
│ │ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 模块加载器 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 模块池 │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ 模块 A │ │ 模块 B │ │ 模块 C │ │ 模块 D │... │ │
│ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ FastMCP适配器 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
└────────────────────────────────────────────────────────────────┘
│
▼
┌────────────────────────────────────────────────────────────────┐
│ Claude Desktop │
└────────────────────────────────────────────────────────────────┘
3. 目录结构
mcp_server/
├── config.json # 全局配置文件
├── main.py # 主入口文件
├── core/ # 核心系统
│ ├── __init__.py
│ ├── module_registry.py # 模块注册表
│ ├── module_loader.py # 模块加载器
│ ├── config_manager.py # 配置管理器
│ └── server.py # FastMCP服务器适配
├── modules/ # 功能模块目录
│ ├── __init__.py
│ ├── hello_world/ # Hello World示例模块
│ │ ├── __init__.py
│ │ └── hello.py
│ ├── file_system/ # 文件系统模块示例
│ │ ├── __init__.py
│ │ └── fs.py
│ └── ... # 其他功能模块
├── utils/ # 工具函数
│ ├── __init__.py
│ └── helpers.py
├── tests/ # 测试
│ ├── __init__.py
│ └── test_core.py
└── README.md # 项目说明文档
4. 核心组件设计
4.1 模块接口 (Module Interface)
所有功能模块都需要实现以下接口:
class ModuleInterface:
"""
模块接口定义,所有模块都需要实现这个接口
"""
def __init__(self, config=None):
"""
初始化模块
Args:
config: 模块配置参数
"""
self.config = config or {}
def get_info(self):
"""
获取模块信息
Returns:
dict: 包含模块名称、版本、作者等信息的字典
"""
raise NotImplementedError
def register(self, server):
"""
向MCP服务器注册功能
Args:
server: FastMCP服务器实例
"""
raise NotImplementedError
4.2 模块注册表 (Module Registry)
负责跟踪所有已加载的模块及其状态:
class ModuleRegistry:
"""
模块注册表,负责跟踪所有已加载的模块
"""
def __init__(self):
"""
初始化模块注册表
"""
self.modules = {}
def register(self, module_name, module_instance):
"""
注册模块实例
Args:
module_name: 模块名称
module_instance: 模块实例
"""
self.modules[module_name] = module_instance
def get_module(self, module_name):
"""
获取模块实例
Args:
module_name: 模块名称
Returns:
ModuleInterface: 模块实例
"""
return self.modules.get(module_name)
def get_all_modules(self):
"""
获取所有已注册模块
Returns:
dict: 模块名称和实例的映射
"""
return self.modules
4.3 模块加载器 (Module Loader)
负责自动发现和加载模块:
class ModuleLoader:
"""
模块加载器,负责自动发现和加载模块
"""
def __init__(self, modules_dir, registry, config_manager):
"""
初始化模块加载器
Args:
modules_dir: 模块目录
registry: 模块注册表实例
config_manager: 配置管理器实例
"""
self.modules_dir = modules_dir
self.registry = registry
self.config_manager = config_manager
def discover_modules(self):
"""
发现可用模块
Returns:
list: 可用模块列表
"""
# 实现模块发现逻辑
pass
def load_module(self, module_name):
"""
加载指定模块
Args:
module_name: 模块名称
Returns:
ModuleInterface: 加载的模块实例
"""
# 实现模块加载逻辑
pass
def load_enabled_modules(self):
"""
加载配置中启用的所有模块
Returns:
list: 已加载模块列表
"""
# 根据配置加载启用的模块
pass
4.4 配置管理器 (Config Manager)
负责加载和管理全局配置:
class ConfigManager:
"""
配置管理器,负责加载和管理全局配置
"""
def __init__(self, config_path):
"""
初始化配置管理器
Args:
config_path: 配置文件路径
"""
self.config_path = config_path
self.config = self._load_config()
def _load_config(self):
"""
加载配置文件
Returns:
dict: 配置字典
"""
# 实现配置加载逻辑
pass
def get_global_config(self):
"""
获取全局配置
Returns:
dict: 全局配置字典
"""
return self.config.get('global', {})
def get_modules_config(self):
"""
获取模块配置
Returns:
dict: 模块配置字典
"""
return self.config.get('modules', {})
def get_module_config(self, module_name):
"""
获取指定模块的配置
Args:
module_name: 模块名称
Returns:
dict: 模块配置
"""
modules_config = self.get_modules_config()
return modules_config.get(module_name, {})
def is_module_enabled(self, module_name):
"""
检查模块是否启用
Args:
module_name: 模块名称
Returns:
bool: 模块是否启用
"""
module_config = self.get_module_config(module_name)
return module_config.get('enabled', False)
4.5 服务器适配器 (Server Adapter)
将模块功能适配到FastMCP服务器:
class MCPServer:
"""
MCP Server适配器,整合核心系统和FastMCP
"""
def __init__(self, config_path='config.json', modules_dir='modules'):
"""
初始化MCP服务器
Args:
config_path: 配置文件路径
modules_dir: 模块目录
"""
self.config_manager = ConfigManager(config_path)
self.registry = ModuleRegistry()
self.module_loader = ModuleLoader(modules_dir, self.registry, self.config_manager)
# 创建FastMCP服务器实例
global_config = self.config_manager.get_global_config()
server_name = global_config.get('server_name', 'MCP Server')
self.server = FastMCP(server_name)
def initialize(self):
"""
初始化服务器,加载模块
"""
# 加载启用的模块
modules = self.module_loader.load_enabled_modules()
# 注册模块功能
for module_name, module in self.registry.get_all_modules().items():
module.register(self.server)
return self
def run(self, transport='stdio'):
"""
运行MCP服务器
Args:
transport: 传输协议,默认为stdio
"""
self.server.run(transport=transport)
5. 配置文件结构
{
"global": {
"server_name": "My MCP Server",
"debug": true,
"log_level": "INFO"
},
"modules": {
"hello_world": {
"enabled": true,
"message": "Hello from MCP Server!"
},
"file_system": {
"enabled": false,
"allowed_paths": ["/home/user/Documents", "/tmp"]
}
}
}
6. 模块示例:Hello World模块
# modules/hello_world/hello.py
from core.module_interface import ModuleInterface
class HelloWorldModule(ModuleInterface):
"""
Hello World示例模块
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
def get_info(self):
"""
获取模块信息
Returns:
dict: 模块信息
"""
return {
"name": "hello_world",
"version": "1.0.0",
"description": "Hello World示例模块",
"author": "kinbos 严富坤",
"email": "fookinbos@gmail.com",
"website": "htttps://www.yanfukun.com"
}
def register(self, server):
"""
向MCP服务器注册功能
Args:
server: FastMCP服务器实例
"""
# 注册一个简单的工具
@server.tool()
def hello(name: str = "World") -> str:
"""
返回问候消息
Args:
name: 要问候的名称
Returns:
str: 问候消息
"""
message = self.config.get('message', 'Hello, {}!')
return message.format(name)
# 注册一个资源
@server.resource("hello://greeting")
def get_greeting() -> str:
"""
获取问候消息
Returns:
str: 问候消息
"""
return "Welcome to MCP Server!"
7. 主入口文件
# main.py
import os
import sys
from pathlib import Path
# 添加当前目录到Python路径
current_dir = Path(__file__).parent
sys.path.append(str(current_dir))
from core.server import MCPServer
def main():
"""
主入口函数
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
# 创建并初始化MCP服务器
server = MCPServer().initialize()
# 运行服务器
server.run()
if __name__ == "__main__":
main()
8. 完整代码实现
接下来,我将为每个核心文件提供完整的代码实现。
8.1 核心模块接口 (core/module_interface.py)
"""
模块接口定义
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
class ModuleInterface:
"""
模块接口定义,所有模块都需要实现这个接口
"""
def __init__(self, config=None):
"""
初始化模块
Args:
config: 模块配置参数
"""
self.config = config or {}
def get_info(self):
"""
获取模块信息
Returns:
dict: 包含模块名称、版本、作者等信息的字典
"""
raise NotImplementedError
def register(self, server):
"""
向MCP服务器注册功能
Args:
server: FastMCP服务器实例
"""
raise NotImplementedError
8.2 模块注册表 (core/module_registry.py)
"""
模块注册表,负责跟踪所有已加载的模块
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
class ModuleRegistry:
"""
模块注册表,负责跟踪所有已加载的模块
"""
def __init__(self):
"""
初始化模块注册表
"""
self.modules = {}
def register(self, module_name, module_instance):
"""
注册模块实例
Args:
module_name: 模块名称
module_instance: 模块实例
"""
self.modules[module_name] = module_instance
print(f"已注册模块: {module_name}")
def get_module(self, module_name):
"""
获取模块实例
Args:
module_name: 模块名称
Returns:
ModuleInterface: 模块实例
"""
return self.modules.get(module_name)
def get_all_modules(self):
"""
获取所有已注册模块
Returns:
dict: 模块名称和实例的映射
"""
return self.modules
def list_modules(self):
"""
列出所有已注册模块
Returns:
list: 模块信息列表
"""
return [
{
"name": name,
"info": module.get_info()
}
for name, module in self.modules.items()
]
8.3 配置管理器 (core/config_manager.py)
"""
配置管理器,负责加载和管理全局配置
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
import json
import os
from pathlib import Path
class ConfigManager:
"""
配置管理器,负责加载和管理全局配置
"""
def __init__(self, config_path):
"""
初始化配置管理器
Args:
config_path: 配置文件路径
"""
self.config_path = Path(config_path)
self.config = self._load_config()
def _load_config(self):
"""
加载配置文件
Returns:
dict: 配置字典
"""
if not self.config_path.exists():
print(f"警告: 配置文件 {self.config_path} 不存在,使用默认配置")
return self._default_config()
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"错误: 加载配置文件失败: {e}")
return self._default_config()
def _default_config(self):
"""
返回默认配置
Returns:
dict: 默认配置字典
"""
return {
"global": {
"server_name": "MCP Server",
"debug": False,
"log_level": "INFO"
},
"modules": {
"hello_world": {
"enabled": True,
"message": "Hello, {}!"
}
}
}
def get_global_config(self):
"""
获取全局配置
Returns:
dict: 全局配置字典
"""
return self.config.get('global', {})
def get_modules_config(self):
"""
获取模块配置
Returns:
dict: 模块配置字典
"""
return self.config.get('modules', {})
def get_module_config(self, module_name):
"""
获取指定模块的配置
Args:
module_name: 模块名称
Returns:
dict: 模块配置
"""
modules_config = self.get_modules_config()
return modules_config.get(module_name, {})
def is_module_enabled(self, module_name):
"""
检查模块是否启用
Args:
module_name: 模块名称
Returns:
bool: 模块是否启用
"""
module_config = self.get_module_config(module_name)
return module_config.get('enabled', False)
def save_config(self):
"""
保存配置到文件
Returns:
bool: 是否保存成功
"""
try:
with open(self.config_path, 'w', encoding='utf-8') as f:
json.dump(self.config, f, indent=2)
return True
except Exception as e:
print(f"错误: 保存配置文件失败: {e}")
return False
8.4 模块加载器 (core/module_loader.py)
"""
模块加载器,负责自动发现和加载模块
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
import importlib
import importlib.util
import inspect
import os
import sys
from pathlib import Path
from core.module_interface import ModuleInterface
class ModuleLoader:
"""
模块加载器,负责自动发现和加载模块
"""
def __init__(self, modules_dir, registry, config_manager):
"""
初始化模块加载器
Args:
modules_dir: 模块目录
registry: 模块注册表实例
config_manager: 配置管理器实例
"""
self.modules_dir = Path(modules_dir)
self.registry = registry
self.config_manager = config_manager
# 确保模块目录存在
if not self.modules_dir.exists():
raise FileNotFoundError(f"模块目录不存在: {modules_dir}")
# 添加模块目录到Python路径
sys.path.append(str(self.modules_dir))
def discover_modules(self):
"""
发现可用模块
Returns:
list: 可用模块名称列表
"""
modules = []
# 遍历模块目录
for item in self.modules_dir.iterdir():
if not item.is_dir() or item.name.startswith('_') or item.name.startswith('.'):
continue
# 检查是否是有效的Python包
init_file = item / "__init__.py"
if not init_file.exists():
continue
modules.append(item.name)
return modules
def _find_module_class(self, module_name):
"""
在模块中查找实现ModuleInterface的类
Args:
module_name: 模块名称
Returns:
type: 模块类
"""
# 尝试导入模块
try:
# 先尝试直接导入主模块
module = importlib.import_module(f"modules.{module_name}")
except ImportError:
try:
# 尝试导入与模块同名的子模块
module = importlib.import_module(f"modules.{module_name}.{module_name}")
except ImportError:
print(f"错误: 无法导入模块 {module_name}")
return None
# 查找实现ModuleInterface的类
for name, obj in inspect.getmembers(module):
if (inspect.isclass(obj) and
issubclass(obj, ModuleInterface) and
obj is not ModuleInterface):
return obj
# 查找所有子模块
for item in (self.modules_dir / module_name).iterdir():
if not item.is_file() or not item.name.endswith('.py') or item.name.startswith('_'):
continue
sub_module_name = item.stem
try:
sub_module = importlib.import_module(f"modules.{module_name}.{sub_module_name}")
for name, obj in inspect.getmembers(sub_module):
if (inspect.isclass(obj) and
issubclass(obj, ModuleInterface) and
obj is not ModuleInterface):
return obj
except ImportError:
continue
return None
def load_module(self, module_name):
"""
加载指定模块
Args:
module_name: 模块名称
Returns:
ModuleInterface: 加载的模块实例
"""
# 获取模块配置
module_config = self.config_manager.get_module_config(module_name)
# 查找模块类
module_class = self._find_module_class(module_name)
if not module_class:
print(f"错误: 模块 {module_name} 中未找到有效的模块类")
return None
# 实例化模块类
try:
module_instance = module_class(module_config)
self.registry.register(module_name, module_instance)
return module_instance
except Exception as e:
print(f"错误: 加载模块 {module_name} 失败: {e}")
return None
def load_enabled_modules(self):
"""
加载配置中启用的所有模块
Returns:
list: 已加载模块列表
"""
loaded_modules = []
# 发现可用模块
available_modules = self.discover_modules()
# 加载启用的模块
for module_name in available_modules:
if self.config_manager.is_module_enabled(module_name):
module = self.load_module(module_name)
if module:
loaded_modules.append(module)
return loaded_modules
8.5 服务器适配器 (core/server.py)
"""
MCP Server适配器,整合核心系统和FastMCP
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
import os
from pathlib import Path
from fastmcp import FastMCP, Context
from core.config_manager import ConfigManager
from core.module_loader import ModuleLoader
from core.module_registry import ModuleRegistry
class MCPServer:
"""
MCP Server适配器,整合核心系统和FastMCP
"""
def __init__(self, config_path='config.json', modules_dir='modules'):
"""
初始化MCP服务器
Args:
config_path: 配置文件路径
modules_dir: 模块目录
"""
# 初始化核心组件
self.config_manager = ConfigManager(config_path)
self.registry = ModuleRegistry()
self.module_loader = ModuleLoader(modules_dir, self.registry, self.config_manager)
# 创建FastMCP服务器实例
global_config = self.config_manager.get_global_config()
server_name = global_config.get('server_name', 'MCP Server')
# 获取配置的依赖
dependencies = global_config.get('dependencies', [])
# 创建FastMCP实例
self.server = FastMCP(
server_name,
dependencies=dependencies,
log_level=global_config.get('log_level', 'INFO'),
debug=global_config.get('debug', False)
)
# 注册核心工具
self._register_core_tools()
def _register_core_tools(self):
"""
注册核心工具
"""
# 注册服务器信息工具
@self.server.tool()
def server_info(ctx: Context) -> dict:
"""
获取服务器信息
Args:
ctx: MCP上下文
Returns:
dict: 服务器信息
"""
global_config = self.config_manager.get_global_config()
return {
"name": global_config.get('server_name', 'MCP Server'),
"version": "1.0.0",
"modules": self.registry.list_modules()
}
def initialize(self):
"""
初始化服务器,加载模块
Returns:
self: 服务器实例,用于链式调用
"""
# 加载启用的模块
modules = self.module_loader.load_enabled_modules()
# 注册模块功能
for module_name, module in self.registry.get_all_modules().items():
try:
module.register(self.server)
print(f"已注册模块功能: {module_name}")
except Exception as e:
print(f"错误: 注册模块 {module_name} 功能失败: {e}")
return self
def run(self, transport='stdio'):
"""
运行MCP服务器
Args:
transport: 传输协议,默认为stdio
"""
global_config = self.config_manager.get_global_config()
transport = global_config.get('transport', transport)
# 运行服务器
print(f"启动MCP服务器,传输协议: {transport}")
self.server.run(transport=transport)
8.6 核心包初始化 (core/init.py)
"""
MCP Server核心包
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
from core.module_interface import ModuleInterface
from core.module_registry import ModuleRegistry
from core.module_loader import ModuleLoader
from core.config_manager import ConfigManager
from core.server import MCPServer
# 导出核心类
__all__ = [
'ModuleInterface',
'ModuleRegistry',
'ModuleLoader',
'ConfigManager',
'MCPServer',
]
8.7 模块包初始化 (modules/init.py)
"""
MCP Server模块包
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
# 这个文件使modules目录成为一个Python包
# 模块发现和加载由模块加载器自动完成
8.8 Hello World模块 (modules/hello_world/hello.py)
"""
Hello World示例模块
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
from core.module_interface import ModuleInterface
class HelloWorldModule(ModuleInterface):
"""
Hello World示例模块,展示了一个简单的MCP模块实现
"""
def get_info(self):
"""
获取模块信息
Returns:
dict: 模块信息
"""
return {
"name": "hello_world",
"version": "1.0.0",
"description": "Hello World示例模块",
"author": "kinbos 严富坤",
"email": "fookinbos@gmail.com",
"website": "htttps://www.yanfukun.com"
}
def register(self, server):
"""
向MCP服务器注册功能
Args:
server: FastMCP服务器实例
"""
# 注册一个简单的工具
@server.tool()
def hello(name: str = "World") -> str:
"""
返回问候消息
Args:
name: 要问候的名称
Returns:
str: 问候消息
"""
message = self.config.get('message', 'Hello, {}!')
return message.format(name)
# 注册一个资源
@server.resource("hello://greeting")
def get_greeting() -> str:
"""
获取问候消息
Returns:
str: 问候消息
"""
return "Welcome to MCP Server!"
# 注册一个提示模板
@server.prompt()
def hello_prompt(name: str = "User") -> str:
"""
创建一个问候提示
Args:
name: 要问候的名称
Returns:
str: 问候提示
"""
return f"Welcome, {name}! How can I assist you today?"
8.9 Hello World模块初始化 (modules/helloworld/_init.py)
"""
Hello World模块初始化
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
from modules.hello_world.hello import HelloWorldModule
# 导出模块类
__all__ = ['HelloWorldModule']
8.10 工具函数 (utils/helpers.py)
"""
工具函数
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
import importlib
import os
import sys
from pathlib import Path
def ensure_directory(path):
"""
确保目录存在,如果不存在则创建
Args:
path: 目录路径
Returns:
Path: 目录路径对象
"""
path = Path(path)
path.mkdir(parents=True, exist_ok=True)
return path
def import_module_from_path(module_name, module_path):
"""
从路径导入模块
Args:
module_name: 模块名称
module_path: 模块路径
Returns:
module: 导入的模块
"""
spec = importlib.util.spec_from_file_location(module_name, module_path)
if spec is None:
return None
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module
def format_error(e):
"""
格式化异常信息
Args:
e: 异常对象
Returns:
str: 格式化后的异常信息
"""
return f"{type(e).__name__}: {str(e)}"
8.11 工具包初始化 (utils/init.py)
"""
工具函数包
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
from utils.helpers import ensure_directory, import_module_from_path, format_error
# 导出工具函数
__all__ = [
'ensure_directory',
'import_module_from_path',
'format_error',
]
8.12 主入口文件 (main.py)
"""
MCP Server主入口文件
作者:kinbos 严富坤
邮箱:fookinbos@gmail.com
个人网站:htttps://www.yanfukun.com
"""
import os
import sys
from pathlib import Path
# 添加当前目录到Python路径
current_dir = Path(__file__).parent
sys.path.append(str(current_dir))
# 导入核心服务器类
from core.server import MCPServer
def main():
"""
主入口函数
"""
# 默认配置文件路径
config_path = current_dir / 'config.json'
# 默认模块目录
modules_dir = current_dir / 'modules'
# 创建并初始化MCP服务器
try:
server = MCPServer(
config_path=config_path,
modules_dir=modules_dir
).initialize()
# 运行服务器
server.run()
except Exception as e:
print(f"错误: 启动MCP服务器失败: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
8.13 默认配置文件 (config.json)
{
"global": {
"server_name": "MCP Server",
"debug": true,
"log_level": "INFO",
"transport": "stdio",
"dependencies": []
},
"modules": {
"hello_world": {
"enabled": true,
"message": "Hello, {}!"
}
}
}
9. 配置和部署说明
9.1 配置MCP Server到Claude Desktop
要将框架配置到Claude Desktop中使用,可以使用以下步骤:
确保已安装Claude Desktop和uv:
# 安装uv (如果未安装)
brew install uv # macOS
# 或者
pip install uv # 其他平台
安装框架:
# 克隆仓库
git clone https://github.com/kinbos/mcp-server.git
cd mcp-server
# 安装依赖
uv pip install fastmcp
将框架安装到Claude Desktop:
# 方法一: 使用FastMCP CLI
fastmcp install main.py
# 方法二: 直接运行
uv run main.py
或者手动编辑Claude Desktop配置文件:
{
"mcpServers": {
"mcp_server": {
"command": "uv",
"args": ["run", "/path/to/mcp-server/main.py"]
}
}
}
9.2 扩展框架
要扩展框架功能,只需按照以下步骤添加新模块:
- 在
modules
目录下创建新的模块目录 - 创建
__init__.py
文件和模块实现文件 - 实现
ModuleInterface
接口 - 在
config.json
中启用模块
框架会自动发现并加载新模块。
10. 测试和验证
要测试框架是否正常工作,可以使用:
- 启动Claude Desktop并连接到MCP Server
- 在Claude对话中测试Hello World模块功能:
- 输入: “使用hello工具向John问好”
- 预期输出: “Hello, John!”
- 输入: “获取问候资源”
- 预期输出: 访问”hello://greeting”资源,显示”Welcome to MCP Server!”
- 输入: “列出服务器信息”
- 预期输出: 显示服务器名称、版本和已加载模块列表
这个MCP Server框架提供了一个灵活、可扩展的基础,用于构建与Claude Desktop集成的MCP服务器。框架的模块化设计使得开发者可以轻松添加新功能,而无需修改核心代码。通过自动发现和配置驱动的方法,框架实现了高度的可配置性和可扩展性,适用于各种MCP应用场景。