重构示例-精简main.py•7.78 kB
#!/usr/bin/env python3
"""
DataMaster MCP Server - 精简版主入口文件
这是重构后的精简版main.py，展示了模块化后的代码结构。
原文件5155行 → 精简后约200行
主要职责：
1. MCP服务器初始化
2. 模块导入和初始化
3. 工具函数注册协调
4. 服务器启动
重构优势：
- 代码结构清晰，易于理解
- 模块职责分离，便于维护
- 新功能开发更加高效
- 减少代码冲突和依赖问题
"""
import asyncio
import logging
from pathlib import Path
from mcp.server.fastmcp import FastMCP
# ================================
# 日志配置
# ================================
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('datamaster_mcp.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# ================================
# MCP服务器初始化
# ================================
# 初始化FastMCP服务器
mcp = FastMCP("DataMaster")
# ================================
# 模块导入和初始化
# ================================
# 导入各功能模块
try:
# 数据库模块
from .core.database import (
connect_data_source,
execute_sql,
query_external_database,
list_data_sources,
manage_database_config,
init_database_module
)
# 数据分析模块
from .core.data_analysis import (
analyze_data,
get_data_info,
init_analysis_module
)
# 数据处理模块
from .core.data_processing import (
process_data,
export_data,
init_processing_module
)
# API管理模块
from .core.api_manager import (
manage_api_config,
fetch_api_data,
api_data_preview,
create_api_storage_session,
list_api_storage_sessions,
init_api_module
)
logger.info("所有功能模块导入成功")
except ImportError as e:
logger.error(f"模块导入失败: {e}")
raise
# ================================
# 模块初始化
# ================================
def initialize_modules():
"""
初始化所有功能模块
这个函数负责：
1. 将MCP实例传递给各个模块
2. 初始化各模块的内部状态
3. 确保工具函数正确注册
"""
try:
# 初始化数据库模块
init_database_module(mcp)
logger.info("数据库模块初始化完成")
# 初始化数据分析模块
init_analysis_module(mcp)
logger.info("数据分析模块初始化完成")
# 初始化数据处理模块
init_processing_module(mcp)
logger.info("数据处理模块初始化完成")
# 初始化API管理模块
init_api_module(mcp)
logger.info("API管理模块初始化完成")
logger.info("所有模块初始化完成")
except Exception as e:
logger.error(f"模块初始化失败: {e}")
raise
# ================================
# 工具函数注册验证
# ================================
def verify_tool_registration():
"""
验证所有工具函数是否正确注册
这个函数检查：
1. 所有17个工具函数是否都已注册
2. 工具函数的元数据是否正确
3. 是否有重复注册的工具
"""
expected_tools = [
'connect_data_source',
'execute_sql',
'get_data_info',
'analyze_data',
'export_data',
'process_data',
'list_data_sources',
'manage_database_config',
'query_external_database',
'manage_api_config',
'fetch_api_data',
'api_data_preview',
'create_api_storage_session',
'list_api_storage_sessions'
]
# 获取已注册的工具
registered_tools = list(mcp.tools.keys())
# 检查是否所有工具都已注册
missing_tools = set(expected_tools) - set(registered_tools)
extra_tools = set(registered_tools) - set(expected_tools)
if missing_tools:
logger.warning(f"缺少工具函数: {missing_tools}")
if extra_tools:
logger.info(f"额外的工具函数: {extra_tools}")
logger.info(f"已注册工具函数数量: {len(registered_tools)}")
logger.info(f"预期工具函数数量: {len(expected_tools)}")
return len(missing_tools) == 0
# ================================
# 服务器启动和配置
# ================================
def setup_server():
"""
设置服务器配置
包括：
1. 服务器元数据配置
2. 错误处理配置
3. 性能优化配置
"""
# 设置服务器信息
mcp.server_info = {
"name": "DataMaster MCP Server",
"version": "1.0.3",
"description": "强大的数据分析和处理MCP服务器",
"author": "DataMaster Team",
"tools_count": len(mcp.tools)
}
logger.info("服务器配置完成")
def main():
"""
主函数 - 服务器启动入口
启动流程：
1. 初始化所有模块
2. 验证工具函数注册
3. 设置服务器配置
4. 启动MCP服务器
"""
try:
logger.info("DataMaster MCP Server 启动中...")
# 初始化模块
initialize_modules()
# 验证工具注册
if verify_tool_registration():
logger.info("所有工具函数注册验证通过")
else:
logger.warning("工具函数注册验证存在问题")
# 设置服务器
setup_server()
logger.info("DataMaster MCP Server 启动成功")
logger.info(f"已注册 {len(mcp.tools)} 个工具函数")
# 启动服务器
return mcp
except Exception as e:
logger.error(f"服务器启动失败: {e}")
raise
# ================================
# 模块入口点
# ================================
if __name__ == "__main__":
# 直接运行模式
server = main()
server.run()
else:
# 作为模块导入模式
server = main()
# ================================
# 模块导出
# ================================
# 导出MCP服务器实例，供外部使用
__all__ = ['mcp', 'server']
# ================================
# 重构对比总结
# ================================
"""
重构前后对比：
📊 代码量对比：
- 重构前：main.py 5155行
- 重构后：main.py ~200行 + 4个模块文件 (~500-800行/个)
- 总代码量基本不变，但结构更清晰
🏗️ 结构对比：
重构前：
- 单一文件包含所有功能
- 17个工具函数混在一起
- 大量辅助函数散布其中
- 难以维护和扩展
重构后：
- 功能模块清晰分离
- 主文件只负责协调和启动
- 每个模块专注特定功能领域
- 易于维护、测试和扩展
✅ 重构优势：
1. 可读性：每个文件专注特定功能，易于理解
2. 可维护性：修改某个功能不影响其他模块
3. 可测试性：模块独立，便于单元测试
4. 可扩展性：新功能可以独立开发和集成
5. 团队协作：不同开发者可以并行开发不同模块
6. 代码复用：模块可以在其他项目中复用
🎯 使用建议：
1. 新功能开发：在对应的功能模块中添加
2. Bug修复：直接定位到相关模块进行修改
3. 性能优化：可以针对特定模块进行优化
4. 代码审查：可以分模块进行代码审查
5. 文档维护：每个模块可以有独立的文档
这个重构方案既保持了功能的完整性，又大大提升了代码的可维护性和可扩展性。
"""