diff --git a/docs/guide/smolagents.md b/docs/guide/smolagents.md index 931e1cf..ebf7705 100644 --- a/docs/guide/smolagents.md +++ b/docs/guide/smolagents.md @@ -122,3 +122,28 @@ You can find a fully working script of this example [here](https://github.com/gr + +## MCP tool execution errors + +A response with `isError=True` raises `MCPToolExecutionError` in both adapter modes. +It is checked before extracting text or `structuredContent`, so a failed lookup +cannot be mistaken for a successful empty result. The exception message includes +all non-empty text blocks and any structured error details. Its `result` attribute +retains the complete original `CallToolResult`, including non-text content and +metadata; `tool_name` identifies the adapted tool. + +```python +from mcpadapt.smolagents_adapter import MCPToolExecutionError + +try: + result = lookup(query="example") # an adapted MCP tool +except MCPToolExecutionError as error: + print(error) + # Record a technical failure; do not substitute {"results": []}. + original_response = error.result +``` + +When invoked by a smolagents agent, the framework handles this exception as a tool +execution failure. Successful responses keep their existing text/JSON behavior. +With `structured_output=True`, available `structuredContent` is returned unchanged, +including an empty dictionary or a valid empty lookup, even with no content blocks. diff --git a/src/mcpadapt/smolagents_adapter.py b/src/mcpadapt/smolagents_adapter.py index d48b542..ca98347 100644 --- a/src/mcpadapt/smolagents_adapter.py +++ b/src/mcpadapt/smolagents_adapter.py @@ -55,6 +55,26 @@ def _sanitize_function_name(name): return name +class MCPToolExecutionError(RuntimeError): + """An MCP tool reported an execution error, with its full result retained.""" + + def __init__(self, tool_name: str, result: mcp.types.CallToolResult): + """Retain the MCP result and expose text and structured error details.""" + self.tool_name = tool_name + self.result = result + details = [ + item.text + for item in result.content + if isinstance(item, mcp.types.TextContent) and item.text + ] + if result.structuredContent is not None: + details.append("structuredContent: " + json.dumps(result.structuredContent)) + message = f"MCP tool {tool_name} failed (isError=True)" + if details: + message += ": " + "\n".join(details) + super().__init__(message) + + class SmolAgentsAdapter(ToolAdapter): """Adapter for the `smolagents` framework. @@ -163,9 +183,9 @@ def forward( else: mcp_output = func(kwargs) - # Early exit for empty content - if not mcp_output.content: - raise ValueError(f"tool {self.name} returned an empty content") + # Errors must not become success-shaped text or structured data. + if mcp_output.isError: + raise MCPToolExecutionError(self.name, mcp_output) # Handle structured features if enabled if self.structured_output: @@ -176,6 +196,9 @@ def forward( ): return mcp_output.structuredContent + if not mcp_output.content: + raise ValueError(f"tool {self.name} returned an empty content") + # Handle multiple content warning (unified for both modes) if len(mcp_output.content) > 1: warning_msg = ( diff --git a/tests/test_smolagents_adapter.py b/tests/test_smolagents_adapter.py index a1ccb2b..fa1c0c3 100644 --- a/tests/test_smolagents_adapter.py +++ b/tests/test_smolagents_adapter.py @@ -1,4 +1,3 @@ -import logging from pathlib import Path from textwrap import dedent @@ -6,7 +5,7 @@ from mcp import StdioServerParameters from mcpadapt.core import MCPAdapt -from mcpadapt.smolagents_adapter import SmolAgentsAdapter +from mcpadapt.smolagents_adapter import MCPToolExecutionError, SmolAgentsAdapter from tests._server_utils import launch_mcp_server, terminate_mcp_server @@ -330,8 +329,8 @@ def string_tool() -> str: assert string_result["result"] == "Hello world" -def test_structured_output_warning(caplog): - """Test that warning is logged when tool returns unparseable JSON for structured output.""" +def test_structured_output_server_error(): + """A server output validation failure must be raised as an MCP error.""" server_script = dedent( ''' from mcp.server.fastmcp import FastMCP @@ -356,13 +355,7 @@ def invalid_tool() -> dict[str, Any]: ) as tools: tool = tools[0] - # Tool should still work but return error string - result = tool() - assert isinstance(result, str) - assert "error" in result.lower() - - # Warning should be logged about unparseable JSON - assert any( - r.levelno == logging.WARNING and "unparseable" in r.message.lower() - for r in caplog.records - ) + with pytest.raises(MCPToolExecutionError, match="isError=True") as exc: + tool() + assert exc.value.result.isError is True + assert "error" in str(exc.value).lower() diff --git a/tests/test_smolagents_adapter_errors.py b/tests/test_smolagents_adapter_errors.py new file mode 100644 index 0000000..be6954a --- /dev/null +++ b/tests/test_smolagents_adapter_errors.py @@ -0,0 +1,133 @@ +"""Offline regressions for MCP execution status through the native Tool call path.""" + +import pytest +from mcp.types import CallToolResult, ImageContent, TextContent, Tool +from smolagents import Tool as SmolTool + +from mcpadapt.smolagents_adapter import MCPToolExecutionError, SmolAgentsAdapter + + +def adapt_result(result: CallToolResult, structured_output: bool) -> SmolTool: + definition = Tool( + name="synthetic_lookup", + description="Offline synthetic lookup", + inputSchema={"type": "object", "properties": {}}, + ) + return SmolAgentsAdapter(structured_output=structured_output).adapt( + lambda arguments: result, definition + ) + + +@pytest.mark.parametrize("structured_output", [False, True]) +@pytest.mark.parametrize("structured_content", [None, {}, {"results": []}]) +def test_mcp_error_retains_all_text_and_full_result( + structured_output: bool, structured_content: dict | None +) -> None: + result = CallToolResult( + content=[ + TextContent(type="text", text="Synthetic backend failure"), + TextContent(type="text", text="Retry after reconnecting"), + ], + structuredContent=structured_content, + isError=True, + ) + with pytest.raises(MCPToolExecutionError) as exc: + adapt_result(result, structured_output)() + assert exc.value.tool_name == "synthetic_lookup" + assert exc.value.result is result + assert "isError=True" in str(exc.value) + assert "Synthetic backend failure" in str(exc.value) + assert "Retry after reconnecting" in str(exc.value) + if structured_content is not None: + assert "structuredContent:" in str(exc.value) + + +@pytest.mark.parametrize("structured_output", [False, True]) +@pytest.mark.parametrize( + "content", + [[], [ImageContent(type="image", data="AA==", mimeType="image/png")]], +) +def test_mcp_error_without_text_is_still_an_execution_error( + structured_output: bool, content: list +) -> None: + result = CallToolResult(content=content, isError=True) + with pytest.raises(MCPToolExecutionError, match="isError=True") as exc: + adapt_result(result, structured_output)() + assert exc.value.result is result + + +@pytest.mark.parametrize("explicit_status", [False, True]) +@pytest.mark.parametrize( + "payload", [{}, {"results": []}, {"results": [{"id": "example"}]}] +) +@pytest.mark.parametrize("content", [[], [TextContent(type="text", text="fallback")]]) +def test_success_returns_available_structured_content_unchanged( + explicit_status: bool, payload: dict, content: list +) -> None: + result = CallToolResult( + content=content, + structuredContent=payload, + **({"isError": False} if explicit_status else {}), + ) + assert adapt_result(result, True)() is result.structuredContent + + +@pytest.mark.parametrize("structured_output", [False, True]) +def test_valid_empty_lookup_keeps_existing_text_behavior( + structured_output: bool, +) -> None: + result = CallToolResult( + content=[TextContent(type="text", text='{"results": []}')], + structuredContent={"results": []}, + isError=False, + ) + assert adapt_result(result, structured_output)() == ( + {"results": []} if structured_output else '{"results": []}' + ) + + +def test_successful_json_text_fallback() -> None: + result = CallToolResult(content=[TextContent(type="text", text='{"results": []}')]) + assert adapt_result(result, True)() == {"results": []} + + +@pytest.mark.parametrize("structured_output", [False, True]) +def test_success_with_no_usable_content_preserves_validation( + structured_output: bool, +) -> None: + with pytest.raises(ValueError, match="returned an empty content"): + adapt_result(CallToolResult(content=[]), structured_output)() + + +def test_smolagents_execution_wraps_error_with_details_and_cause() -> None: + from smolagents import ToolCallingAgent + from smolagents.utils import AgentToolExecutionError + + result = CallToolResult( + content=[TextContent(type="text", text="Synthetic backend failure")], + structuredContent={"results": [], "code": "BACKEND_UNAVAILABLE"}, + isError=True, + ) + tool = adapt_result(result, True) + # Executing a tool does not call the model; no API key or model inference needed. + agent = ToolCallingAgent(tools=[tool], model=lambda *args, **kwargs: None) + with pytest.raises(AgentToolExecutionError) as exc: + agent.execute_tool_call(tool.name, {}) + assert "Synthetic backend failure" in str(exc.value) + assert "BACKEND_UNAVAILABLE" in str(exc.value) + assert isinstance(exc.value.__cause__, MCPToolExecutionError) + assert exc.value.__cause__.result is result + + +def test_error_with_only_structured_details() -> None: + result = CallToolResult( + content=[], structuredContent={"code": "BACKEND_UNAVAILABLE"}, isError=True + ) + with pytest.raises(MCPToolExecutionError, match="BACKEND_UNAVAILABLE"): + adapt_result(result, True)() + + +def test_successful_unparseable_text_still_warns(caplog) -> None: + result = CallToolResult(content=[TextContent(type="text", text="plain text")]) + assert adapt_result(result, True)() == "plain text" + assert "unparseable text" in caplog.text