Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions docs/guide/smolagents.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
29 changes: 26 additions & 3 deletions src/mcpadapt/smolagents_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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:
Expand All @@ -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 = (
Expand Down
21 changes: 7 additions & 14 deletions tests/test_smolagents_adapter.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
import logging
from pathlib import Path
from textwrap import dedent

import pytest
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


Expand Down Expand Up @@ -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
Expand All @@ -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()
133 changes: 133 additions & 0 deletions tests/test_smolagents_adapter_errors.py
Original file line number Diff line number Diff line change
@@ -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