test: Extract method to extract text from mcp result (#3154)

* Reduce code duplicated between tool invoke methods

* Add test for mcp result text extraction

* Add docstring, fix type and fallback

* Address reviewer comments

* skip tests if OPENAI_API_KEY exists but is empty

* skip tests if OPENAI_API_KEY exists but is empty

---------

Co-authored-by: Julian Risch <julian.risch@deepset.ai>

Author: Mofef

integrations/mcp/src/haystack_integrations/tools/mcp/mcp_tool.py

@@ -63,6 +63,27 @@ def _resolve_headers(headers: dict[str, str | Secret] | None) -> dict[str, str]
     return resolved_headers
 
 
+def _extract_first_text_element(tool_call_result: str) -> str | dict[str, Any]:
+    """
+    Return the first text content block from an MCP tool call result.
+
+    MCP tool call results may include mixed content types such as text, image, or
+    audio blocks. This helper extracts the first text block because the tool
+    invoker expects a single parsed payload rather than the full content list.
+    """
+    parsed: dict = json.loads(tool_call_result)
+    content: list = parsed.get("content", [])
+    for block in content:
+        if isinstance(block, dict) and block.get("type") == "text":
+            text = block.get("text", "")
+            try:
+                return json.loads(text)
+            except (json.JSONDecodeError, TypeError):
+                return text
+    # No TextContent found, return full parsed response as fallback
+    return parsed
+
+
 class AsyncExecutor:
     """Thread-safe event loop executor for running async code from sync contexts."""
 
@@ -1088,21 +1109,7 @@ async def invoke() -> Any:
             # Parse JSON to dict only when outputs_to_state is configured.
             # ToolInvoker requires dict for _merge_tool_outputs(); ToolCallResult.result expects str otherwise.
             if self.outputs_to_state:
-                parsed = json.loads(result)
-
-                # Per MCP spec, content[] may contain TextContent, ImageContent, AudioContent, etc.
-                # Parse only first TextContent block (ToolInvoker requires dict, not list).
-                content = parsed.get("content", [])
-                for block in content:
-                    if isinstance(block, dict) and block.get("type") == "text":
-                        text = block.get("text", "")
-                        try:
-                            return json.loads(text)
-                        except (json.JSONDecodeError, TypeError):
-                            return text
-
-                # No TextContent found, return full parsed response as fallback
-                return parsed
+                return _extract_first_text_element(result)
 
             return result
         except (MCPError, TimeoutError) as e:
@@ -1133,21 +1140,7 @@ async def ainvoke(self, **kwargs: Any) -> str | dict[str, Any]:
             # Parse JSON to dict only when outputs_to_state is configured.
             # ToolInvoker requires dict for _merge_tool_outputs(); ToolCallResult.result expects str otherwise.
             if self.outputs_to_state:
-                parsed = json.loads(result)
-
-                # Per MCP spec, content[] may contain TextContent, ImageContent, AudioContent, etc.
-                # Parse only first TextContent block (ToolInvoker requires dict, not list).
-                content = parsed.get("content", [])
-                for block in content:
-                    if isinstance(block, dict) and block.get("type") == "text":
-                        text = block.get("text", "")
-                        try:
-                            return json.loads(text)
-                        except (json.JSONDecodeError, TypeError):
-                            return text
-
-                # No TextContent found, return full parsed response as fallback
-                return parsed
+                return _extract_first_text_element(result)
 
             return result
         except asyncio.TimeoutError as e:

integrations/mcp/tests/test_mcp_tool.py

@@ -15,7 +15,7 @@
     MCPTool,
     StdioServerInfo,
 )
-from haystack_integrations.tools.mcp.mcp_tool import StdioClient
+from haystack_integrations.tools.mcp.mcp_tool import StdioClient, _extract_first_text_element
 
 from .mcp_memory_transport import InMemoryServerInfo
 from .mcp_servers_fixtures import calculator_mcp, echo_mcp
@@ -27,6 +27,25 @@ def simple_haystack_tool(name: str) -> str:
     return f"Hello, {name}!"
 
 
+# from https://modelcontextprotocol.io/specification/draft/server/tools#output-schema
+EXAMPLE_MCP_TOOL_CALL_RESULT = {
+    "content": [{"type": "text", "text": '{"temperature": 22.5, "conditions": "Partly cloudy", "humidity": 65}'}],
+    "structuredContent": {"temperature": 22.5, "conditions": "Partly cloudy", "humidity": 65},
+}
+
+
+def test_extract_first_text_element():
+    """Test that extract_first_text skips non-text blocks and parses the first text block."""
+    tool_call_result = EXAMPLE_MCP_TOOL_CALL_RESULT
+    tool_call_result["content"].insert(0, {"type": "image", "data": "ignored"})
+    tool_call_result["content"].insert(1, {"type": "text", "text": '{"answer": 42}'})  # target
+    tool_call_result = json.dumps(tool_call_result)
+
+    extracted = _extract_first_text_element(tool_call_result)
+
+    assert extracted == {"answer": 42}
+
+
 class TestMCPTool:
     """Tests for the MCPTool class using in-memory servers."""
 
@@ -236,7 +255,7 @@ async def test_stdio_client_stderr_handling(self, fileno_side_effect, fileno_ret
             else:
                 assert errlog is mock_stderr
 
-    @pytest.mark.skipif("OPENAI_API_KEY" not in os.environ, reason="OPENAI_API_KEY not set")
+    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
     @pytest.mark.integration
     def test_pipeline_warmup_with_mcp_tool(self):
         """Test lazy connection with Pipeline.warm_up() - replicates time_pipeline.py."""
@@ -259,7 +278,7 @@ def test_pipeline_warmup_with_mcp_tool(self):
             if tool:
                 tool.close()
 
-    @pytest.mark.skipif("OPENAI_API_KEY" not in os.environ, reason="OPENAI_API_KEY not set")
+    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
     @pytest.mark.integration
     def test_agent_with_state_mapping(self):
         """Test Agent with MCPTool using state-mapping to inject location from state."""

integrations/mcp/tests/test_mcp_toolset.py

@@ -382,7 +382,7 @@ async def test_toolset_no_state_config(self, calculator_toolset):
             assert tool.outputs_to_state is None
             assert tool.outputs_to_string is None
 
-    @pytest.mark.skipif("OPENAI_API_KEY" not in os.environ, reason="OPENAI_API_KEY not set")
+    @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
     @pytest.mark.integration
     async def test_pipeline_warmup_with_mcp_toolset(self):
         """Test lazy connection with Pipeline.warm_up() - replicates time_pipeline.py."""