import asyncio
import logging
from typing import Dict, Any, List
from mcp.server import Server
from mcp.types import Tool, TextContent
from config.settings import settings
from tools.api_proxy import EmployeeManagementTools
# 配置日志
logging.basicConfig(
level=getattr(logging, settings.log_level),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# 创建服务器实例
server = Server("employee-management-mcp")
employee_tools = EmployeeManagementTools()
@server.list_tools()
async def list_tools() -> List[Tool]:
"""返回所有可用的工具列表"""
tools = employee_tools.get_all_tools()
logger.info(f"返回 {len(tools)} 个工具")
return tools
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""处理工具调用"""
try:
logger.info(f"调用工具: {name}, 参数: {arguments}")
# 执行工具
result = await employee_tools.execute(name, arguments)
# 格式化响应
if result.get("success"):
content = f"✅ {result.get('message', '操作成功')}\n\n"
if result.get("data"):
content += f"数据: {result['data']}"
else:
content = f"❌ {result.get('error', '操作失败')}"
logger.info(f"工具 {name} 执行完成")
return [TextContent(type="text", text=content)]
except Exception as e:
error_msg = f"工具 {name} 执行失败: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=f"❌ {error_msg}")]
async def main():
"""主函数"""
logger.info(f"启动MCP服务器: {settings.mcp_host}:{settings.mcp_port}")
# 使用stdio服务器
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
from mcp.server import InitializationOptions, NotificationOptions
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="employee-management-mcp",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
asyncio.run(main())
