earlier failure for for_each

Author: kantord

mcp_client.py

@@ -382,27 +382,57 @@ async def batch_call_tool(
 
     # Connect once and call the tool multiple times with timeout per call
     results = []
+    total_items = len(arguments_list)
+
+    async def execute_calls(session):
+        nonlocal results
+        for idx, arguments in enumerate(arguments_list):
+            try:
+                result = await asyncio.wait_for(
+                    session.call_tool(tool_name, arguments=arguments),
+                    timeout=timeout,
+                )
+                results.append(result)
+            except Exception as e:
+                # Build informative error message with progress info
+                completed = len(results)
+                pending = total_items - idx - 1
+                failed_item = idx + 1  # 1-indexed for user-friendly message
+
+                # Extract partial results text for context
+                partial_results_text = []
+                for r in results:
+                    if hasattr(r, "content"):
+                        for content_item in r.content:
+                            if hasattr(content_item, "text"):
+                                partial_results_text.append(content_item.text)
+                    else:
+                        partial_results_text.append(str(r))
+
+                error_parts = [
+                    f"Batch tool call failed at item {failed_item} of {total_items}.",
+                    f"Completed: {completed} successful, {pending} pending.",
+                    f"Error: {str(e)}",
+                ]
+
+                if partial_results_text:
+                    error_parts.append(
+                        f"Partial results from successful calls:\n"
+                        + "\n".join(partial_results_text)
+                    )
+
+                raise RuntimeError("\n".join(error_parts)) from e
 
     if proxy_mode == "sse":
         async with sse_client(url) as (read, write):
             async with ClientSession(read, write) as session:
                 await session.initialize()
-                for arguments in arguments_list:
-                    result = await asyncio.wait_for(
-                        session.call_tool(tool_name, arguments=arguments),
-                        timeout=timeout,
-                    )
-                    results.append(result)
+                await execute_calls(session)
     elif proxy_mode == "streamable-http" or transport_type == "streamable-http":
         async with streamablehttp_client(url) as (read, write, _):
             async with ClientSession(read, write) as session:
                 await session.initialize()
-                for arguments in arguments_list:
-                    result = await asyncio.wait_for(
-                        session.call_tool(tool_name, arguments=arguments),
-                        timeout=timeout,
-                    )
-                    results.append(result)
+                await execute_calls(session)
     else:
         raise ValueError(
             f"Transport/proxy mode '{proxy_mode or transport_type}' not supported"

tests/test_mcp_client.py

@@ -895,6 +895,125 @@ async def test_batch_call_tool_workload_not_found(self, mocker):
             )
 
 
+@pytest.mark.asyncio
+class TestBatchCallToolPartialFailure:
+    """Test error reporting when some calls in a batch fail."""
+
+    async def test_batch_call_tool_reports_failure_index(self, mocker):
+        """Test that batch_call_tool reports which item failed."""
+        workload = {
+            "name": "test-server",
+            "status": "running",
+            "transport_type": "streamable-http",
+            "url": "http://localhost:8080/mcp",
+        }
+
+        mocker.patch("mcp_client.get_workloads", return_value=[workload])
+        mocker.patch(
+            "toolhive_client.discover_toolhive", return_value=("localhost", 8080)
+        )
+
+        call_count = 0
+
+        async def failing_on_third_call(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 3:
+                raise RuntimeError("API rate limit exceeded")
+            mock_result = MagicMock()
+            mock_result.content = [MagicMock(text=f"result_{call_count}")]
+            return mock_result
+
+        mock_session = MagicMock()
+        mock_session.initialize = AsyncMock()
+        mock_session.call_tool = MagicMock(side_effect=failing_on_third_call)
+
+        mock_client_session_instance = MagicMock()
+        mock_client_session_instance.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_client_session_instance.__aexit__ = AsyncMock(return_value=None)
+
+        mock_http = MagicMock()
+        mock_http.__aenter__ = AsyncMock(return_value=("read", "write", lambda: None))
+        mock_http.__aexit__ = AsyncMock(return_value=None)
+
+        mocker.patch("mcp_client.streamablehttp_client", return_value=mock_http)
+        mocker.patch("mcp_client.ClientSession", return_value=mock_client_session_instance)
+
+        call_args_list = [{"id": i} for i in range(5)]
+
+        with pytest.raises(RuntimeError) as exc_info:
+            await mcp_client.batch_call_tool("test-server", "fetch", call_args_list)
+
+        error_msg = str(exc_info.value)
+
+        # Should report which item failed (item 3 of 5)
+        assert "item 3" in error_msg.lower() or "3 of 5" in error_msg, (
+            f"Error should mention which item failed (item 3). Got: {error_msg}"
+        )
+
+        # Should report how many completed successfully
+        assert "2 successful" in error_msg.lower() or "2 completed" in error_msg.lower(), (
+            f"Error should mention 2 items completed successfully. Got: {error_msg}"
+        )
+
+        # Should report how many are still pending
+        assert "2 pending" in error_msg.lower(), (
+            f"Error should mention 2 items still pending. Got: {error_msg}"
+        )
+
+    async def test_batch_call_tool_includes_partial_results(self, mocker):
+        """Test that batch_call_tool includes partial results in error."""
+        workload = {
+            "name": "test-server",
+            "status": "running",
+            "transport_type": "streamable-http",
+            "url": "http://localhost:8080/mcp",
+        }
+
+        mocker.patch("mcp_client.get_workloads", return_value=[workload])
+        mocker.patch(
+            "toolhive_client.discover_toolhive", return_value=("localhost", 8080)
+        )
+
+        call_count = 0
+
+        async def failing_on_third_call(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 3:
+                raise RuntimeError("Connection timeout")
+            mock_result = MagicMock()
+            mock_result.content = [MagicMock(text=f"result_{call_count}")]
+            return mock_result
+
+        mock_session = MagicMock()
+        mock_session.initialize = AsyncMock()
+        mock_session.call_tool = MagicMock(side_effect=failing_on_third_call)
+
+        mock_client_session_instance = MagicMock()
+        mock_client_session_instance.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_client_session_instance.__aexit__ = AsyncMock(return_value=None)
+
+        mock_http = MagicMock()
+        mock_http.__aenter__ = AsyncMock(return_value=("read", "write", lambda: None))
+        mock_http.__aexit__ = AsyncMock(return_value=None)
+
+        mocker.patch("mcp_client.streamablehttp_client", return_value=mock_http)
+        mocker.patch("mcp_client.ClientSession", return_value=mock_client_session_instance)
+
+        call_args_list = [{"url": f"http://example.com/{i}"} for i in range(5)]
+
+        with pytest.raises(RuntimeError) as exc_info:
+            await mcp_client.batch_call_tool("test-server", "fetch", call_args_list)
+
+        error_msg = str(exc_info.value)
+
+        # Should include partial results that succeeded
+        assert "result_1" in error_msg and "result_2" in error_msg, (
+            f"Error should include partial results (result_1, result_2). Got: {error_msg}"
+        )
+
+
 @pytest.mark.asyncio
 class TestToolCallTimeout:
     """Test timeout handling for tool calls."""
@@ -1007,7 +1126,8 @@ async def hanging_call_tool(*args, **kwargs):
         import time
         start = time.time()
 
-        with pytest.raises(asyncio.TimeoutError):
+        # batch_call_tool wraps timeout errors in RuntimeError with progress info
+        with pytest.raises(RuntimeError) as exc_info:
             await mcp_client.batch_call_tool(
                 "test-server", "slow_tool", [{"id": 1}, {"id": 2}], timeout=test_timeout
             )
@@ -1022,6 +1142,12 @@ async def hanging_call_tool(*args, **kwargs):
             f"Batch call returned too quickly ({elapsed}s), timeout may not be working"
         )
 
+        # The error should be wrapped with progress info and mention it's a timeout
+        error_msg = str(exc_info.value)
+        assert "item 1 of 2" in error_msg.lower(), (
+            f"Error should indicate which item failed. Got: {error_msg}"
+        )
+
 
 @pytest.mark.asyncio
 class TestSelfFiltering: