Fix(mcp): Unreachable structured content branch in invoke_mcp_tool (#1250)

### Summary

This PR handles the MCP tool output where structured content could never
be returned exclusively when `use_structured_content=True`. The
conditional logic checked for content length first, making the
structured content branch unreachable when both content types were
present.

**Before (broken logic):**
```python
if len(result.content) == 1:
    tool_output = result.content[0].model_dump_json()
elif server.use_structured_content and result.structuredContent:  # Never reached!
    tool_output = json.dumps(result.structuredContent)
```

**After (fixed logic):**
```python
if server.use_structured_content and result.structuredContent:
    tool_output = json.dumps(result.structuredContent)
else:
    # Fall back to regular text content processing
    if len(result.content) == 1:
        tool_output = result.content[0].model_dump_json()
```

**Example usage:**
```python
# MCP server with structured content enabled
server = MCPServer(use_structured_content=True)

# When invoke_mcp_tool processes a tool that returns both text and structured content:
# - Text content: "Found 2 users"
# - Structured content: {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}

result = await MCPUtil.invoke_mcp_tool(server, tool, context, input_json)

# Before fix: Would return '{"type":"text","text":"Found 2 users","annotations":null,"meta":null}'
# After fix: Returns '{"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}'
```

### Test plan

I've added thorough test coverage to make sure this fix works properly:
- Created tests for 9 different scenarios to verify structured content
takes priority when it should, and falls back correctly when it
shouldn't
- Added specific tests to ensure the core priority logic works as
expected and that we didn't break any existing functionality
- Confirmed that the tests catch the original bug (unreachable code) and
validate that our fix actually solves the problem

### Issue number

Fixes #1236

### Checks

- [x] I've added new tests (if relevant) - Added comprehensive
parameterized tests and individual test functions for all structured
content scenarios
- [x] I've added/updated the relevant documentation - Code comments
updated to explain the fix and logic flow
- [x] I've run `make lint` and `make format` - Code follows project
formatting standards
- [x] I've made sure tests pass - All existing tests continue to pass,
and new structured content tests validate the fix

---------

Co-authored-by: SyedMohamedHyder <[email protected]>

Author: Kunmeer-SyedMohamedHyder

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 = "[]"
+            # Fall back to regular text content processing
+            # 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 CallToolResult, TextContent, Tool as MCPTool
 from pydantic import BaseModel, TypeAdapter
 
 from agents import Agent, FunctionTool, RunContextWrapper
@@ -351,3 +351,327 @@ 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[Any] = []
+        self._test_structured_content: dict[str, Any] | None = None
+
+    def set_test_result(self, content: list[Any], structured_content: dict[str, Any] | 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[Any],
+    structured_content: dict[str, Any] | 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.
+
+    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: dict[str, Any] = {}  # 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"