From 762c5364331f87038c3181de15da8a5f8d7f9645 Mon Sep 17 00:00:00 2001 From: SyedMohamedHyder Date: Fri, 25 Jul 2025 22:01:02 +0530 Subject: [PATCH 1/2] use_structured_content given preference when available --- src/agents/mcp/util.py | 28 ++- tests/mcp/test_mcp_util.py | 345 ++++++++++++++++++++++++++++++++++++- 2 files changed, 357 insertions(+), 16 deletions(-) diff --git a/src/agents/mcp/util.py b/src/agents/mcp/util.py index 6b2b4679f..85d0e8eeb 100644 --- a/src/agents/mcp/util.py +++ b/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: diff --git a/tests/mcp/test_mcp_util.py b/tests/mcp/test_mcp_util.py index af63665f8..f7c680d73 100644 --- a/tests/mcp/test_mcp_util.py +++ b/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" From f11fbdbc1ab05fbc0cf7aa6766b52c9cffddb735 Mon Sep 17 00:00:00 2001 From: SyedMohamedHyder Date: Mon, 28 Jul 2025 13:17:20 +0530 Subject: [PATCH 2/2] Fix lint and mypy --- src/agents/mcp/util.py | 6 +- tests/mcp/test_mcp_util.py | 293 +++++++++++++++++-------------------- 2 files changed, 140 insertions(+), 159 deletions(-) diff --git a/src/agents/mcp/util.py b/src/agents/mcp/util.py index 85d0e8eeb..07c556439 100644 --- a/src/agents/mcp/util.py +++ b/src/agents/mcp/util.py @@ -194,13 +194,13 @@ async def invoke_mcp_tool( else: logger.debug(f"MCP tool {tool.name} returned {result}") - # 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: - # The MCP tool result is a list of content items, whereas OpenAI tool outputs are a single - # string. We'll try to convert. + # 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: diff --git a/tests/mcp/test_mcp_util.py b/tests/mcp/test_mcp_util.py index f7c680d73..e434f7542 100644 --- a/tests/mcp/test_mcp_util.py +++ b/tests/mcp/test_mcp_util.py @@ -3,7 +3,7 @@ import pytest from inline_snapshot import snapshot -from mcp.types import Tool as MCPTool, CallToolResult, TextContent +from mcp.types import CallToolResult, TextContent, Tool as MCPTool from pydantic import BaseModel, TypeAdapter from agents import Agent, FunctionTool, RunContextWrapper @@ -355,165 +355,143 @@ async def test_util_adds_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): + 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 + 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.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 + 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 over text content. - + """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 @@ -522,53 +500,55 @@ async def test_structured_content_priority_over_text(): @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 +@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 @@ -577,26 +557,27 @@ async def test_backwards_compatibility_unchanged(): @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 + 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" @@ -605,29 +586,30 @@ async def test_empty_structured_content_fallback(): @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"]}} + {"id": 2, "name": "Item 2", "metadata": {"tags": ["c", "d"]}}, ], "pagination": {"page": 1, "total": 2}, - "status": "success" + "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 @@ -637,30 +619,31 @@ async def test_complex_structured_content(): @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") + 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 @@ -671,24 +654,22 @@ async def test_multiple_content_items_with_structured(): @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") - ] + 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