Bloom_Ware / tests /test_mcp_server_contract.py
XiaoBai1221's picture
feat: overhaul MCP architecture with structured tool schemas, comprehensive care-mode skill definitions, and enhanced test coverage for pipelines and service integration.
79df050
import pytest
from features.mcp.server import FeaturesMCPServer
from features.mcp.types import Tool, ToolCallResult
@pytest.mark.asyncio
async def test_tools_list_includes_output_schema():
server = FeaturesMCPServer()
server.tools.clear()
server.register_tool(
Tool(
name="example",
description="Example tool",
inputSchema={"type": "object", "properties": {}},
outputSchema={
"type": "object",
"properties": {"success": {"type": "boolean"}},
"required": ["success"],
},
)
)
result = await server._handle_tools_list({})
assert result["tools"][0]["outputSchema"]["properties"]["success"]["type"] == "boolean"
@pytest.mark.asyncio
async def test_tools_call_preserves_structured_content():
async def handler(arguments):
return {
"success": True,
"content": "ok",
"value": arguments["value"],
}
server = FeaturesMCPServer()
server.tools.clear()
server.register_tool(
Tool(
name="example",
description="Example tool",
inputSchema={
"type": "object",
"properties": {"value": {"type": "string"}},
"required": ["value"],
},
handler=handler,
outputSchema={
"type": "object",
"properties": {
"success": {"type": "boolean"},
"content": {"type": "string"},
"value": {"type": "string"},
},
"required": ["success", "content", "value"],
},
)
)
result = await server._handle_tools_call({"name": "example", "arguments": {"value": "42"}})
assert result["content"] == [{"type": "text", "text": "ok"}]
assert result["structuredContent"]["value"] == "42"
assert result["isError"] is False
@pytest.mark.asyncio
async def test_tools_call_error_uses_safe_error_contract():
async def handler(arguments):
raise RuntimeError("secret internal detail")
server = FeaturesMCPServer()
server.tools.clear()
server.register_tool(
Tool(
name="bad_tool",
description="Bad tool",
inputSchema={"type": "object", "properties": {}},
handler=handler,
)
)
result = await server._handle_tools_call({"name": "bad_tool", "arguments": {}})
assert result["content"] == [{"type": "text", "text": "工具執行失敗"}]
assert result["structuredContent"]["error_code"] == "TOOL_EXECUTION_ERROR"
assert result["structuredContent"]["tool_name"] == "bad_tool"
assert result["isError"] is True
@pytest.mark.asyncio
async def test_tools_call_rejects_output_schema_violation():
async def handler(arguments):
return {
"success": True,
"content": "ok",
}
server = FeaturesMCPServer()
server.tools.clear()
server.register_tool(
Tool(
name="bad_output",
description="Bad output",
inputSchema={"type": "object", "properties": {}},
handler=handler,
outputSchema={
"type": "object",
"properties": {
"success": {"type": "boolean"},
"content": {"type": "string"},
"value": {"type": "string"},
},
"required": ["success", "content", "value"],
},
)
)
result = await server._handle_tools_call({"name": "bad_output", "arguments": {}})
assert result["content"] == [{"type": "text", "text": "工具輸出格式不符合契約"}]
assert result["structuredContent"]["error_code"] == "TOOL_OUTPUT_VALIDATION"
assert result["structuredContent"]["tool_name"] == "bad_output"
assert result["isError"] is True
def test_tool_call_result_serializes_mcp_fields():
result = ToolCallResult(
content=[{"type": "text", "text": "ok"}],
structuredContent={"success": True},
)
assert result.to_dict() == {
"content": [{"type": "text", "text": "ok"}],
"structuredContent": {"success": True},
"isError": False,
}