test_mcp_client.py•3.49 kB
#!/usr/bin/env python3
"""
测试MCP客户端连接
模拟MaxKB的MCP客户端行为
"""
import requests
import json
def test_mcp_connection():
"""测试MCP连接"""
base_url = "http://192.168.110.231:8000"
# 测试健康检查
print("1. 测试健康检查...")
try:
response = requests.get(f"{base_url}/health", timeout=5)
print(f"健康检查: {response.status_code} - {response.json()}")
except Exception as e:
print(f"健康检查失败: {e}")
return False
# 测试MCP初始化
print("\n2. 测试MCP初始化...")
init_request = {
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
},
"id": "init"
}
try:
response = requests.post(
f"{base_url}/mcp",
json=init_request,
headers={"Content-Type": "application/json"},
timeout=10
)
print(f"初始化响应: {response.status_code}")
if response.status_code == 200:
print(f"初始化成功: {response.json()}")
else:
print(f"初始化失败: {response.text}")
return False
except Exception as e:
print(f"初始化请求失败: {e}")
return False
# 测试工具列表
print("\n3. 测试工具列表...")
tools_request = {
"jsonrpc": "2.0",
"method": "tools/list",
"id": "tools"
}
try:
response = requests.post(
f"{base_url}/mcp",
json=tools_request,
headers={"Content-Type": "application/json"},
timeout=10
)
print(f"工具列表响应: {response.status_code}")
if response.status_code == 200:
result = response.json()
tools = result.get("result", {}).get("tools", [])
print(f"工具数量: {len(tools)}")
for tool in tools[:3]: # 只显示前3个工具
print(f" - {tool.get('name')}: {tool.get('description')}")
else:
print(f"工具列表失败: {response.text}")
return False
except Exception as e:
print(f"工具列表请求失败: {e}")
return False
# 测试工具调用
print("\n4. 测试工具调用...")
call_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "get_all_employees",
"arguments": {}
},
"id": "call"
}
try:
response = requests.post(
f"{base_url}/mcp",
json=call_request,
headers={"Content-Type": "application/json"},
timeout=30
)
print(f"工具调用响应: {response.status_code}")
if response.status_code == 200:
result = response.json()
print(f"工具调用成功: {result.get('result', {}).get('content', [{}])[0].get('text', '')[:100]}...")
else:
print(f"工具调用失败: {response.text}")
return False
except Exception as e:
print(f"工具调用请求失败: {e}")
return False
print("\n✅ 所有测试通过！MCP服务器工作正常")
return True
if __name__ == "__main__":
test_mcp_connection()