MCP structured content handling and preservation bugs

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #5979 fix(mcp): prefer structuredContent over text + no_mcp sentinel (salvage #5898 + #5841)

  merged to main

  files: hermes_cli/tools_config.py, tests/hermes_cli/test_tools_config.py, tests/tools/test_mcp_structured_content.py, tools/mcp_tool.py

  Show the diff

  • hermes_cli/tools_config.py

    @@ -554,6 +554,7 @@ def _get_platform_tools(
         # MCP servers are expected to be available on all platforms by default.
         # If the platform explicitly lists one or more MCP server names, treat that
         # as an allowlist. Otherwise include every globally enabled MCP server.
    +    # Special sentinel: "no_mcp" in the toolset list disables all MCP servers.
         mcp_servers = config.get("mcp_servers") or {}
         enabled_mcp_servers = {
             name
    @@ -561,10 +562,15 @@ def _get_platform_tools(
             if isinstance(server_cfg, dict)
             and _parse_enabled_flag(server_cfg.get("enabled", True), default=True)
         }
    -    explicit_mcp_servers = explicit_passthrough & enabled_mcp_servers
    -    enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers)
    +    # Allow "no_mcp" sentinel to opt out of all MCP servers for this platform
    +    if "no_mcp" in toolset_names:
    +        explicit_mcp_servers = set()
    +        enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers - {"no_mcp"})
    +    else:
    +        explicit_mcp_servers = explicit_passthrough & enabled_mcp_servers
    +        enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers)
         if include_default_mcp_servers:
    -        if explicit_mcp_servers:
    +        if explicit_mcp_servers or "no_mcp" in toolset_names:
                 enabled_toolsets.update(explicit_mcp_servers)
             else:
                 enabled_toolsets.update(enabled_mcp_servers)

  • tests/hermes_cli/test_tools_config.py

    @@ -72,6 +72,45 @@ def test_get_platform_tools_keeps_enabled_mcp_servers_with_explicit_builtin_sele
         assert "web-search-prime" in enabled
     
     
    +def test_get_platform_tools_no_mcp_sentinel_excludes_all_mcp_servers():
    +    """The 'no_mcp' sentinel in platform_toolsets excludes all MCP servers."""
    +    config = {
    +        "platform_toolsets": {"cli": ["web", "terminal", "no_mcp"]},
    +        "mcp_servers": {
    +            "exa": {"url": "https://mcp.exa.ai/mcp"},
    +            "web-search-prime": {"url": "https://api.z.ai/api/mcp/web_search_prime/mcp"},
    +        },
    +    }
    +
    +    enabled = _get_platform_tools(config, "cli")
    +
    +    assert "web" in enabled
    +    assert "terminal" in enabled
    +    assert "exa" not in enabled
    +    assert "web-search-prime" not in enabled
    +    assert "no_mcp" not in enabled
    +
    +
    +def test_get_platform_tools_no_mcp_sentinel_does_not_affect_other_platforms():
    +    """The 'no_mcp' sentinel only affects the platform it's configured on."""
    +    config = {
    +        "platform_toolsets": {
    +            "api_server": ["web", "terminal", "no_mcp"],
    +        },
    +        "mcp_servers": {
    +            "exa": {"url": "https://mcp.exa.ai/mcp"},
    +        },
    +    }
    +
    +    # api_server should exclude MCP
    +    api_enabled = _get_platform_tools(config, "api_server")
    +    assert "exa" not in api_enabled
    +
    +    # cli (not configured with no_mcp) should include MCP
    +    cli_enabled = _get_platform_tools(config, "cli")
    +    assert "exa" in cli_enabled
    +
    +
     def test_toolset_has_keys_for_vision_accepts_codex_auth(tmp_path, monkeypatch):
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
         (tmp_path / "auth.json").write_text(

  • tests/tools/test_mcp_structured_content.py

    @@ -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()
    +
    +
    [redacted:email]
    +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

  • tools/mcp_tool.py

    @@ -1253,7 +1253,13 @@ async def _call():
                 for block in (result.content or []):
                     if hasattr(block, "text"):
                         parts.append(block.text)
    -            return json.dumps({"result": "\n".join(parts) if parts else ""})
    +            text_result = "\n".join(parts) if parts else ""
    +
    +            # Prefer structuredContent (machine-readable JSON) over plain text
    +            structured = getattr(result, "structuredContent", None)
    +            if structured is not None:
    +                return json.dumps({"result": structured})
    +            return json.dumps({"result": text_result})
     
             try:
                 return _run_on_mcp_loop(_call(), timeout=tool_timeout)