Spaces:
Sleeping
Sleeping
File size: 4,313 Bytes
79df050 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | 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,
}
|