fix(ci): resolve pyright errors on MCP content type union

Extract _extract_text() helper that uses hasattr() to safely access
.text on MCP content items (union of TextContent, ImageContent,
EmbeddedResource, etc.). Pyright now passes with 0 errors.

Author: mariomeyer

tests/integration/conftest.py

@@ -1,12 +1,23 @@
 """Shared fixtures for integration tests."""
 
 import json
+from typing import Any
+
 import pytest
 from fastmcp import Client
 
 _server_initialized = False
 
 
+def _extract_text(result: Any) -> str:
+    """Extract text from MCP CallToolResult, handling content type union."""
+    if result.content:
+        item = result.content[0]
+        if hasattr(item, "text"):
+            return item.text  # type: ignore[no-any-return]
+    return "[]"
+
+
 @pytest.fixture(scope="session")
 def mcp_server():
     """Build the full MCP server with all providers and transforms wired up.
@@ -37,31 +48,30 @@ def mcp_server():
     return mcp
 
 
-async def call_tool(mcp_server, tool_name: str, arguments: dict | None = None) -> dict:
+async def call_tool(mcp_server: Any, tool_name: str, arguments: dict | None = None) -> dict:
     """Call a tool through the MCP Client layer and return parsed JSON."""
     async with Client(mcp_server) as client:
         result = await client.call_tool('call_tool', {
             'name': tool_name,
             'arguments': arguments or {},
         })
-        return json.loads(result.content[0].text)
+        return json.loads(_extract_text(result))  # type: ignore[no-any-return]
 
 
-async def discover(mcp_server, query: str) -> list[dict]:
+async def discover(mcp_server: Any, query: str) -> list[dict]:
     """Search for tools via BM25 discover_tools and return parsed results."""
     async with Client(mcp_server) as client:
         result = await client.call_tool('discover_tools', {'query': query})
-        # discover_tools may return results in different formats
-        text = result.content[0].text if result.content else "[]"
+        text = _extract_text(result)
         try:
             parsed = json.loads(text)
             return parsed if isinstance(parsed, list) else []
         except json.JSONDecodeError:
             return []
 
 
-async def call_direct_tool(mcp_server, tool_name: str, arguments: dict | None = None) -> dict:
+async def call_direct_tool(mcp_server: Any, tool_name: str, arguments: dict | None = None) -> dict:
     """Call an always-visible tool directly (not through call_tool proxy)."""
     async with Client(mcp_server) as client:
         result = await client.call_tool(tool_name, arguments or {})
-        return json.loads(result.content[0].text)
+        return json.loads(_extract_text(result))  # type: ignore[no-any-return]

tests/integration/test_tool_scenarios.py

@@ -610,20 +610,20 @@ async def test_selective_loading_discover_tools(self, mcp_server):
             call_tool_name="call_tool",
         ))
 
+        from tests.integration.conftest import _extract_text
+
         async with Client(selective_mcp) as client:
             # Weather tools should NOT appear
             result = await client.call_tool("discover_tools", {"query": "weather forecast"})
-            text = result.content[0].text if result.content else "[]"
-            weather_results = json.loads(text) if text else []
+            weather_results = json.loads(_extract_text(result))
             weather_names = [r["name"] for r in weather_results] if isinstance(weather_results, list) else []
             assert not any("wx_" in n for n in weather_names), (
                 f"Weather tools found in selective (bank_of_canada only) server: {weather_names}"
             )
 
             # BOC tools SHOULD appear
             result2 = await client.call_tool("discover_tools", {"query": "exchange rates"})
-            text2 = result2.content[0].text if result2.content else "[]"
-            boc_results = json.loads(text2) if text2 else []
+            boc_results = json.loads(_extract_text(result2))
             boc_names = [r["name"] for r in boc_results] if isinstance(boc_results, list) else []
             assert any("boc_" in n for n in boc_names), (
                 f"No BOC tools found in selective server. Got: {boc_names}"