Fix(mcp): Unreachable structured content branch in invoke_mcp_tool

src/agents/mcp/util.py

@@ -194,23 +194,21 @@ async def invoke_mcp_tool(
         else:
             logger.debug(f"MCP tool {tool.name} returned {result}")
 
-        # The MCP tool result is a list of content items, whereas OpenAI tool outputs are a single
-        # string. We'll try to convert.
-        if len(result.content) == 1:
-            tool_output = result.content[0].model_dump_json()
-            # Append structured content if it exists and we're using it.
-            if server.use_structured_content and result.structuredContent:
-                tool_output = f"{tool_output}\n{json.dumps(result.structuredContent)}"
-        elif len(result.content) > 1:
-            tool_results = [item.model_dump(mode="json") for item in result.content]
-            if server.use_structured_content and result.structuredContent:
-                tool_results.append(result.structuredContent)
-            tool_output = json.dumps(tool_results)
-        elif server.use_structured_content and result.structuredContent:
+
+        # If structured content is requested and available, use it exclusively
+        if server.use_structured_content and result.structuredContent:
             tool_output = json.dumps(result.structuredContent)
         else:
-            # Empty content is a valid result (e.g., "no results found")
-            tool_output = "[]"
+            # The MCP tool result is a list of content items, whereas OpenAI tool outputs are a single
+            # string. We'll try to convert.
+            if len(result.content) == 1:
+                tool_output = result.content[0].model_dump_json()
+            elif len(result.content) > 1:
+                tool_results = [item.model_dump(mode="json") for item in result.content]
+                tool_output = json.dumps(tool_results)
+            else:
+                # Empty content is a valid result (e.g., "no results found")
+                tool_output = "[]"
 
         current_span = get_current_span()
         if current_span:

tests/mcp/test_mcp_util.py

@@ -3,7 +3,7 @@
 
 import pytest
 from inline_snapshot import snapshot
-from mcp.types import Tool as MCPTool
+from mcp.types import Tool as MCPTool, CallToolResult, TextContent
 from pydantic import BaseModel, TypeAdapter
 
 from agents import Agent, FunctionTool, RunContextWrapper
@@ -351,3 +351,346 @@ async def test_util_adds_properties():
     assert tool.params_json_schema == snapshot(
         {"type": "object", "description": "Test tool", "properties": {}}
     )
+
+
+class StructuredContentTestServer(FakeMCPServer):
+    """Test server that allows setting both content and structured content for testing."""
+
+    def __init__(self, use_structured_content: bool = False, **kwargs):
+        super().__init__(**kwargs)
+        self.use_structured_content = use_structured_content
+        self._test_content: list = []
+        self._test_structured_content: dict | None = None
+
+    def set_test_result(self, content: list, structured_content: dict | None = None):
+        """Set the content and structured content that will be returned by call_tool."""
+        self._test_content = content
+        self._test_structured_content = structured_content
+
+    async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
+        """Return test result with specified content and structured content."""
+        self.tool_calls.append(tool_name)
+
+        return CallToolResult(
+            content=self._test_content,
+            structuredContent=self._test_structured_content
+        )
+
+
+@pytest.mark.parametrize("use_structured_content,content,structured_content,expected_output", [
+    # Scenario 1: use_structured_content=True with structured content available
+    # Should return only structured content
+    (
+        True,
+        [TextContent(text="text content", type="text")],
+        {"data": "structured_value", "type": "structured"},
+        '{"data": "structured_value", "type": "structured"}'
+    ),
+
+    # Scenario 2: use_structured_content=False with structured content available
+    # Should return text content only (structured content ignored)
+    (
+        False,
+        [TextContent(text="text content", type="text")],
+        {"data": "structured_value", "type": "structured"},
+        '{"type":"text","text":"text content","annotations":null,"meta":null}'
+    ),
+
+    # Scenario 3: use_structured_content=True but no structured content
+    # Should fall back to text content
+    (
+        True,
+        [TextContent(text="fallback text", type="text")],
+        None,
+        '{"type":"text","text":"fallback text","annotations":null,"meta":null}'
+    ),
+
+    # Scenario 4: use_structured_content=True with empty structured content (falsy)
+    # Should fall back to text content
+    (
+        True,
+        [TextContent(text="fallback text", type="text")],
+        {},
+        '{"type":"text","text":"fallback text","annotations":null,"meta":null}'
+    ),
+
+    # Scenario 5: use_structured_content=True, structured content available, empty text content
+    # Should return structured content
+    (
+        True,
+        [],
+        {"message": "only structured"},
+        '{"message": "only structured"}'
+    ),
+
+    # Scenario 6: use_structured_content=False, multiple text content items
+    # Should return JSON array of text content
+    (
+        False,
+        [
+            TextContent(text="first", type="text"), 
+            TextContent(text="second", type="text")
+        ],
+        {"ignored": "structured"},
+        '[{"type": "text", "text": "first", "annotations": null, "meta": null}, {"type": "text", "text": "second", "annotations": null, "meta": null}]'
+    ),
+
+    # Scenario 7: use_structured_content=True, multiple text content, with structured content
+    # Should return only structured content (text content ignored)
+    (
+        True,
+        [
+            TextContent(text="ignored first", type="text"), 
+            TextContent(text="ignored second", type="text")
+        ],
+        {"priority": "structured"},
+        '{"priority": "structured"}'
+    ),
+
+    # Scenario 8: use_structured_content=False, empty content
+    # Should return empty array
+    (
+        False,
+        [],
+        None,
+        "[]"
+    ),
+
+    # Scenario 9: use_structured_content=True, empty content, no structured content
+    # Should return empty array
+    (
+        True,
+        [],
+        None,
+        "[]"
+    ),
+])
+@pytest.mark.asyncio
+async def test_structured_content_handling(
+    use_structured_content: bool,
+    content: list,
+    structured_content: dict | None,
+    expected_output: str
+):
+    """Test that structured content handling works correctly with various scenarios.
+
+    This test verifies the fix for the MCP tool output logic where:
+    - When use_structured_content=True and structured content exists, it's used exclusively
+    - When use_structured_content=False or no structured content, falls back to text content
+    - The old unreachable code path has been fixed
+    """
+
+    server = StructuredContentTestServer(use_structured_content=use_structured_content)
+    server.add_tool("test_tool", {})
+    server.set_test_result(content, structured_content)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="test_tool", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+    assert result == expected_output
+
+
+@pytest.mark.asyncio
+async def test_structured_content_priority_over_text():
+    """Test that when use_structured_content=True, structured content takes priority over text content.
+
+    This verifies the core fix: structured content should be used exclusively when available
+    and requested, not concatenated with text content.
+    """
+
+    server = StructuredContentTestServer(use_structured_content=True)
+    server.add_tool("priority_test", {})
+
+    # Set both text and structured content
+    text_content = [TextContent(text="This should be ignored", type="text")]
+    structured_content = {"important": "This should be returned", "value": 42}
+    server.set_test_result(text_content, structured_content)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="priority_test", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+
+    # Should return only structured content
+    import json
+    parsed_result = json.loads(result)
+    assert parsed_result == structured_content
+    assert "This should be ignored" not in result
+
+
+@pytest.mark.asyncio
+async def test_structured_content_fallback_behavior():
+    """Test fallback behavior when structured content is requested but not available.
+
+    This verifies that the logic properly falls back to text content processing
+    when use_structured_content=True but no structured content is provided.
+    """
+
+    server = StructuredContentTestServer(use_structured_content=True)
+    server.add_tool("fallback_test", {})
+
+    # Set only text content, no structured content
+    text_content = [TextContent(text="Fallback content", type="text")]
+    server.set_test_result(text_content, None)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="fallback_test", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+
+    # Should fall back to text content
+    import json
+    parsed_result = json.loads(result)
+    assert parsed_result["text"] == "Fallback content"
+    assert parsed_result["type"] == "text"
+
+
+@pytest.mark.asyncio 
+async def test_backwards_compatibility_unchanged():
+    """Test that default behavior (use_structured_content=False) remains unchanged.
+
+    This ensures the fix doesn't break existing behavior for servers that don't use
+    structured content or have it disabled.
+    """
+
+    server = StructuredContentTestServer(use_structured_content=False)
+    server.add_tool("compat_test", {})
+
+    # Set both text and structured content
+    text_content = [TextContent(text="Traditional text output", type="text")]
+    structured_content = {"modern": "structured output"}
+    server.set_test_result(text_content, structured_content)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="compat_test", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+
+    # Should return only text content (structured content ignored)
+    import json
+    parsed_result = json.loads(result)
+    assert parsed_result["text"] == "Traditional text output"
+    assert "modern" not in result
+
+
+@pytest.mark.asyncio
+async def test_empty_structured_content_fallback():
+    """Test that empty structured content (falsy values) falls back to text content.
+
+    This tests the condition: if server.use_structured_content and result.structuredContent
+    where empty dict {} should be falsy and trigger fallback.
+    """
+
+    server = StructuredContentTestServer(use_structured_content=True)
+    server.add_tool("empty_structured_test", {})
+
+    # Set text content and empty structured content
+    text_content = [TextContent(text="Should use this text", type="text")]
+    empty_structured = {}  # This should be falsy
+    server.set_test_result(text_content, empty_structured)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="empty_structured_test", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+
+    # Should fall back to text content because empty dict is falsy
+    import json
+    parsed_result = json.loads(result)
+    assert parsed_result["text"] == "Should use this text"
+    assert parsed_result["type"] == "text"
+
+
+@pytest.mark.asyncio
+async def test_complex_structured_content():
+    """Test handling of complex structured content with nested objects and arrays."""
+
+    server = StructuredContentTestServer(use_structured_content=True)
+    server.add_tool("complex_test", {})
+
+    # Set complex structured content
+    complex_structured = {
+        "results": [
+            {"id": 1, "name": "Item 1", "metadata": {"tags": ["a", "b"]}},
+            {"id": 2, "name": "Item 2", "metadata": {"tags": ["c", "d"]}}
+        ],
+        "pagination": {"page": 1, "total": 2},
+        "status": "success"
+    }
+
+    server.set_test_result([], complex_structured)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="complex_test", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+
+    # Should return the complex structured content as-is
+    import json
+    parsed_result = json.loads(result)
+    assert parsed_result == complex_structured
+    assert len(parsed_result["results"]) == 2
+    assert parsed_result["pagination"]["total"] == 2
+
+
+@pytest.mark.asyncio
+async def test_multiple_content_items_with_structured():
+    """Test that multiple text content items are ignored when structured content is available.
+
+    This verifies that the new logic prioritizes structured content over multiple text items,
+    which was one of the scenarios that had unclear behavior in the old implementation.
+    """
+
+    server = StructuredContentTestServer(use_structured_content=True)
+    server.add_tool("multi_content_test", {})
+
+    # Set multiple text content items and structured content
+    text_content = [
+        TextContent(text="First text item", type="text"),
+        TextContent(text="Second text item", type="text"),
+        TextContent(text="Third text item", type="text")
+    ]
+    structured_content = {"chosen": "structured over multiple text items"}
+    server.set_test_result(text_content, structured_content)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="multi_content_test", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+
+    # Should return only structured content, ignoring all text items
+    import json
+    parsed_result = json.loads(result)
+    assert parsed_result == structured_content
+    assert "First text item" not in result
+    assert "Second text item" not in result
+    assert "Third text item" not in result
+
+
+@pytest.mark.asyncio
+async def test_multiple_content_items_without_structured():
+    """Test that multiple text content items are properly handled when no structured content."""
+
+    server = StructuredContentTestServer(use_structured_content=True)
+    server.add_tool("multi_text_test", {})
+
+    # Set multiple text content items without structured content
+    text_content = [
+        TextContent(text="First", type="text"),
+        TextContent(text="Second", type="text")
+    ]
+    server.set_test_result(text_content, None)
+
+    ctx = RunContextWrapper(context=None)
+    tool = MCPTool(name="multi_text_test", inputSchema={})
+
+    result = await MCPUtil.invoke_mcp_tool(server, tool, ctx, "{}")
+
+    # Should return JSON array of text content items
+    import json
+    parsed_result = json.loads(result)
+    assert isinstance(parsed_result, list)
+    assert len(parsed_result) == 2
+    assert parsed_result[0]["text"] == "First"
+    assert parsed_result[1]["text"] == "Second"