Reduce code duplicated between tool invoke methods

Author: Mofef

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

@@ -63,6 +63,21 @@ def _resolve_headers(headers: dict[str, str | Secret] | None) -> dict[str, str]
     return resolved_headers
 
 
+def extract_first_text(result: str) -> str:
+    # Per MCP spec, content[] may contain TextContent, ImageContent, AudioContent, etc.
+    # Parse only first TextContent block (ToolInvoker requires dict, not list).
+    parsed: dict = json.loads(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):
+                # No TextContent found, return full parsed response as fallback
+                return text
+
+
 class AsyncExecutor:
     """Thread-safe event loop executor for running async code from sync contexts."""
 
@@ -1088,21 +1103,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 self.extract_first_text(result)
 
             return result
         except (MCPError, TimeoutError) as e:
@@ -1133,21 +1134,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(result)
 
             return result
         except asyncio.TimeoutError as e: