server.py•10.6 kB
"""
Excel MCP Server
基于Model Context Protocol的Excel处理服务器
"""
import logging
import sys
import json
from pathlib import Path
from typing import Dict, Any, List, Optional
from mcp.server import FastMCP
from .config import ConfigManager
from .security import SecurityValidator
from .file_manager import FileManager
from .data_processor import DataProcessor
from .sheet_operations import SheetOperations
# 初始化MCP服务器
mcp = FastMCP("Excel MCP Server")
# 全局组件
config_manager: Optional[ConfigManager] = None
security_validator: Optional[SecurityValidator] = None
file_manager: Optional[FileManager] = None
data_processor: Optional[DataProcessor] = None
sheet_operations: Optional[SheetOperations] = None
def setup_logging():
"""设置日志系统"""
global config_manager
if not config_manager:
return
log_dir = Path("logs")
log_dir.mkdir(exist_ok=True)
log_level = config_manager.get('logging.level', 'INFO')
log_file = config_manager.get('logging.file', './logs/excel_mcp.log')
logging.basicConfig(
level=getattr(logging, log_level.upper()),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(log_file),
logging.StreamHandler(sys.stderr) # 避免污染stdout
]
)
def initialize_components():
"""初始化所有组件"""
global config_manager, security_validator, file_manager, data_processor, sheet_operations
# 初始化配置管理器
config_manager = ConfigManager()
# 设置日志
setup_logging()
# 初始化安全验证器
security_validator = SecurityValidator(config_manager)
# 初始化文件管理器
file_manager = FileManager(config_manager, security_validator)
# 初始化数据处理器
data_processor = DataProcessor(file_manager)
# 初始化工作表操作器
sheet_operations = SheetOperations(file_manager)
logging.info("Excel MCP Server 组件初始化完成")
# MCP工具定义
@mcp.tool()
def read_excel_file(
file_path: str,
sheet_name: str = None,
cell_range: str = None
) -> Dict[str, Any]:
"""
读取Excel文件内容
Args:
file_path: Excel文件路径
sheet_name: 工作表名称（可选，默认读取第一个工作表）
cell_range: 单元格范围（可选，如 "A1:C10"，默认读取全部）
Returns:
包含数据和元信息的字典
"""
try:
# 获取文件信息
file_info = file_manager.get_file_info(file_path)
if not file_info:
return {"error": "无法获取文件信息，请检查文件路径和权限"}
# 确定工作表名称
if not sheet_name:
sheet_name = file_info["sheet_names"][0] if file_info["sheet_names"] else None
if not sheet_name:
return {"error": "文件中没有找到工作表"}
# 读取数据
result = data_processor.read_range(file_path, sheet_name, cell_range)
# 添加文件信息
if result.get("success"):
result["file_info"] = file_info
return result
except Exception as e:
logging.error(f"读取Excel文件失败: {e}")
return {"error": f"读取失败: {str(e)}"}
@mcp.tool()
def write_excel_file(
file_path: str,
data: List[List[Any]],
sheet_name: str = "Sheet1",
start_cell: str = "A1"
) -> Dict[str, Any]:
"""
写入数据到Excel文件
Args:
file_path: Excel文件路径
data: 要写入的数据（二维列表）
sheet_name: 工作表名称（默认为 "Sheet1"）
start_cell: 起始单元格（默认为 "A1"）
Returns:
操作结果
"""
try:
return data_processor.write_range(file_path, sheet_name, start_cell, data)
except Exception as e:
logging.error(f"写入Excel文件失败: {e}")
return {"error": f"写入失败: {str(e)}"}
@mcp.tool()
def create_excel_file(
file_path: str,
sheet_names: List[str] = None
) -> Dict[str, Any]:
"""
创建新的Excel文件
Args:
file_path: 新文件的路径
sheet_names: 工作表名称列表（可选）
Returns:
创建结果
"""
try:
if not sheet_names:
sheet_names = ["Sheet1"]
success = file_manager.create_workbook(file_path, sheet_names)
if success:
return {
"success": True,
"message": f"成功创建Excel文件: {file_path}",
"file_path": file_path,
"sheet_names": sheet_names
}
else:
return {"error": "创建Excel文件失败"}
except Exception as e:
logging.error(f"创建Excel文件失败: {e}")
return {"error": f"创建失败: {str(e)}"}
@mcp.tool()
def query_excel_data(
file_path: str,
sheet_name: str,
conditions: Dict[str, Any] = None,
columns: List[str] = None
) -> Dict[str, Any]:
"""
查询Excel数据
Args:
file_path: Excel文件路径
sheet_name: 工作表名称
conditions: 查询条件字典
columns: 返回的列名列表（可选）
Returns:
查询结果
"""
try:
result = data_processor.query_data(file_path, sheet_name, conditions)
# 如果指定了列，则过滤结果
if result.get("success") and columns:
all_columns = result.get("columns", [])
data = result.get("data", [])
# 找到指定列的索引
column_indices = []
for col in columns:
if col in all_columns:
column_indices.append(all_columns.index(col))
# 过滤数据
filtered_data = []
for row in data:
filtered_row = [row[i] if i < len(row) else None for i in column_indices]
filtered_data.append(filtered_row)
result["data"] = filtered_data
result["columns"] = columns
return result
except Exception as e:
logging.error(f"查询Excel数据失败: {e}")
return {"error": f"查询失败: {str(e)}"}
@mcp.tool()
def get_excel_info(file_path: str) -> Dict[str, Any]:
"""
获取Excel文件信息
Args:
file_path: Excel文件路径
Returns:
文件信息
"""
try:
info = file_manager.get_file_info(file_path)
if info:
return {"success": True, **info}
else:
return {"error": "无法获取文件信息"}
except Exception as e:
logging.error(f"获取Excel文件信息失败: {e}")
return {"error": f"获取信息失败: {str(e)}"}
@mcp.tool()
def create_worksheet(
file_path: str,
sheet_name: str,
position: int = None
) -> Dict[str, Any]:
"""
创建新工作表
Args:
file_path: Excel文件路径
sheet_name: 新工作表名称
position: 插入位置（可选）
Returns:
操作结果
"""
try:
return sheet_operations.create_sheet(file_path, sheet_name, position)
except Exception as e:
logging.error(f"创建工作表失败: {e}")
return {"error": f"创建失败: {str(e)}"}
@mcp.tool()
def delete_worksheet(
file_path: str,
sheet_name: str
) -> Dict[str, Any]:
"""
删除工作表
Args:
file_path: Excel文件路径
sheet_name: 要删除的工作表名称
Returns:
操作结果
"""
try:
return sheet_operations.delete_sheet(file_path, sheet_name)
except Exception as e:
logging.error(f"删除工作表失败: {e}")
return {"error": f"删除失败: {str(e)}"}
@mcp.tool()
def rename_worksheet(
file_path: str,
old_name: str,
new_name: str
) -> Dict[str, Any]:
"""
重命名工作表
Args:
file_path: Excel文件路径
old_name: 原工作表名称
new_name: 新工作表名称
Returns:
操作结果
"""
try:
return sheet_operations.rename_sheet(file_path, old_name, new_name)
except Exception as e:
logging.error(f"重命名工作表失败: {e}")
return {"error": f"重命名失败: {str(e)}"}
@mcp.tool()
def apply_formula(
file_path: str,
sheet_name: str,
cell: str,
formula: str
) -> Dict[str, Any]:
"""
应用Excel公式
Args:
file_path: Excel文件路径
sheet_name: 工作表名称
cell: 目标单元格（如 "A1"）
formula: 公式字符串
Returns:
操作结果
"""
try:
return data_processor.calculate_formula(file_path, sheet_name, cell, formula)
except Exception as e:
logging.error(f"应用公式失败: {e}")
return {"error": f"公式应用失败: {str(e)}"}
@mcp.tool()
def list_worksheets(file_path: str) -> Dict[str, Any]:
"""
列出所有工作表
Args:
file_path: Excel文件路径
Returns:
工作表列表和信息
"""
try:
return sheet_operations.list_sheets(file_path)
except Exception as e:
logging.error(f"列出工作表失败: {e}")
return {"error": f"列出失败: {str(e)}"}
@mcp.tool()
def get_range_info(
file_path: str,
sheet_name: str,
cell_range: str
) -> Dict[str, Any]:
"""
获取单元格范围信息
Args:
file_path: Excel文件路径
sheet_name: 工作表名称
cell_range: 单元格范围（如 "A1:C10"）
Returns:
范围信息
"""
try:
return data_processor.get_range_info(file_path, sheet_name, cell_range)
except Exception as e:
logging.error(f"获取范围信息失败: {e}")
return {"error": f"获取失败: {str(e)}"}
def main():
"""主函数"""
try:
# 初始化组件
initialize_components()
# 启动MCP服务器
logging.info("启动Excel MCP Server...")
mcp.run(transport="stdio")
except Exception as e:
logging.error(f"服务器启动失败: {e}")
sys.exit(1)
if __name__ == "__main__":
main()