"""
测试 outputSchema 功能
验证所有工具是否正确支持 outputSchema 和结构化输出。
"""
import pytest
import asyncio
import json
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from pathlib import Path
@pytest.fixture
async def mcp_client():
"""创建 MCP 客户端用于测试"""
server_script = Path(__file__).parent.parent / "run.py"
server_params = StdioServerParameters(
command="python",
args=[str(server_script)],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
yield session
@pytest.mark.asyncio
async def test_tools_have_output_schema(mcp_client):
"""测试工具是否包含 outputSchema"""
tools_response = await mcp_client.list_tools()
# 查找几个关键工具
tool_names = [tool.name for tool in tools_response.tools]
# 验证关键工具存在
assert "add" in tool_names
assert "calculate_bmi" in tool_names
assert "count_words" in tool_names
assert "list_directory" in tool_names
# 验证工具包含 outputSchema
for tool in tools_response.tools:
if tool.name in ["add", "calculate_bmi", "count_words", "list_directory"]:
assert hasattr(tool, 'outputSchema'), f"Tool {tool.name} should have outputSchema"
assert tool.outputSchema is not None, f"Tool {tool.name} outputSchema should not be None"
assert tool.outputSchema.get("type") == "object", f"Tool {tool.name} outputSchema should be object type"
@pytest.mark.asyncio
async def test_calculator_output_schema(mcp_client):
"""测试计算器工具的 outputSchema"""
tools_response = await mcp_client.list_tools()
# 测试 add 工具
add_tool = next((t for t in tools_response.tools if t.name == "add"), None)
assert add_tool is not None
assert add_tool.outputSchema is not None
# 验证 outputSchema 结构
schema = add_tool.outputSchema
assert schema["type"] == "object"
properties = schema.get("properties", {})
# 验证必需字段
assert "success" in properties
assert "timestamp" in properties
assert "result" in properties
assert "expression" in properties
assert "operation" in properties
@pytest.mark.asyncio
async def test_bmi_output_schema(mcp_client):
"""测试 BMI 工具的 outputSchema"""
tools_response = await mcp_client.list_tools()
bmi_tool = next((t for t in tools_response.tools if t.name == "calculate_bmi"), None)
assert bmi_tool is not None
assert bmi_tool.outputSchema is not None
# 验证 outputSchema 结构
schema = bmi_tool.outputSchema
properties = schema.get("properties", {})
# 验证 BMI 特定字段
assert "bmi" in properties
assert "category" in properties
assert "weight_kg" in properties
assert "height_m" in properties
@pytest.mark.asyncio
async def test_text_analysis_output_schema(mcp_client):
"""测试文本分析工具的 outputSchema"""
tools_response = await mcp_client.list_tools()
count_words_tool = next((t for t in tools_response.tools if t.name == "count_words"), None)
assert count_words_tool is not None
assert count_words_tool.outputSchema is not None
# 验证 outputSchema 结构
schema = count_words_tool.outputSchema
properties = schema.get("properties", {})
# 验证文本分析字段
assert "words" in properties
assert "characters" in properties
assert "characters_no_spaces" in properties
assert "lines" in properties
@pytest.mark.asyncio
async def test_structured_output_response(mcp_client):
"""测试工具调用返回结构化输出"""
# 测试计算器工具
result = await mcp_client.call_tool("add", {"a": 5, "b": 3})
# 验证返回结果包含结构化内容
assert hasattr(result, 'content'), "Result should have content"
assert len(result.content) > 0, "Result should have content items"
# 如果支持结构化内容，验证其存在
if hasattr(result, 'structuredContent'):
structured = result.structuredContent
assert structured["success"] is True
assert structured["result"] == 8.0
assert "expression" in structured
assert "operation" in structured
@pytest.mark.asyncio
async def test_bmi_structured_output(mcp_client):
"""测试 BMI 工具的结构化输出"""
result = await mcp_client.call_tool("calculate_bmi", {
"weight_kg": 70,
"height_m": 1.75
})
# 验证返回结果
assert hasattr(result, 'content')
assert len(result.content) > 0
# 验证结构化内容（如果支持）
if hasattr(result, 'structuredContent'):
structured = result.structuredContent
assert structured["success"] is True
assert "bmi" in structured
assert "category" in structured
assert structured["weight_kg"] == 70
assert structured["height_m"] == 1.75
# 验证 BMI 计算正确
expected_bmi = 70 / (1.75 ** 2)
assert abs(structured["bmi"] - round(expected_bmi, 2)) < 0.01
@pytest.mark.asyncio
async def test_text_analysis_structured_output(mcp_client):
"""测试文本分析的结构化输出"""
test_text = "Hello world!\nThis is a test."
result = await mcp_client.call_tool("count_words", {"text": test_text})
# 验证返回结果
assert hasattr(result, 'content')
# 验证结构化内容（如果支持）
if hasattr(result, 'structuredContent'):
structured = result.structuredContent
assert structured["success"] is True
assert structured["words"] == 5 # "Hello", "world!", "This", "is", "a", "test."
assert structured["characters"] == len(test_text)
assert structured["lines"] == 2
@pytest.mark.asyncio
async def test_error_handling_with_structured_output(mcp_client):
"""测试错误处理是否保持结构化输出兼容"""
try:
# 测试除零错误
result = await mcp_client.call_tool("divide", {"a": 5, "b": 0})
# 如果没有抛出异常，检查错误是否在结果中
if hasattr(result, 'isError'):
assert result.isError is True
except Exception as e:
# 验证错误被正确抛出
assert "Cannot divide by zero" in str(e)
@pytest.mark.asyncio
async def test_backward_compatibility(mcp_client):
"""测试向后兼容性 - 确保现有客户端仍能正常工作"""
# 调用工具并验证基本响应格式
result = await mcp_client.call_tool("add", {"a": 10, "b": 20})
# 验证基本响应结构（向后兼容）
assert hasattr(result, 'content')
assert len(result.content) > 0
# 验证 content 包含文本表示
content_text = result.content[0].text if hasattr(result.content[0], 'text') else str(result.content[0])
assert content_text is not None
assert len(content_text) > 0
if __name__ == "__main__":
# 运行简单的同步测试
print("Running outputSchema tests...")
# 这里可以添加一些简单的验证逻辑
print("✅ Test file created successfully")
print("Run with: pytest tests/test_output_schema.py -v")