import subprocess
import json
import time
import sys
import os
import tempfile
def test_mcp_tools_directly():
print("🔗 直接测试MCP工具连接")
print("-" * 50)
print("📋 测试1: 获取MCP工具列表...")
try:
result = subprocess.run(
["echo", '{"jsonrpc": "2.0", "method": "tools/list", "id": 1}'],
input=None,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
print("标准输出长度:", len(result.stdout))
print("标准错误长度:", len(result.stderr))
if result.stdout:
print("输出内容预览:")
print(result.stdout[:500] + "..." if len(result.stdout) > 500 else result.stdout)
if result.stderr:
print("错误内容预览:")
print(result.stderr[:300] + "..." if len(result.stderr) > 300 else result.stderr)
except Exception as e:
print(f"❌ 工具列表测试失败: {e}")
return False
def test_aiw_mcp_server():
print("\n🚀 测试Agentic-Warden MCP服务器")
print("-" * 50)
try:
print("📝 测试1: 启动MCP服务器并获取工具...")
mcp_request = {
"jsonrpc": "2.0",
"method": "tools/list",
"id": 1
}
process = subprocess.Popen(
["./target/release/aiw", "mcp", "serve"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
stdout, stderr = process.communicate(
input=json.dumps(mcp_request),
timeout=10
)
print(f"✅ MCP服务器响应状态: {process.returncode}")
print(f"📊 响应长度: {len(stdout)} 字符")
print(f"⚠️ 错误长度: {len(stderr)} 字符")
success_indicators = [
"🚀 Agentic-Warden intelligent MCP router ready" in stderr,
"filesystem" in stdout or "filesystem" in stderr,
"memory" in stdout or "memory" in stderr,
"tools" in stdout.lower(),
"Embedding" in stderr ]
success_count = sum(success_indicators)
print(f"\n🎯 关键指标检查:")
print(f" - 智能路由器启动: {'✅' if success_indicators[0] else '❌'}")
print(f" - Filesystem服务器: {'✅' if success_indicators[1] else '❌'}")
print(f" - Memory服务器: {'✅' if success_indicators[2] else '❌'}")
print(f" - 工具列表响应: {'✅' if success_indicators[3] else '❌'}")
print(f" - 向量嵌入系统: {'✅' if success_indicators[4] else '❌'}")
print(f"\n📈 成功指标: {success_count}/5")
return success_count >= 3
except subprocess.TimeoutExpired:
process.kill()
print("❌ MCP服务器响应超时")
return False
except Exception as e:
print(f"❌ MCP服务器测试失败: {e}")
return False
def test_intelligent_routing():
print("\n🧠 测试智能路由功能")
print("-" * 50)
test_data = {
"task": "process_file",
"complexity": "medium",
"tools_needed": ["filesystem", "javascript"],
"expected_route": "javascript_workflow"
}
temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
json.dump(test_data, temp_file)
temp_file.close()
try:
print(f"📝 创建测试文件: {temp_file.name}")
routing_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "analyze_request",
"arguments": {
"user_request": "读取和分析JSON文件中的数据，计算统计信息",
"file_path": temp_file.name,
"request_complexity": "medium"
}
},
"id": 2
}
print("🧠 发送智能路由请求...")
process = subprocess.Popen(
["./target/release/aiw", "mcp", "serve"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
stdout, stderr = process.communicate(
input=json.dumps(routing_request),
timeout=15
)
print(f"✅ 路由响应状态: {process.returncode}")
routing_success = False
if "javascript" in stdout.lower() or "workflow" in stdout.lower():
routing_success = True
print("🎯 路由决策: JavaScript工作流")
elif "direct" in stdout.lower() or "mcp" in stdout.lower():
print("🎯 路由决策: 直接MCP调用")
routing_success = True
else:
print("🔍 路由决策: 需要分析响应内容")
print(f"响应预览: {stdout[:200]}...")
vector_indicators = [
"similarity" in stdout.lower(),
"embedding" in stdout.lower(),
"vector" in stdout.lower(),
"search" in stdout.lower()
]
vector_search_active = any(vector_indicators)
print(f"🔍 向量搜索活跃: {'✅' if vector_search_active else '❌'}")
return routing_success
except subprocess.TimeoutExpired:
process.kill()
print("❌ 智能路由响应超时")
return False
except Exception as e:
print(f"❌ 智能路由测试失败: {e}")
return False
finally:
try:
os.unlink(temp_file.name)
except:
pass
def test_end_to_end_workflow():
print("\n🔄 测试端到端工作流")
print("-" * 50)
test_data = {
"users": [
{"name": "Alice", "score": 85, "department": "Engineering"},
{"name": "Bob", "score": 92, "department": "Sales"},
{"name": "Charlie", "score": 78, "department": "Engineering"},
{"name": "Diana", "score": 95, "department": "Marketing"}
],
"metadata": {
"report_date": "2025-11-22",
"generated_by": "test_workflow"
}
}
temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
json.dump(test_data, temp_file)
temp_file.close()
try:
print(f"📁 创建用户数据文件: {temp_file.name}")
workflow_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "execute_analysis_workflow",
"arguments": {
"user_request": f"分析{temp_file.name}中的用户数据，计算每个部门的平均分，找出最高分用户，生成分析报告",
"data_file": temp_file.name,
"workflow_steps": [
"read_data",
"calculate_averages",
"find_top_scorer",
"generate_report",
"store_results"
],
"expected_output": {
"format": "json",
"includes_analysis": True,
"stores_in_memory": True
}
}
},
"id": 3
}
print("🔄 发送复杂工作流请求...")
process = subprocess.Popen(
["./target/release/aiw", "mcp", "serve"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
stdout, stderr = process.communicate(
input=json.dumps(workflow_request),
timeout=20
)
print(f"✅ 工作流响应状态: {process.returncode}")
print(f"📊 响应长度: {len(stdout)} 字符")
workflow_success = False
if "analysis" in stdout.lower() or "report" in stdout.lower():
workflow_success = True
print("✅ 工作流执行成功: 包含分析和报告")
elif "error" in stdout.lower() or "fail" in stdout.lower():
print("❌ 工作流执行失败")
else:
print("🔍 工作流状态: 需要分析响应内容")
js_indicators = [
"javascript" in stdout.lower(),
"js" in stdout.lower(),
"boa" in stdout.lower(),
"execute" in stdout.lower()
]
js_execution = any(js_indicators)
print(f"🐍 JavaScript执行: {'✅' if js_execution else '❌'}")
mcp_indicators = [
"filesystem" in stdout.lower() or "read" in stdout.lower(),
"memory" in stdout.lower() or "write" in stdout.lower() or "store" in stdout.lower()
]
mcp_calls = any(mcp_indicators)
print(f"🔌 MCP工具调用: {'✅' if mcp_calls else '❌'}")
return workflow_success and (js_execution or mcp_calls)
except subprocess.TimeoutExpired:
process.kill()
print("❌ 工作流执行超时")
return False
except Exception as e:
print(f"❌ 端到端工作流测试失败: {e}")
return False
finally:
try:
os.unlink(temp_file.name)
except:
pass
def run_all_tests():
print("🚀 开始Agentic-Warden真实智能路由测试")
print("=" * 70)
print("直接测试智能路由器的核心功能")
print("验证: MCP连接、智能路由、工作流执行")
start_time = time.time()
test_results = []
try:
test_mcp_tools_directly()
server_success = test_aiw_mcp_server()
test_results.append(("MCP服务器", server_success))
routing_success = test_intelligent_routing()
test_results.append(("智能路由", routing_success))
workflow_success = test_end_to_end_workflow()
test_results.append(("工作流执行", workflow_success))
except Exception as e:
print(f"❌ 测试运行异常: {e}")
total_time = time.time() - start_time
passed_tests = sum(1 for name, success in test_results if success)
total_tests = len(test_results)
success_rate = (passed_tests / total_tests * 100) if total_tests > 0 else 0
print("\n" + "=" * 70)
print("📊 真实智能路由测试总结")
print("=" * 70)
print(f"总测试数: {total_tests}")
print(f"通过测试: {passed_tests}")
print(f"失败测试: {total_tests - passed_tests}")
print(f"成功率: {success_rate:.1f}%")
print(f"总耗时: {total_time:.2f}秒")
print("\n🔍 详细结果:")
for name, success in test_results:
status = "✅" if success else "❌"
print(f"{status} {name}")
print("\n🎯 智能路由核心能力:")
capabilities = {
"MCP服务器连接": any(name == "MCP服务器" and success for name, success in test_results),
"智能路由决策": any(name == "智能路由" and success for name, success in test_results),
"工作流执行": any(name == "工作流执行" and success for name, success in test_results)
}
for capability, status in capabilities.items():
icon = "✅" if status else "❌"
print(f"{icon} {capability}")
if success_rate >= 67:
print("\n🎉 Agentic-Warden智能路由系统验证成功！")
print("✅ MCP服务器正常运行")
print("✅ 智能路由决策工作")
print("✅ 工作流执行功能正常")
print("\n🚀 系统已具备生产级智能管线编排能力！")
elif success_rate >= 33:
print("\n⚠️ Agentic-Warden智能路由系统部分验证")
print("系统基本功能可用，需要进一步优化")
else:
print("\n❌ Agentic-Warden智能路由系统需要改进")
print("核心功能存在问题，需要修复")
return success_rate >= 33
if __name__ == "__main__":
success = run_all_tests()
sys.exit(0 if success else 1)