Fix(mcp): Unreachable structured content branch in invoke_mcp_tool

[Kunmeer-SyedMohamedHyder]

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):

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):

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:

# 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

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

[seratch]

Could you resolve the lint and typecheck errors?

[Kunmeer-SyedMohamedHyder]

Could you resolve the lint and typecheck errors?

Hey @seratch. Fixed them!

[seratch]

Looks good to me; @rm-openai can you do final check before merging it?

[Kunmeer-SyedMohamedHyder]

Following is the code I ran!

#!/usr/bin/env python3
import asyncio
import json
from typing import Any

from mcp.types import CallToolResult, TextContent, Tool as MCPTool
from agents.run_context import RunContextWrapper
from agents.mcp import MCPServer, MCPUtil


class TestMCPServer(MCPServer):
    def __init__(self, use_structured_content: bool = False):
        super().__init__(use_structured_content=use_structured_content)
        self._server_name = "test_server"
    
    async def cleanup(self) -> None:
        pass
    
    async def connect(self) -> None:
        pass
    
    async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None):
        raise NotImplementedError()
    
    async def list_prompts(self, run_context: RunContextWrapper[Any], agent):
        return []
    
    @property
    def name(self) -> str:
        return self._server_name
    
    async def list_tools(self, run_context: RunContextWrapper[Any], agent) -> list:
        return [MCPTool(name="search_users", description="test", inputSchema={"type": "object"})]
    
    async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
        return CallToolResult(
            content=[TextContent(text="Found 2 users", type="text")],
            structuredContent={
                "users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}],
                "total": 2
            }
        )


async def test_structured_content_fix():
    print("Testing MCP Structured Content Fix (Issue #1236)")
    
    run_context = RunContextWrapper(context=None)
    tool = MCPTool(name="search_users", description="test", inputSchema={"type": "object"})
    
    # Test 1: use_structured_content=False (returns text content)
    print("\nTest 1: use_structured_content=False")
    server_text = TestMCPServer(use_structured_content=False)
    result = await MCPUtil.invoke_mcp_tool(server_text, tool, run_context, "{}")
    parsed = json.loads(result)
    print(f"Result: {result}")
    print(f"PASS: Returns text content" if "text" in parsed else "FAIL: Expected text content")
    
    # Test 2: use_structured_content=True (THE FIX - returns structured content)
    print("\nTest 2: use_structured_content=True (THE FIX)")
    print("Before fix: Would return text content (unreachable path)")
    print("After fix: Returns structured content exclusively")
    
    server_structured = TestMCPServer(use_structured_content=True)
    result_structured = await MCPUtil.invoke_mcp_tool(server_structured, tool, run_context, "{}")
    parsed_structured = json.loads(result_structured)
    print(f"Result: {result_structured}")
    
    if "users" in parsed_structured and "text" not in parsed_structured:
        print(f"PASS: Returns structured content exclusively")
        print(f"Found {len(parsed_structured['users'])} users")
        print("FIX CONFIRMED: Previously unreachable code now works!")
    else:
        print("FAIL: Expected structured content or found text mixing")
    
    # Test 3: Fallback when no structured content
    print("\nTest 3: Fallback when no structured content")
    
    class FallbackServer(TestMCPServer):
        async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None):
            return CallToolResult(
                content=[TextContent(text="No structured data", type="text")],
                structuredContent=None
            )
    
    server_fallback = FallbackServer(use_structured_content=True)
    result_fallback = await MCPUtil.invoke_mcp_tool(server_fallback, tool, run_context, "{}")
    parsed_fallback = json.loads(result_fallback)
    print(f"Result: {result_fallback}")
    print("PASS: Fallback to text content works" if parsed_fallback.get("text") == "No structured data" else "FAIL: Fallback behavior not working")


if __name__ == "__main__":
    asyncio.run(test_structured_content_fix())

Did a quick test @seratch!! BTW thanks to GPT! 😄

[seratch]

@rm-openai can you take a look at this before making next release?