test_mcp_server.py•4.25 kB
#!/usr/bin/env python3
"""
MCP服务器测试脚本
用于测试员工管理系统MCP服务器的各项功能
"""
import asyncio
import json
import sys
import os
# 添加src目录到Python路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from src.tools.api_proxy import EmployeeManagementTools
async def test_tools():
"""测试所有工具"""
print("🧪 开始测试MCP服务器工具...")
# 创建工具实例
tools = EmployeeManagementTools()
# 测试1: 获取所有工具定义
print("\n1. 测试工具定义获取...")
all_tools = tools.get_all_tools()
print(f"✅ 成功获取 {len(all_tools)} 个工具:")
for tool in all_tools:
print(f" - {tool.name}: {tool.description}")
# 测试2: 测试工具参数验证
print("\n2. 测试工具参数验证...")
# 测试get_employee_by_id工具
test_tool = tools.get_employee_by_id_tool()
print(f"✅ {test_tool.name} 工具定义:")
print(f" 描述: {test_tool.description}")
print(f" 参数: {json.dumps(test_tool.inputSchema, indent=2, ensure_ascii=False)}")
# 测试add_employee工具
add_tool = tools.add_employee_tool()
print(f"✅ {add_tool.name} 工具定义:")
print(f" 描述: {add_tool.description}")
print(f" 必需参数: {add_tool.inputSchema.get('required', [])}")
print("\n✅ 所有工具定义测试通过!")
# 测试3: 模拟工具执行（不实际调用API）
print("\n3. 测试工具执行逻辑...")
# 模拟参数验证
test_args = {"userId": 1001}
if tools.validate_arguments(test_args):
print("✅ 参数验证通过")
else:
print("❌ 参数验证失败")
# 测试错误响应格式化
error_response = tools.format_error_response("测试错误")
print(f"✅ 错误响应格式化: {error_response}")
# 测试成功响应格式化
success_response = tools.format_success_response({"test": "data"}, "测试成功")
print(f"✅ 成功响应格式化: {success_response}")
print("\n✅ 所有测试通过!")
def test_config():
"""测试配置加载"""
print("\n🔧 测试配置加载...")
try:
from src.config.settings import settings
print(f"✅ MCP服务器配置:")
print(f" 主机: {settings.mcp_host}")
print(f" 端口: {settings.mcp_port}")
print(f" API基础URL: {settings.api_base_url}")
print(f" API超时: {settings.api_timeout}秒")
print(f" 日志级别: {settings.log_level}")
print("✅ 配置加载成功!")
except Exception as e:
print(f"❌ 配置加载失败: {e}")
def test_models():
"""测试数据模型"""
print("\n📋 测试数据模型...")
try:
from src.models.schemas import Employee, EmployeeSearchParams
# 测试Employee模型
employee_data = {
"firstName": "三",
"lastName": "张",
"salary": 8000.0,
"currency": "CNY",
"birthdate": "1990-01-01",
"isActive": True,
"level": "3"
}
employee = Employee(**employee_data)
print(f"✅ Employee模型创建成功: {employee.firstName} {employee.lastName}")
# 测试搜索参数模型
search_params = EmployeeSearchParams(lastName="张", isActive=True)
print(f"✅ 搜索参数模型创建成功: {search_params}")
print("✅ 数据模型测试通过!")
except Exception as e:
print(f"❌ 数据模型测试失败: {e}")
async def main():
"""主测试函数"""
print("🚀 MCP服务器测试开始")
print("=" * 50)
# 测试配置
test_config()
# 测试数据模型
test_models()
# 测试工具
await test_tools()
print("\n" + "=" * 50)
print("🎉 所有测试完成!")
print("\n📝 下一步:")
print("1. 确保后端API服务已启动 (http://localhost:10086)")
print("2. 运行 'python main.py' 启动MCP服务器")
print("3. 使用MCP客户端连接服务器进行测试")
if __name__ == "__main__":
asyncio.run(main())