Function unwrap_tool_result

src/mcp_cli/utils/serialization.py:11–44

Unwrap middleware ``ToolExecutionResult`` wrappers and MCP result dicts. When middleware is enabled, ``ToolManager`` returns the result wrapped in a ``ToolExecutionResult`` object (from ``chuk_tool_processor.mcp.middleware``). The inner payload is typically ``{"isError": bool, "content"

(obj: Any, *, max_depth: int = _UNWRAP_MAX_DEPTH)

def unwrap_tool_result(obj: Any, *, max_depth: int = _UNWRAP_MAX_DEPTH) -> Any:
    """Unwrap middleware ``ToolExecutionResult`` wrappers and MCP result dicts.

    When middleware is enabled, ``ToolManager`` returns the result wrapped in
    a ``ToolExecutionResult`` object (from ``chuk_tool_processor.mcp.middleware``).
    The inner payload is typically ``{"isError": bool, "content": ToolResult}``.
    This peels off those layers to get the actual content.

    Raises ``RuntimeError`` if any wrapper layer reports failure.
    """
    depth = 0
    while (
        hasattr(obj, "success")
        and hasattr(obj, "result")
        and not isinstance(obj, dict)
    ):
        if depth >= max_depth:
            raise RuntimeError(f"Exceeded max unwrap depth ({max_depth})")
        if not obj.success:
            error = getattr(obj, "error", None) or "Unknown tool error"
            raise RuntimeError(error)
        obj = obj.result
        depth += 1

    # Unwrap MCP call_tool dict pattern: {"isError": ..., "content": ...}
    if isinstance(obj, dict) and "content" in obj and "isError" in obj:
        if obj["isError"]:
            error_msg = obj.get("error") or obj.get("content") or "Tool returned an error"
            if not isinstance(error_msg, str):
                error_msg = str(error_msg)
            raise RuntimeError(error_msg)
        obj = obj["content"]

    return obj


def to_serializable(obj: Any) -> Any: