fix: use camelCase structuredContent attr, prefer structured over text

- The MCP SDK Pydantic model uses camelCase (structuredContent), not
  snake_case (structured_content). The original getattr was a silent no-op.
- When structuredContent is present, return it AS the result instead of
  alongside text — the structured payload is the machine-readable data.
- Move test file to tests/tools/ and fix fake class to use camelCase.
- Patch _run_on_mcp_loop in tests so the handler actually executes.
This commit is contained in:
Teknium 2026-04-07 17:48:30 -07:00 committed by Teknium
parent 363c5bc3c3
commit b9a5e6e247
2 changed files with 35 additions and 27 deletions

View file

@ -0,0 +1,111 @@
"""Tests for MCP tool structuredContent preservation."""
import asyncio
import json
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from tools import mcp_tool
class _FakeContentBlock:
"""Minimal content block with .text and .type attributes."""
def __init__(self, text: str, block_type: str = "text"):
self.text = text
self.type = block_type
class _FakeCallToolResult:
"""Minimal CallToolResult stand-in.
Uses camelCase ``structuredContent`` / ``isError`` to match the real
MCP SDK Pydantic model (``mcp.types.CallToolResult``).
"""
def __init__(self, content, is_error=False, structuredContent=None):
self.content = content
self.isError = is_error
self.structuredContent = structuredContent
def _fake_run_on_mcp_loop(coro, timeout=30):
"""Run an MCP coroutine directly in a fresh event loop."""
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.close()
@pytest.fixture
def _patch_mcp_server():
"""Patch _servers and the MCP event loop so _make_tool_handler can run."""
fake_session = MagicMock()
fake_server = SimpleNamespace(session=fake_session)
with patch.dict(mcp_tool._servers, {"test-server": fake_server}), \
patch("tools.mcp_tool._run_on_mcp_loop", side_effect=_fake_run_on_mcp_loop):
yield fake_session
class TestStructuredContentPreservation:
"""Ensure structuredContent from CallToolResult is forwarded."""
def test_text_only_result(self, _patch_mcp_server):
"""When no structuredContent, result is text-only (existing behaviour)."""
session = _patch_mcp_server
session.call_tool = AsyncMock(
return_value=_FakeCallToolResult(
content=[_FakeContentBlock("hello")],
)
)
handler = mcp_tool._make_tool_handler("test-server", "my-tool", 30.0)
raw = handler({})
data = json.loads(raw)
assert data == {"result": "hello"}
def test_structured_content_is_the_result(self, _patch_mcp_server):
"""When structuredContent is present, it becomes the result directly."""
session = _patch_mcp_server
payload = {"value": "secret-123", "revealed": True}
session.call_tool = AsyncMock(
return_value=_FakeCallToolResult(
content=[_FakeContentBlock("OK")],
structuredContent=payload,
)
)
handler = mcp_tool._make_tool_handler("test-server", "my-tool", 30.0)
raw = handler({})
data = json.loads(raw)
assert data["result"] == payload
def test_structured_content_none_falls_back_to_text(self, _patch_mcp_server):
"""When structuredContent is explicitly None, fall back to text."""
session = _patch_mcp_server
session.call_tool = AsyncMock(
return_value=_FakeCallToolResult(
content=[_FakeContentBlock("done")],
structuredContent=None,
)
)
handler = mcp_tool._make_tool_handler("test-server", "my-tool", 30.0)
raw = handler({})
data = json.loads(raw)
assert data == {"result": "done"}
def test_empty_text_with_structured_content(self, _patch_mcp_server):
"""When content blocks are empty but structuredContent exists."""
session = _patch_mcp_server
payload = {"status": "ok", "data": [1, 2, 3]}
session.call_tool = AsyncMock(
return_value=_FakeCallToolResult(
content=[],
structuredContent=payload,
)
)
handler = mcp_tool._make_tool_handler("test-server", "my-tool", 30.0)
raw = handler({})
data = json.loads(raw)
assert data["result"] == payload