fix(proxy): harden MCP isError failure logging (cost, sync dedup, non-result responses)

Make the error helper tolerate non-CallToolResult responses (dict or bare
content list) instead of raising AttributeError. Preserve the configured MCP
per-query cost on the failure path rather than zeroing it, and suppress the
@client wrapper's sync success log in addition to the async one so an isError
call isn't double-logged as a success.

Co-authored-by: Cursor <cursoragent@cursor.com>

litellm/litellm_core_utils/litellm_logging.py

@@ -2655,7 +2655,17 @@ class Logging(LiteLLMLoggingBaseClass):
         # chunks already delivered; the router stashes that recovered usage as
         # ``combined_usage_object`` and pre-computes its cost, so preserve it
         # here instead of zeroing the spend on an otherwise-failed request.
-        if self.model_call_details.get("combined_usage_object") is None:
+        # An MCP tool call that fails at the tool level (CallToolResult.isError)
+        # likewise still incurs the configured per-query cost, computed before
+        # this handler runs; don't zero it like a generic failed LLM call.
+        preserve_mcp_cost = (
+            self.call_type == CallTypes.call_mcp_tool.value
+            and self.model_call_details.get("response_cost")
+        )
+        if (
+            self.model_call_details.get("combined_usage_object") is None
+            and not preserve_mcp_cost
+        ):
             self.model_call_details["response_cost"] = 0
 
         if hasattr(exception, "headers") and isinstance(exception.headers, dict):

litellm/proxy/_experimental/mcp_server/server.py

@@ -178,6 +178,15 @@ def _mcp_session_id_from_headers(
     return None
 
 
+def _mcp_content_block_text(block: object) -> Optional[str]:
+    """The ``text`` of an MCP content block (``TextContent`` object or ``dict``),
+    or ``None`` when the block carries no usable text."""
+    text = (
+        block.get("text") if isinstance(block, dict) else getattr(block, "text", None)
+    )
+    return text if isinstance(text, str) and text else None
+
+
 def _jsonrpc_text_has_top_level_method(text: str) -> bool:
     """Whether a (possibly truncated) JSON-RPC envelope has a ``method`` key at
     the root object's top level.
@@ -2529,23 +2538,28 @@ if MCP_AVAILABLE:
 
         return response
 
-    def _get_mcp_tool_call_error_message(
-        response: "CallToolResult",
-    ) -> Optional[str]:
+    def _get_mcp_tool_call_error_message(response: object) -> Optional[str]:
         """The error text of a failed MCP ``tools/call`` result, or ``None`` when
         it succeeded.
 
         A failed tool call returns HTTP 200 with ``isError: true`` in the JSON-RPC
         result (MCP spec), so the result object itself is the only failure signal.
-        The message is the concatenated text of the result's ``TextContent``
-        blocks, falling back to a generic string when the tool reported no text.
+        Accepts a ``CallToolResult``, an equivalent ``dict`` (``isError`` true), or
+        any other shape (e.g. a bare content list from a legacy path), which is
+        always treated as success. The message is the concatenated text of the
+        result's content blocks, falling back to a generic string when the tool
+        reported no text.
         """
-        if not response.isError:
+        if isinstance(response, dict):
+            is_error = response.get("isError")
+            content = response.get("content") or ()
+        else:
+            is_error = getattr(response, "isError", None)
+            content = getattr(response, "content", None) or ()
+        if not is_error:
             return None
         texts = tuple(
-            text
-            for block in (response.content or ())
-            if isinstance((text := getattr(block, "text", None)), str) and text
+            text for block in content if (text := _mcp_content_block_text(block))
         )
         return "\n".join(texts) or "MCP tool call returned an error result"
 
@@ -2634,16 +2648,27 @@ if MCP_AVAILABLE:
                 # A tool returning ``isError: true`` (auth denied, upstream
                 # error, guardrail block, ...) is a failed call even though the
                 # MCP spec keeps it on HTTP 200. Log it as a failure so spend
-                # logs and the OTel span are marked ERROR, then mark async
-                # success as already handled so the @client wrapper doesn't
-                # also emit a success log for the same call.
+                # logs and the OTel span are marked ERROR. The configured MCP
+                # per-query cost still applies, so compute it up front; the
+                # failure handler preserves an MCP cost rather than zeroing it.
+                from litellm.proxy._experimental.mcp_server.cost_calculator import (
+                    MCPCostCalculator,
+                )
+
+                litellm_logging_obj.model_call_details["response_cost"] = (
+                    MCPCostCalculator.calculate_mcp_tool_call_cost(litellm_logging_obj)
+                )
                 await litellm_logging_obj.async_failure_handler(
                     MCPToolCallError(tool_error_message),
                     "",
                     start_time,
                     end_time,
                 )
+                # The @client wrapper still dispatches both async and sync
+                # success logging for this returned result; mark both handled so
+                # the same isError call isn't double-logged as a success.
                 litellm_logging_obj.has_run_logging(event_type="async_success")
+                litellm_logging_obj.has_run_logging(event_type="sync_success")
             else:
                 await litellm_logging_obj.async_success_handler(
                     result=response, start_time=start_time, end_time=end_time

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py

@@ -4188,6 +4188,20 @@ def test_get_mcp_tool_call_error_message():
         == "MCP tool call returned an error result"
     )
 
+    # dict-shaped result (isError true) is handled like a CallToolResult.
+    assert (
+        _get_mcp_tool_call_error_message(
+            {"isError": True, "content": [{"type": "text", "text": "boom"}]}
+        )
+        == "boom"
+    )
+
+    # A bare content list (legacy path, no isError attribute) is never an error,
+    # and must not raise: this is what reached the helper in the wild.
+    assert (
+        _get_mcp_tool_call_error_message([TextContent(type="text", text="ok")]) is None
+    )
+
 
 async def _run_call_mcp_tool_with_result(call_tool_result):
     """Drive ``call_mcp_tool`` against a stubbed ``execute_mcp_tool`` that returns
@@ -4306,6 +4320,126 @@ async def test_call_mcp_tool_success_result_logged_as_success():
     assert slp["status"] == "success"
 
 
+@pytest.mark.asyncio
+async def test_call_mcp_tool_legacy_list_response_logged_as_success():
+    """A bare content-list response (legacy path, no ``isError`` attribute) must
+    not crash the logging branch and is treated as a success.
+
+    Regression: the error helper assumed a CallToolResult and raised
+    AttributeError on a list, breaking the whole tool call."""
+    try:
+        from mcp.types import TextContent
+    except ImportError:
+        pytest.skip("MCP server not available")
+
+    list_response = [TextContent(type="text", text="ok")]
+    response, logging_obj = await _run_call_mcp_tool_with_result(list_response)
+
+    assert response == list_response
+    slp = logging_obj.model_call_details["standard_logging_object"]
+    assert slp["status"] == "success"
+
+
+@pytest.mark.asyncio
+async def test_call_mcp_tool_iserror_still_charges_configured_cost():
+    """A failed tools/call still bills the configured MCP per-query cost; routing
+    through the failure handler must not zero the spend like a generic failed LLM
+    call would.
+
+    Regression: the failure path zeroed response_cost, letting an authenticated
+    caller trigger an isError result to dodge the configured MCP cost."""
+    try:
+        from mcp.types import CallToolResult, TextContent
+    except ImportError:
+        pytest.skip("MCP server not available")
+
+    import uuid
+    from datetime import timezone
+
+    from litellm.proxy._experimental.mcp_server.server import (
+        call_mcp_tool,
+        global_mcp_server_manager,
+    )
+    from litellm.utils import Rules, function_setup
+
+    mock_server = MCPServer(
+        server_id="server-cost",
+        name="test_server",
+        alias="test_server",
+        server_name="test_server",
+        url="https://test-server.com/mcp",
+        transport=MCPTransport.http,
+        mcp_info={
+            "server_name": "test_server",
+            "mcp_server_cost_info": {"default_cost_per_query": 0.5},
+        },
+    )
+    user_auth = UserAPIKeyAuth(api_key="test-key", user_id="test-user")
+
+    start_time = datetime.now(timezone.utc)
+    logging_obj, _ = function_setup(
+        original_function="call_mcp_tool",
+        rules_obj=Rules(),
+        start_time=start_time,
+        litellm_call_id=str(uuid.uuid4()),
+        name="test_server-send_email",
+        arguments={"x": 1},
+    )
+
+    error_result = CallToolResult(
+        content=[TextContent(type="text", text="Error: denied")], isError=True
+    )
+
+    with (
+        patch.object(
+            global_mcp_server_manager,
+            "get_allowed_mcp_servers",
+            new_callable=AsyncMock,
+            return_value=[mock_server.server_id],
+        ),
+        patch.object(
+            global_mcp_server_manager,
+            "get_mcp_server_by_id",
+            return_value=mock_server,
+        ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._get_allowed_mcp_servers_from_mcp_server_names",
+            new_callable=AsyncMock,
+            return_value=[mock_server],
+        ),
+        patch.object(
+            global_mcp_server_manager,
+            "_get_mcp_server_from_tool_name",
+            return_value=mock_server,
+        ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server.global_mcp_tool_registry"
+        ) as mock_registry,
+        patch(
+            "litellm.proxy._experimental.mcp_server.server._handle_managed_mcp_tool",
+            new_callable=AsyncMock,
+            return_value=error_result,
+        ),
+        patch(
+            "litellm.proxy._experimental.mcp_server.server.MCPRequestHandler.is_tool_allowed",
+            return_value=True,
+        ),
+    ):
+        mock_registry.get_tool.return_value = None
+        response = await call_mcp_tool(
+            name="test_server-send_email",
+            arguments={"x": 1},
+            user_api_key_auth=user_auth,
+            mcp_servers=["test_server"],
+            litellm_logging_obj=logging_obj,
+        )
+
+    assert response.isError is True
+    slp = logging_obj.model_call_details["standard_logging_object"]
+    assert slp["status"] == "failure"
+    assert slp["response_cost"] == 0.5
+
+
 @pytest.mark.asyncio
 async def test_get_tools_from_mcp_servers_logs_list_tools_to_spendlogs_when_enabled():
     """