"""JXLS MCP服务器主入口"""
import asyncio
import logging
from typing import Any, Dict, List
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from pydantic import ValidationError
from .models import GenerateTemplateRequest
from .services import TemplateGenerator
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler()
]
)
# 设置日志编码
for handler in logging.root.handlers:
if hasattr(handler, 'stream') and hasattr(handler.stream, 'reconfigure'):
handler.stream.reconfigure(encoding='utf-8')
logger = logging.getLogger(__name__)
class JxlsTemplateServer:
"""JXLS模板生成MCP服务器"""
def __init__(self, output_directory: str = "./templates"):
self.template_generator = TemplateGenerator(output_directory)
self.server = Server("jxls-template-generator")
self._register_handlers()
def _register_handlers(self) -> None:
"""注册MCP处理器"""
@self.server.list_tools()
async def list_tools() -> List[Tool]:
"""列出可用工具"""
return [
Tool(
name="generateJxlsTemplate",
description="根据输入参数生成符合JXLS规范的Excel模板文件",
inputSchema={
"type": "object",
"properties": {
"templateName": {
"type": "string",
"description": "模板文件名称",
"minLength": 1,
"maxLength": 100
},
"dataStruct": {
"type": "object",
"description": "数据结构定义",
"properties": {
"collectName": {
"type": "string",
"description": "集合变量名称",
"minLength": 1
},
"itemVariable": {
"type": "string",
"description": "循环项变量名",
"minLength": 1
},
"dataFields": {
"type": "array",
"description": "字段定义数组",
"minItems": 1,
"items": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "列标题显示名称"
},
"field": {
"type": "string",
"description": "JSON格式字段名"
},
"index": {
"type": "integer",
"description": "数组格式索引位置",
"minimum": 0
}
},
"required": ["name"]
}
}
},
"required": ["collectName", "itemVariable", "dataFields"]
},
"dataFormat": {
"type": "string",
"enum": ["json", "array"],
"description": "数据格式类型"
},
"sampleData": {
"type": "array",
"description": "示例数据（可选）",
"items": {}
},
"outputPath": {
"type": "string",
"description": "导出文件路径（可选）"
}
},
"required": ["templateName", "dataStruct", "dataFormat"]
}
)
]
@self.server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""处理工具调用"""
if name == "generateJxlsTemplate":
return await self._handle_generate_template(arguments)
else:
raise ValueError(f"未知工具: {name}")
async def _handle_generate_template(self, arguments: Dict[str, Any]) -> List[TextContent]:
"""处理generateJxlsTemplate工具调用"""
try:
# 验证和解析请求
request = GenerateTemplateRequest(**arguments)
logger.info(f"生成模板请求: {request.templateName}, 格式: {request.dataFormat}")
# 生成模板
response = self.template_generator.generate_template(request)
# 记录结果
if response.success:
logger.info(f"模板生成成功: {response.templatePath}")
else:
logger.error(f"模板生成失败: {response.error} - {response.details}")
# 返回响应
return [TextContent(
type="text",
text=response.model_dump_json(indent=2)
)]
except ValidationError as e:
logger.error(f"请求参数验证失败: {e}")
error_response = {
"success": False,
"message": "请求参数验证失败",
"error": "参数验证失败",
"details": str(e)
}
return [TextContent(
type="text",
text=str(error_response)
)]
except Exception as e:
logger.error(f"处理请求时发生未知错误: {e}")
error_response = {
"success": False,
"message": "处理请求时发生未知错误",
"error": "内部服务器错误",
"details": str(e)
}
return [TextContent(
type="text",
text=str(error_response)
)]
async def run_stdio(self) -> None:
"""运行stdio模式"""
logger.info("Starting JXLS Template Generator MCP Server (stdio mode)")
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
self.server.create_initialization_options()
)
# 服务器实例
_server_instance = None
def get_server(output_directory: str = "./templates") -> JxlsTemplateServer:
"""获取服务器实例（单例模式）"""
global _server_instance
if _server_instance is None:
_server_instance = JxlsTemplateServer(output_directory)
return _server_instance
async def main() -> None:
"""主入口函数"""
import os
# 从环境变量获取输出目录
output_dir = os.getenv("JXLS_OUTPUT_DIR", "./templates")
# 创建服务器实例并运行
server = get_server(output_dir)
await server.run_stdio()
if __name__ == "__main__":
asyncio.run(main())
