#!/usr/bin/env python3
"""
MCP SDK 客户端测试脚本
测试基于官方 MCP Python SDK 的服务器
"""
import asyncio
import json
import logging
import subprocess
import sys
from typing import Any, Dict
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MCPSDKTester:
"""MCP SDK 测试客户端"""
def __init__(self):
self.session: ClientSession = None
async def connect_to_server(self, server_script: str = "mcp_server_sdk.py"):
"""连接到 MCP 服务器"""
try:
logger.info(f"正在连接到 MCP 服务器: {server_script}")
# 创建 STDIO 客户端 - 修复参数传递
from mcp.client.stdio import StdioServerParameters
server_config = StdioServerParameters(
command=sys.executable,
args=[server_script]
)
async with stdio_client(server_config) as (read, write):
async with ClientSession(read, write) as session:
self.session = session
# 初始化连接
await session.initialize()
logger.info("✅ 成功连接到 MCP 服务器")
# 运行测试
await self.run_tests()
except Exception as e:
logger.error(f"❌ 连接服务器失败: {e}")
raise
async def run_tests(self):
"""运行所有测试"""
logger.info("\n" + "="*60)
logger.info("🚀 开始 MCP SDK 服务器测试")
logger.info("="*60)
# 测试工具列表
await self.test_list_tools()
# 测试各个工具
await self.test_calculator_tool()
await self.test_text_analyzer_tool()
await self.test_file_reader_tool()
# 测试资源
await self.test_list_resources()
await self.test_read_resources()
# 测试错误处理
await self.test_error_handling()
logger.info("\n" + "="*60)
logger.info("🎉 所有测试完成！")
logger.info("="*60)
async def test_list_tools(self):
"""测试获取工具列表"""
logger.info("\n📋 测试: 获取工具列表")
try:
result = await self.session.list_tools()
logger.info("✅ 成功获取工具列表")
tools = result.tools if hasattr(result, 'tools') else []
logger.info(f" 发现 {len(tools)} 个工具:")
for tool in tools:
logger.info(f" - {tool.name}: {tool.description}")
except Exception as e:
logger.error(f"❌ 获取工具列表失败: {e}")
async def test_calculator_tool(self):
"""测试计算器工具"""
logger.info("\n🧮 测试: 计算器工具")
test_cases = [
("2 + 3", "简单加法"),
("10 * 5 - 2", "混合运算"),
("sqrt(16)", "平方根函数"),
("sin(pi/2)", "三角函数"),
("1/0", "除零错误测试")
]
for expression, description in test_cases:
try:
logger.info(f" 测试 {description}: {expression}")
result = await self.session.call_tool("calculator", {"expression": expression})
# 解析结果
content = result.content[0].text if result.content else "无结果"
result_data = json.loads(content)
if result_data.get("success"):
logger.info(f" ✅ 结果: {result_data.get('result')}")
else:
logger.info(f" ⚠️ 错误: {result_data.get('error')}")
except Exception as e:
logger.error(f" ❌ 计算器测试失败: {e}")
async def test_text_analyzer_tool(self):
"""测试文本分析工具"""
logger.info("\n📝 测试: 文本分析工具")
test_text = """
这是一个测试文本。它包含多个句子，用于测试文本分析功能。
文本分析可以统计字符数、词数、句子数等信息。
This is a test text in English. It contains multiple sentences for testing.
"""
try:
logger.info(" 分析测试文本...")
result = await self.session.call_tool("text_analyzer", {"text": test_text})
content = result.content[0].text if result.content else "无结果"
result_data = json.loads(content)
if result_data.get("success"):
stats = result_data.get("statistics", {})
logger.info(f" ✅ 字符数: {stats.get('characters')}")
logger.info(f" ✅ 词数: {stats.get('words')}")
logger.info(f" ✅ 句子数: {stats.get('sentences')}")
logger.info(f" ✅ 预计阅读时间: {result_data.get('reading_time_minutes')} 分钟")
else:
logger.info(f" ❌ 分析失败: {result_data.get('error')}")
except Exception as e:
logger.error(f" ❌ 文本分析测试失败: {e}")
async def test_file_reader_tool(self):
"""测试文件读取工具"""
logger.info("\n📁 测试: 文件读取工具")
test_files = [
("README.md", "读取README文件"),
("requirements.txt", "读取依赖文件"),
("nonexistent.txt", "不存在的文件"),
]
for file_path, description in test_files:
try:
logger.info(f" 测试 {description}: {file_path}")
result = await self.session.call_tool("file_reader", {"file_path": file_path})
content = result.content[0].text if result.content else "无结果"
result_data = json.loads(content)
if result_data.get("success"):
file_size = result_data.get("file_size", 0)
logger.info(f" ✅ 成功读取文件，大小: {file_size} 字节")
else:
logger.info(f" ⚠️ 读取失败: {result_data.get('error')}")
except Exception as e:
logger.error(f" ❌ 文件读取测试失败: {e}")
async def test_list_resources(self):
"""测试获取资源列表"""
logger.info("\n📚 测试: 获取资源列表")
try:
result = await self.session.list_resources()
logger.info("✅ 成功获取资源列表")
resources = result.resources if hasattr(result, 'resources') else []
logger.info(f" 发现 {len(resources)} 个资源:")
for resource in resources:
logger.info(f" - {resource.uri}: {resource.name}")
except Exception as e:
logger.error(f"❌ 获取资源列表失败: {e}")
async def test_read_resources(self):
"""测试读取资源"""
logger.info("\n📖 测试: 读取资源")
resources_to_test = [
"config://server",
"status://server",
"logs://server"
]
for resource_uri in resources_to_test:
try:
logger.info(f" 读取资源: {resource_uri}")
result = await self.session.read_resource(resource_uri)
if result.contents:
content = result.contents[0]
logger.info(f" ✅ 成功读取资源，类型: {content.mimeType}")
else:
logger.info(f" ⚠️ 资源内容为空")
except Exception as e:
logger.error(f" ❌ 读取资源失败: {e}")
async def test_error_handling(self):
"""测试错误处理"""
logger.info("\n⚠️ 测试: 错误处理")
# 测试调用不存在的工具
try:
logger.info(" 测试调用不存在的工具...")
await self.session.call_tool("nonexistent_tool", {})
except Exception as e:
logger.info(f" ✅ 正确处理了不存在的工具: {type(e).__name__}")
# 测试读取不存在的资源
try:
logger.info(" 测试读取不存在的资源...")
await self.session.read_resource("nonexistent://resource")
except Exception as e:
logger.info(f" ✅ 正确处理了不存在的资源: {type(e).__name__}")
async def main():
"""主函数"""
tester = MCPSDKTester()
try:
await tester.connect_to_server()
except KeyboardInterrupt:
logger.info("\n👋 测试被用户中断")
except Exception as e:
logger.error(f"❌ 测试失败: {e}")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
