"""
MCP 客户端测试脚本
测试 MCP FastAPI 服务器的各项功能
"""
import asyncio
import httpx
import json
import sys
from typing import Dict, Any
class MCPClient:
"""MCP 客户端类"""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url
self.client = httpx.AsyncClient()
async def close(self):
"""关闭客户端连接"""
await self.client.aclose()
async def health_check(self) -> Dict[str, Any]:
"""健康检查"""
try:
response = await self.client.get(f"{self.base_url}/health")
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
async def get_server_info(self) -> Dict[str, Any]:
"""获取服务器信息"""
try:
response = await self.client.get(f"{self.base_url}/")
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
async def list_tools(self) -> Dict[str, Any]:
"""获取工具列表"""
try:
response = await self.client.post(f"{self.base_url}/mcp/tools/list")
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""调用工具"""
try:
payload = {
"name": name,
"arguments": arguments
}
response = await self.client.post(
f"{self.base_url}/mcp/tools/call",
json=payload
)
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
async def list_resources(self) -> Dict[str, Any]:
"""获取资源列表"""
try:
response = await self.client.post(f"{self.base_url}/mcp/resources/list")
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
async def read_resource(self, uri: str) -> Dict[str, Any]:
"""读取资源"""
try:
payload = {"uri": uri}
response = await self.client.post(
f"{self.base_url}/mcp/resources/read",
json=payload
)
response.raise_for_status()
return response.json()
except Exception as e:
return {"error": str(e)}
def print_section(title: str):
"""打印测试章节标题"""
print(f"\n{'='*50}")
print(f" {title}")
print('='*50)
def print_result(test_name: str, result: Dict[str, Any], success: bool = True):
"""打印测试结果"""
status = "✅ PASS" if success and "error" not in result else "❌ FAIL"
print(f"\n{status} {test_name}")
print("-" * 30)
print(json.dumps(result, indent=2, ensure_ascii=False))
async def run_tests():
"""运行所有测试"""
client = MCPClient()
try:
print_section("MCP FastAPI 服务器测试")
# 1. 健康检查测试
print_section("1. 健康检查测试")
health_result = await client.health_check()
print_result("健康检查", health_result)
# 2. 服务器信息测试
print_section("2. 服务器信息测试")
server_info = await client.get_server_info()
print_result("服务器信息", server_info)
# 3. 工具列表测试
print_section("3. 工具功能测试")
tools_result = await client.list_tools()
print_result("获取工具列表", tools_result)
# 4. 工具调用测试
if "tools" in tools_result:
# 测试计算器工具
calc_result = await client.call_tool(
"calculator",
{"expression": "2 + 3 * 4"}
)
print_result("计算器工具 (2 + 3 * 4)", calc_result)
# 测试数学函数
math_result = await client.call_tool(
"calculator",
{"expression": "sqrt(16) + sin(0)"}
)
print_result("数学函数 (sqrt(16) + sin(0))", math_result)
# 测试文本分析工具
text_result = await client.call_tool(
"text_analyzer",
{"text": "这是一个测试文本。它包含多个句子。用于测试文本分析功能。"}
)
print_result("文本分析工具", text_result)
# 测试文件读取工具
file_result = await client.call_tool(
"file_reader",
{"file_path": "README.md", "max_size": 2048}
)
print_result("文件读取工具 (README.md)", file_result)
# 测试错误处理
error_result = await client.call_tool(
"calculator",
{"expression": "1/0"} # 除零错误
)
print_result("错误处理测试 (除零)", error_result, success=False)
# 5. 资源功能测试
print_section("4. 资源功能测试")
resources_result = await client.list_resources()
print_result("获取资源列表", resources_result)
if "resources" in resources_result:
# 读取各种资源
for resource in resources_result["resources"]:
resource_content = await client.read_resource(resource["uri"])
print_result(f"读取资源: {resource['name']}", resource_content)
# 6. 边界测试
print_section("5. 边界和错误测试")
# 测试不存在的工具
invalid_tool = await client.call_tool("nonexistent_tool", {})
print_result("调用不存在的工具", invalid_tool, success=False)
# 测试不存在的资源
invalid_resource = await client.read_resource("invalid://resource")
print_result("读取不存在的资源", invalid_resource, success=False)
# 测试无效参数
invalid_args = await client.call_tool("calculator", {"wrong_param": "value"})
print_result("无效参数测试", invalid_args, success=False)
print_section("测试完成")
print("🎉 所有测试已执行完毕！")
print("请检查上面的结果，确认服务器功能正常。")
except Exception as e:
print(f"❌ 测试过程中发生错误: {e}")
import traceback
traceback.print_exc()
finally:
await client.close()
def main():
"""主函数"""
print("🚀 启动 MCP FastAPI 服务器测试...")
print("请确保服务器已在 http://localhost:8000 上运行")
print("启动命令: uvicorn main:app --reload")
try:
asyncio.run(run_tests())
except KeyboardInterrupt:
print("\n👋 测试被用户中断")
except Exception as e:
print(f"❌ 运行测试时发生错误: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
