From 970277c171bd5532fefeeabaa1cfcb53939bf3f6 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sun, 28 Jun 2026 21:32:44 -0500 Subject: [PATCH 01/11] fix(mcp): emit standard_logging_object on /mcp/ JSON-RPC protocol-level rejections (#28929) --- .../proxy/_experimental/mcp_server/server.py | 195 ++++++++++++++ tests/mcp_tests/test_mcp_logging.py | 248 ++++++++++++++++++ 2 files changed, 443 insertions(+) diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index 4b55510a629..9b9125a2033 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -2736,6 +2736,162 @@ if MCP_AVAILABLE: mcp_session_id=session_id, ) + def _jsonrpc_rejection_reason(error_code: Optional[int]) -> str: + """Map a JSON-RPC error code to a stable, low-cardinality reason string + for spend-log metadata. Unknown codes degrade to ``protocol_error``.""" + return { + -32700: "parse_error", + -32600: "invalid_request", + -32601: "unknown_method", + -32602: "malformed_params", + -32603: "internal_error", + }.get(error_code, "protocol_error") + + async def _log_mcp_protocol_rejection( + request_method: Optional[str], + params: Dict[str, Any], + jsonrpc_error: Dict[str, Any], + request_id: Any, + user_api_key_auth: Optional[UserAPIKeyAuth], + raw_headers: Optional[Dict[str, str]], + start_time: datetime, + ) -> None: + """Emit a ``standard_logging_object`` for a JSON-RPC request rejected at + the protocol layer by the MCP SDK (e.g. unknown method, malformed + params). Without this, such rejections produce no spend-log record at + all — see https://github.com/BerriAI/litellm/issues/28929. + + Purely additive and best-effort: any failure here is swallowed so a + logging/serialization error can never alter or drop the client response. + Reuses the same ``post_call_failure_hook`` path already used by + ``call_mcp_tool`` and ``list_mcp_tools``. + """ + try: + from litellm.proxy.guardrails.tool_name_extraction import ( + _extract_mcp_tool_names, + ) + from litellm.proxy.proxy_server import proxy_logging_obj + + if proxy_logging_obj is None or user_api_key_auth is None: + return + + tool_names = _extract_mcp_tool_names(params or {}) + error_code = jsonrpc_error.get("code") + error_message = jsonrpc_error.get("message") or "MCP JSON-RPC protocol error" + + spend_logs_metadata: Dict[str, Any] = { + "mcp_operation": request_method, + "rejection_reason": _jsonrpc_rejection_reason(error_code), + "jsonrpc_error_code": error_code, + } + if tool_names: + spend_logs_metadata["mcp_tool_name"] = tool_names[0] + + request_data: Dict[str, Any] = { + "model": "MCP: protocol_error", + "call_type": CallTypes.call_mcp_tool.value, + "litellm_call_id": str(uuid.uuid4()), + "litellm_trace_id": get_chain_id_from_headers(raw_headers), + "metadata": { + "spend_logs_metadata": spend_logs_metadata, + }, + "input": [ + { + "role": "system", + # Keep message content a string: the downstream + # standard-logging builder may run token-counting / + # serialization over messages, and a non-string content + # can raise there (silently swallowed below -> the very + # record this fix emits would be dropped). The structured + # fields live in spend_logs_metadata above. + "content": json.dumps( + { + "mcp_operation": request_method, + "jsonrpc_id": request_id, + "jsonrpc_error_code": error_code, + } + ), + } + ], + } + + LiteLLMProxyRequestSetup.add_user_api_key_auth_to_request_metadata( + data=request_data, + user_api_key_dict=user_api_key_auth, + _metadata_variable_name="metadata", + ) + + await proxy_logging_obj.post_call_failure_hook( + request_data=request_data, + original_exception=Exception(error_message), + user_api_key_dict=user_api_key_auth, + route="/mcp/protocol_error", + traceback_str="", + ) + except Exception as log_exc: + verbose_logger.debug( + "MCP protocol-error logging failed (continuing): %s", log_exc + ) + + def _wrap_send_for_protocol_error_logging( + inner_send: Send, + request_method: Optional[str], + params: Dict[str, Any], + request_id: Any, + user_api_key_auth: Optional[UserAPIKeyAuth], + raw_headers: Optional[Dict[str, str]], + start_time: datetime, + ) -> Send: + """Wrap an ASGI ``send`` so a JSON-RPC error response emitted by the MCP + SDK (an ``error`` envelope with no ``result``) produces exactly one + ``standard_logging_object``. Only JSON-RPC *requests* (carrying a + ``method``) are loggable; everything else is forwarded untouched. + + The wrapper always forwards the original message FIRST and never mutates + it, so the client response is byte-for-byte unchanged on every path. + """ + if request_method is None: + return inner_send + + already_logged = {"done": False} + + async def _logging_send(message: Message) -> None: + await inner_send(message) + if already_logged["done"]: + return + try: + if message.get("type") != "http.response.body": + return + raw_body = message.get("body") or b"" + if not raw_body: + return + payload = json.loads(raw_body) + if ( + isinstance(payload, dict) + and "error" in payload + and "result" not in payload + ): + already_logged["done"] = True + await _log_mcp_protocol_rejection( + request_method=request_method, + params=params, + jsonrpc_error=payload.get("error") or {}, + request_id=request_id, + user_api_key_auth=user_api_key_auth, + raw_headers=raw_headers, + start_time=start_time, + ) + except (json.JSONDecodeError, TypeError, ValueError): + # Non-JSON / streaming chunk — nothing to log here. + return + except Exception as wrap_exc: # never break the response on logging + verbose_logger.debug( + "MCP protocol-error send wrapper failed (continuing): %s", + wrap_exc, + ) + + return _logging_send + async def _handle_managed_mcp_tool( server_name: str, name: str, @@ -3644,6 +3800,33 @@ if MCP_AVAILABLE: ) -> None: _increment_active_request_session(initialized_session_id) + # Parse the (already-peeked) JSON-RPC request so we can emit a + # standard_logging_object if the MCP SDK rejects it at the protocol + # layer (unknown method / malformed params). The SDK writes that + # error directly to ``send``, bypassing call_mcp_tool's logging, so + # we wrap ``send`` below. See issue #28929. + _protocol_log_method: Optional[str] = None + _protocol_log_params: Dict[str, Any] = {} + _protocol_log_id: Any = None + if body and request_method == "POST" and not is_jsonrpc_response: + try: + _req = json.loads(body) + if ( + isinstance(_req, dict) + and _req.get("jsonrpc") == "2.0" + and isinstance(_req.get("method"), str) + ): + _protocol_log_method = _req.get("method") + _protocol_log_id = _req.get("id") + _raw_params = _req.get("params") + if isinstance(_raw_params, dict): + _protocol_log_params = _raw_params + except (json.JSONDecodeError, TypeError): + # Truncated/oversized peek — skip protocol-error logging for + # this request rather than risk a wrong tool-name attribution. + _protocol_log_method = None + _protocol_log_start_time = datetime.now() + async def _dispatch() -> None: auth_user = _set_or_update_auth_context( user_api_key_auth=user_api_key_auth, @@ -3658,6 +3841,18 @@ if MCP_AVAILABLE: copy_existing_session_auth_context=is_initialize, ) local_send = send + # Additively observe protocol-level JSON-RPC rejections (#28929) + # without touching the success path. Applied first so the + # stateful-session wrapper (below) still wraps the outermost send. + local_send = _wrap_send_for_protocol_error_logging( + local_send, + request_method=_protocol_log_method, + params=_protocol_log_params, + request_id=_protocol_log_id, + user_api_key_auth=user_api_key_auth, + raw_headers=raw_headers, + start_time=_protocol_log_start_time, + ) if use_stateful and is_initialize: local_send = _wrap_send_with_stateful_session_auth_context( local_send, diff --git a/tests/mcp_tests/test_mcp_logging.py b/tests/mcp_tests/test_mcp_logging.py index 55b49aa0d29..02a6ddca46a 100644 --- a/tests/mcp_tests/test_mcp_logging.py +++ b/tests/mcp_tests/test_mcp_logging.py @@ -446,3 +446,251 @@ async def test_mcp_tool_call_hook(): logged_standard_logging_payload is not None ), "Standard logging payload should not be None" assert logged_standard_logging_payload["response_cost"] == 1.42 + + +# --------------------------------------------------------------------------- +# Tests for #28929: standard_logging_object on /mcp/ JSON-RPC protocol-level +# rejections (unknown method / malformed params). These exercise the additive +# send-wrapper + helper added to handle_streamable_http_mcp's dispatch path. +# --------------------------------------------------------------------------- +import json +from datetime import datetime + +from litellm.proxy._experimental.mcp_server.server import ( + _jsonrpc_rejection_reason, + _log_mcp_protocol_rejection, + _wrap_send_for_protocol_error_logging, +) +from litellm.types.utils import CallTypes + + +def _jsonrpc_error_body(code, message="boom", _id=1): + return json.dumps( + {"jsonrpc": "2.0", "id": _id, "error": {"code": code, "message": message}} + ).encode("utf-8") + + +def _jsonrpc_result_body(_id=1): + return json.dumps({"jsonrpc": "2.0", "id": _id, "result": {"ok": True}}).encode( + "utf-8" + ) + + +def test_mcp_jsonrpc_rejection_reason_mapping(): + assert _jsonrpc_rejection_reason(-32601) == "unknown_method" + assert _jsonrpc_rejection_reason(-32602) == "malformed_params" + assert _jsonrpc_rejection_reason(-32700) == "parse_error" + assert _jsonrpc_rejection_reason(-32600) == "invalid_request" + # Unknown / None codes degrade to a generic, low-cardinality reason. + assert _jsonrpc_rejection_reason(12345) == "protocol_error" + assert _jsonrpc_rejection_reason(None) == "protocol_error" + + +@pytest.mark.asyncio +async def test_mcp_protocol_rejection_emits_failure_record(): + """A JSON-RPC error response (no result) must produce exactly one + post_call_failure_hook call carrying a standard-logging-shaped request_data + with status-bearing failure semantics, call_type=call_mcp_tool, and the + parsed rejection metadata (#28929).""" + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") + + sent_messages = [] + + async def fake_send(message): + sent_messages.append(message) + + captured = {} + + class _FakeProxyLogging: + async def post_call_failure_hook(self, **kwargs): + captured.update(kwargs) + + with patch( + "litellm.proxy.proxy_server.proxy_logging_obj", _FakeProxyLogging() + ): + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "totally_made_up", "arguments": {}}, + request_id=7, + user_api_key_auth=user_auth, + raw_headers={}, + start_time=datetime.now(), + ) + body = _jsonrpc_error_body(-32602, "Invalid request parameters", _id=7) + await wrapped({"type": "http.response.start", "status": 200, "headers": []}) + await wrapped({"type": "http.response.body", "body": body}) + + # Response forwarded byte-for-byte and unchanged. + assert any( + m.get("type") == "http.response.body" and m.get("body") == body + for m in sent_messages + ), "original response bytes must be forwarded unchanged" + + # Exactly one failure record produced with the expected shape. + assert captured, "post_call_failure_hook should have been called" + request_data = captured["request_data"] + assert request_data["call_type"] == CallTypes.call_mcp_tool.value + assert captured["route"] == "/mcp/protocol_error" + spend_meta = request_data["metadata"]["spend_logs_metadata"] + assert spend_meta["mcp_operation"] == "tools/call" + assert spend_meta["rejection_reason"] == "malformed_params" + assert spend_meta["jsonrpc_error_code"] == -32602 + assert spend_meta["mcp_tool_name"] == "totally_made_up" + + +@pytest.mark.asyncio +async def test_mcp_protocol_unknown_method_reason(): + """An unknown-method (-32601) rejection is logged with the right reason.""" + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") + + async def fake_send(message): + pass + + captured = {} + + class _FakeProxyLogging: + async def post_call_failure_hook(self, **kwargs): + captured.update(kwargs) + + with patch( + "litellm.proxy.proxy_server.proxy_logging_obj", _FakeProxyLogging() + ): + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/totally_made_up", + params={}, + request_id=1, + user_api_key_auth=user_auth, + raw_headers={}, + start_time=datetime.now(), + ) + await wrapped( + { + "type": "http.response.body", + "body": _jsonrpc_error_body(-32601, "Method not found"), + } + ) + + assert captured, "unknown method should be logged" + spend_meta = captured["request_data"]["metadata"]["spend_logs_metadata"] + assert spend_meta["rejection_reason"] == "unknown_method" + # No tool name parsable from empty params -> key omitted. + assert "mcp_tool_name" not in spend_meta + + +@pytest.mark.asyncio +async def test_mcp_protocol_success_path_not_logged(): + """A JSON-RPC success response (has result) must NOT emit a failure + record, and the response must still be forwarded unchanged.""" + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") + + sent_messages = [] + + async def fake_send(message): + sent_messages.append(message) + + call_count = {"n": 0} + + class _FakeProxyLogging: + async def post_call_failure_hook(self, **kwargs): + call_count["n"] += 1 + + with patch( + "litellm.proxy.proxy_server.proxy_logging_obj", _FakeProxyLogging() + ): + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add", "arguments": {}}, + request_id=1, + user_api_key_auth=user_auth, + raw_headers={}, + start_time=datetime.now(), + ) + body = _jsonrpc_result_body(_id=1) + await wrapped({"type": "http.response.body", "body": body}) + + assert call_count["n"] == 0, "success path must not be logged as a failure" + assert any(m.get("body") == body for m in sent_messages) + + +@pytest.mark.asyncio +async def test_mcp_protocol_wrapper_noop_when_not_a_request(): + """When the POST carries no JSON-RPC method (request_method is None) the + wrapper returns the original send untouched (no logging seam at all).""" + + async def fake_send(message): + pass + + returned = _wrap_send_for_protocol_error_logging( + fake_send, + request_method=None, + params={}, + request_id=None, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + assert returned is fake_send, "non-request POSTs must not be wrapped" + + +@pytest.mark.asyncio +async def test_mcp_protocol_logging_failure_does_not_break_response(): + """If the logging callback raises, the wrapper must swallow it and still + forward the client response (observability must never drop a response).""" + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") + + sent_messages = [] + + async def fake_send(message): + sent_messages.append(message) + + class _ExplodingProxyLogging: + async def post_call_failure_hook(self, **kwargs): + raise RuntimeError("logging backend down") + + with patch( + "litellm.proxy.proxy_server.proxy_logging_obj", _ExplodingProxyLogging() + ): + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=user_auth, + raw_headers={}, + start_time=datetime.now(), + ) + body = _jsonrpc_error_body(-32602) + # Must not raise. + await wrapped({"type": "http.response.body", "body": body}) + + assert any(m.get("body") == body for m in sent_messages), ( + "response must be forwarded even when logging raises" + ) + + +@pytest.mark.asyncio +async def test_mcp_log_protocol_rejection_no_auth_is_noop(): + """The helper is a no-op (no exception) when there is no user auth.""" + + class _FakeProxyLogging: + def __init__(self): + self.called = False + + async def post_call_failure_hook(self, **kwargs): + self.called = True + + fake = _FakeProxyLogging() + with patch("litellm.proxy.proxy_server.proxy_logging_obj", fake): + await _log_mcp_protocol_rejection( + request_method="tools/call", + params={"name": "x"}, + jsonrpc_error={"code": -32602, "message": "bad"}, + request_id=1, + user_api_key_auth=None, + raw_headers={}, + start_time=datetime.now(), + ) + assert fake.called is False From 39239f6532aa067a97002c1cb46d4eddb480acbf Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sun, 28 Jun 2026 21:53:51 -0500 Subject: [PATCH 02/11] chore(mcp): ruff format server.py to satisfy lint --- litellm/proxy/_experimental/mcp_server/server.py | 16 +++------------- 1 file changed, 3 insertions(+), 13 deletions(-) diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index 9b9125a2033..17cc5bc402f 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -2829,9 +2829,7 @@ if MCP_AVAILABLE: traceback_str="", ) except Exception as log_exc: - verbose_logger.debug( - "MCP protocol-error logging failed (continuing): %s", log_exc - ) + verbose_logger.debug("MCP protocol-error logging failed (continuing): %s", log_exc) def _wrap_send_for_protocol_error_logging( inner_send: Send, @@ -2866,11 +2864,7 @@ if MCP_AVAILABLE: if not raw_body: return payload = json.loads(raw_body) - if ( - isinstance(payload, dict) - and "error" in payload - and "result" not in payload - ): + if isinstance(payload, dict) and "error" in payload and "result" not in payload: already_logged["done"] = True await _log_mcp_protocol_rejection( request_method=request_method, @@ -3811,11 +3805,7 @@ if MCP_AVAILABLE: if body and request_method == "POST" and not is_jsonrpc_response: try: _req = json.loads(body) - if ( - isinstance(_req, dict) - and _req.get("jsonrpc") == "2.0" - and isinstance(_req.get("method"), str) - ): + if isinstance(_req, dict) and _req.get("jsonrpc") == "2.0" and isinstance(_req.get("method"), str): _protocol_log_method = _req.get("method") _protocol_log_id = _req.get("id") _raw_params = _req.get("params") From 9b4868b6fff4c16f4e74534a222bee73a5eba721 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sun, 28 Jun 2026 23:03:19 -0500 Subject: [PATCH 03/11] fix(mcp): emit real standard_logging_object + address review (client-error class, bounded parse, batch) --- .../proxy/_experimental/mcp_server/server.py | 143 +++- tests/mcp_tests/test_mcp_logging.py | 628 +++++++++++++----- 2 files changed, 577 insertions(+), 194 deletions(-) diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index 17cc5bc402f..63aef10f4b6 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -98,6 +98,10 @@ _MAX_STATEFUL_SESSIONS_PER_OWNER = 100 # prevents an authenticated client from forcing the proxy to buffer an # arbitrarily large body just to make a routing decision. _MCP_ROUTING_PEEK_MAX_BYTES = 4096 +# Upper bound on a response body the protocol-error send wrapper will parse to +# detect a JSON-RPC rejection. Rejection envelopes are tiny; anything larger is +# a permitted tool result and must not be re-deserialized after it was sent. +_MCP_PROTOCOL_ERROR_PEEK_MAX_BYTES = 64 * 1024 def _invalidate_byok_cred_cache(user_id: str, server_id: str) -> None: @@ -2747,39 +2751,87 @@ if MCP_AVAILABLE: -32603: "internal_error", }.get(error_code, "protocol_error") + def _parse_jsonrpc_request_for_logging( + body: bytes, + ) -> tuple[Optional[str], dict[str, Any], Any]: + """Parse a peeked JSON-RPC request body into ``(method, params, id)`` for + protocol-rejection logging (#28929). + + - A single request object yields its ``method`` (only when ``jsonrpc`` + is ``"2.0"`` and ``method`` is a string), ``params`` (only if a dict), + and ``id``. + - A top-level array (JSON-RPC batch) yields the synthetic method + ``"batch"``: the MCP SDK validates the whole array as one message and, + if it can't, emits a single error response (it does NOT process + elements individually), so we record the one rejection without + attributing a tool name. + - A truncated/oversized peek or any non-request shape yields + ``(None, {}, None)`` so the caller installs no logging seam (a wrong + tool-name attribution is worse than no record). + """ + method: Optional[str] = None + params: dict[str, Any] = {} + request_id: Any = None + try: + req = json.loads(body) + except (json.JSONDecodeError, TypeError): + return None, {}, None + if isinstance(req, dict): + method_val = req.get("method") + if req.get("jsonrpc") == "2.0" and isinstance(method_val, str): + method = method_val + request_id = req.get("id") + raw_params = req.get("params") + if isinstance(raw_params, dict): + params = raw_params + elif isinstance(req, list): + method = "batch" + return method, params, request_id + async def _log_mcp_protocol_rejection( request_method: Optional[str], - params: Dict[str, Any], - jsonrpc_error: Dict[str, Any], + params: dict[str, Any], + jsonrpc_error: dict[str, Any], request_id: Any, user_api_key_auth: Optional[UserAPIKeyAuth], - raw_headers: Optional[Dict[str, str]], + raw_headers: Optional[dict[str, str]], start_time: datetime, ) -> None: - """Emit a ``standard_logging_object`` for a JSON-RPC request rejected at - the protocol layer by the MCP SDK (e.g. unknown method, malformed - params). Without this, such rejections produce no spend-log record at - all — see https://github.com/BerriAI/litellm/issues/28929. + """Emit a real ``standard_logging_object`` for a JSON-RPC request + rejected at the protocol layer by the MCP SDK (e.g. unknown method, + malformed params). Without this, such rejections produce no spend-log + record at all — see https://github.com/BerriAI/litellm/issues/28929. + + This builds a genuine ``LiteLLMLoggingObj`` via ``function_setup`` and + calls its ``async_failure_handler`` — the same mechanism + ``list_mcp_tools`` / ``call_mcp_tool`` use — so the + ``standard_logging_object`` is actually synthesized and dispatched to + registered callbacks. (``proxy_logging_obj.post_call_failure_hook`` is + NOT used here: it only synthesizes a ``standard_logging_object`` for + proxy-only *LLM-API* errors and requires a pre-existing + ``litellm_logging_obj`` in ``request_data`` — neither holds for a + protocol-level MCP rejection, so it would deliver a raw dict and no + record. Driving ``async_failure_handler`` directly also means a + malformed *client* request never fires the ``llm_exceptions`` alert that + lives in ``post_call_failure_hook`` -- these are 4xx client errors, not + LLM API failures.) Purely additive and best-effort: any failure here is swallowed so a logging/serialization error can never alter or drop the client response. - Reuses the same ``post_call_failure_hook`` path already used by - ``call_mcp_tool`` and ``list_mcp_tools``. """ try: from litellm.proxy.guardrails.tool_name_extraction import ( _extract_mcp_tool_names, ) - from litellm.proxy.proxy_server import proxy_logging_obj - if proxy_logging_obj is None or user_api_key_auth is None: + if user_api_key_auth is None: return tool_names = _extract_mcp_tool_names(params or {}) error_code = jsonrpc_error.get("code") error_message = jsonrpc_error.get("message") or "MCP JSON-RPC protocol error" - spend_logs_metadata: Dict[str, Any] = { + spend_logs_metadata: dict[str, Any] = { "mcp_operation": request_method, "rejection_reason": _jsonrpc_rejection_reason(error_code), "jsonrpc_error_code": error_code, @@ -2787,7 +2839,7 @@ if MCP_AVAILABLE: if tool_names: spend_logs_metadata["mcp_tool_name"] = tool_names[0] - request_data: Dict[str, Any] = { + request_data: dict[str, Any] = { "model": "MCP: protocol_error", "call_type": CallTypes.call_mcp_tool.value, "litellm_call_id": str(uuid.uuid4()), @@ -2821,23 +2873,40 @@ if MCP_AVAILABLE: _metadata_variable_name="metadata", ) - await proxy_logging_obj.post_call_failure_hook( - request_data=request_data, - original_exception=Exception(error_message), - user_api_key_dict=user_api_key_auth, - route="/mcp/protocol_error", - traceback_str="", + # Build a genuine logging object (mirrors list_mcp_tools at the + # ``function_setup`` call site) and drive its failure handler, which + # synthesizes ``model_call_details["standard_logging_object"]`` and + # dispatches it to every registered async failure callback. A + # function_setup failure is caught by the outer best-effort handler. + litellm_logging_obj, _ = function_setup( + original_function="call_mcp_tool", + rules_obj=Rules(), + start_time=start_time, + **request_data, ) - except Exception as log_exc: + + if litellm_logging_obj is None: + return + + litellm_logging_obj.call_type = CallTypes.call_mcp_tool.value + litellm_logging_obj.model = "MCP: protocol_error" + + await litellm_logging_obj.async_failure_handler( + Exception(error_message), + traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG), + start_time, + datetime.now(), + ) + except Exception as log_exc: # noqa: BLE001 # best-effort: logging must never break the MCP response verbose_logger.debug("MCP protocol-error logging failed (continuing): %s", log_exc) def _wrap_send_for_protocol_error_logging( inner_send: Send, request_method: Optional[str], - params: Dict[str, Any], + params: dict[str, Any], request_id: Any, user_api_key_auth: Optional[UserAPIKeyAuth], - raw_headers: Optional[Dict[str, str]], + raw_headers: Optional[dict[str, str]], start_time: datetime, ) -> Send: """Wrap an ASGI ``send`` so a JSON-RPC error response emitted by the MCP @@ -2863,6 +2932,15 @@ if MCP_AVAILABLE: raw_body = message.get("body") or b"" if not raw_body: return + # Bound the work: a JSON-RPC protocol-error envelope is tiny. + # Skip the full deserialization for anything larger than the cap + # (a permitted tool result) and only parse when the bytes look + # like an error envelope. This avoids re-deserializing every + # successful response body after it has already been sent. + if len(raw_body) > _MCP_PROTOCOL_ERROR_PEEK_MAX_BYTES: + return + if b'"error"' not in raw_body or b'"result"' in raw_body: + return payload = json.loads(raw_body) if isinstance(payload, dict) and "error" in payload and "result" not in payload: already_logged["done"] = True @@ -2878,7 +2956,7 @@ if MCP_AVAILABLE: except (json.JSONDecodeError, TypeError, ValueError): # Non-JSON / streaming chunk — nothing to log here. return - except Exception as wrap_exc: # never break the response on logging + except Exception as wrap_exc: # noqa: BLE001 # best-effort: never break the response on logging verbose_logger.debug( "MCP protocol-error send wrapper failed (continuing): %s", wrap_exc, @@ -3800,21 +3878,14 @@ if MCP_AVAILABLE: # error directly to ``send``, bypassing call_mcp_tool's logging, so # we wrap ``send`` below. See issue #28929. _protocol_log_method: Optional[str] = None - _protocol_log_params: Dict[str, Any] = {} + _protocol_log_params: dict[str, Any] = {} _protocol_log_id: Any = None if body and request_method == "POST" and not is_jsonrpc_response: - try: - _req = json.loads(body) - if isinstance(_req, dict) and _req.get("jsonrpc") == "2.0" and isinstance(_req.get("method"), str): - _protocol_log_method = _req.get("method") - _protocol_log_id = _req.get("id") - _raw_params = _req.get("params") - if isinstance(_raw_params, dict): - _protocol_log_params = _raw_params - except (json.JSONDecodeError, TypeError): - # Truncated/oversized peek — skip protocol-error logging for - # this request rather than risk a wrong tool-name attribution. - _protocol_log_method = None + ( + _protocol_log_method, + _protocol_log_params, + _protocol_log_id, + ) = _parse_jsonrpc_request_for_logging(body) _protocol_log_start_time = datetime.now() async def _dispatch() -> None: diff --git a/tests/mcp_tests/test_mcp_logging.py b/tests/mcp_tests/test_mcp_logging.py index 02a6ddca46a..dc559e18e86 100644 --- a/tests/mcp_tests/test_mcp_logging.py +++ b/tests/mcp_tests/test_mcp_logging.py @@ -6,9 +6,7 @@ from typing import Optional from unittest.mock import AsyncMock, patch -sys.path.insert( - 0, os.path.abspath("../../..") -) # Adds the parent directory to the system path +sys.path.insert(0, os.path.abspath("../../..")) # Adds the parent directory to the system path import litellm from litellm.types.utils import StandardLoggingPayload from litellm.integrations.custom_logger import CustomLogger @@ -54,9 +52,7 @@ def _set_authorized_user(server_ids): async def test_mcp_cost_tracking(): # Create a mock tool call result litellm.logging_callback_manager._reset_all_callbacks() - mock_result = CallToolResult( - content=[TextContent(type="text", text="Test response")], isError=False - ) + mock_result = CallToolResult(content=[TextContent(type="text", text="Test response")], isError=False) # Create a mock MCPClient mock_client = AsyncMock() @@ -117,7 +113,6 @@ async def test_mcp_cost_tracking(): local_mcp_server_manager, ), ): - _set_authorized_user(local_mcp_server_manager.get_all_mcp_server_ids()) print( @@ -126,12 +121,10 @@ async def test_mcp_cost_tracking(): ) # Manually add the tool mapping to ensure it's available (since mocking might not capture it properly) - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping[ - "add_tools" - ] = "zapier_gmail_server" - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping[ - "zapier_gmail_server-add_tools" - ] = "zapier_gmail_server" + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["add_tools"] = "zapier_gmail_server" + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["zapier_gmail_server-add_tools"] = ( + "zapier_gmail_server" + ) # Call mcp tool response = await mcp_server_tool_call( @@ -151,9 +144,7 @@ async def test_mcp_cost_tracking(): if isinstance(response, CallToolResult): response_list = response.content else: - response_list = list( - response - ) # Convert iterable to list for backward compatibility + response_list = list(response) # Convert iterable to list for backward compatibility assert len(response_list) == 1 assert isinstance(response_list[0], TextContent) assert response_list[0].text == "Test response" @@ -164,9 +155,7 @@ async def test_mcp_cost_tracking(): ###### # verify response cost is 1.2 as set on default_cost_per_query # Critical - the cost is tracked as $1.2 - assert ( - logged_standard_logging_payload is not None - ), "Standard logging payload should not be None" + assert logged_standard_logging_payload is not None, "Standard logging payload should not be None" assert logged_standard_logging_payload["response_cost"] == 1.2 @@ -175,9 +164,7 @@ async def test_mcp_cost_tracking_per_tool(): """Test that individual tool costs are tracked correctly when tool_name_to_cost_per_query is configured""" # Create a mock tool call result litellm.logging_callback_manager._reset_all_callbacks() - mock_result = CallToolResult( - content=[TextContent(type="text", text="Test response")], isError=False - ) + mock_result = CallToolResult(content=[TextContent(type="text", text="Test response")], isError=False) # Create a mock MCPClient mock_client = AsyncMock() @@ -240,18 +227,10 @@ async def test_mcp_cost_tracking_per_tool(): await local_mcp_server_manager._initialize_tool_name_to_mcp_server_name_mapping() # Manually add the tool mapping to ensure it's available (since mocking might not capture it properly) - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping[ - "expensive_tool" - ] = "test_server" - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping[ - "test_server-expensive_tool" - ] = "test_server" - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["cheap_tool"] = ( - "test_server" - ) - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping[ - "test_server-cheap_tool" - ] = "test_server" + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["expensive_tool"] = "test_server" + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["test_server-expensive_tool"] = "test_server" + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["cheap_tool"] = "test_server" + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["test_server-cheap_tool"] = "test_server" # Patch the global manager in both modules where it's used with ( @@ -264,7 +243,6 @@ async def test_mcp_cost_tracking_per_tool(): local_mcp_server_manager, ), ): - _set_authorized_user(local_mcp_server_manager.get_all_mcp_server_ids()) print( @@ -282,14 +260,10 @@ async def test_mcp_cost_tracking_per_tool(): await asyncio.sleep(2) logged_standard_logging_payload_1 = test_logger.standard_logging_payload - print( - "logged_standard_logging_payload_1", logged_standard_logging_payload_1 - ) + print("logged_standard_logging_payload_1", logged_standard_logging_payload_1) # Verify expensive tool cost - assert ( - logged_standard_logging_payload_1 is not None - ), "Standard logging payload 1 should not be None" + assert logged_standard_logging_payload_1 is not None, "Standard logging payload 1 should not be None" assert logged_standard_logging_payload_1["response_cost"] == 5.0 # Reset logger for second test @@ -305,14 +279,10 @@ async def test_mcp_cost_tracking_per_tool(): await asyncio.sleep(2) logged_standard_logging_payload_2 = test_logger.standard_logging_payload - print( - "logged_standard_logging_payload_2", logged_standard_logging_payload_2 - ) + print("logged_standard_logging_payload_2", logged_standard_logging_payload_2) # Verify cheap tool cost - assert ( - logged_standard_logging_payload_2 is not None - ), "Standard logging payload 2 should not be None" + assert logged_standard_logging_payload_2 is not None, "Standard logging payload 2 should not be None" assert logged_standard_logging_payload_2["response_cost"] == 0.1 # Add basic response assertions @@ -356,9 +326,7 @@ class MCPLoggerHook(CustomLogger): async def test_mcp_tool_call_hook(): # Create a mock tool call result litellm.logging_callback_manager._reset_all_callbacks() - mock_result = CallToolResult( - content=[TextContent(type="text", text="Test response")], isError=False - ) + mock_result = CallToolResult(content=[TextContent(type="text", text="Test response")], isError=False) # Create a mock MCPClient mock_client = AsyncMock() @@ -404,12 +372,10 @@ async def test_mcp_tool_call_hook(): await local_mcp_server_manager._initialize_tool_name_to_mcp_server_name_mapping() # Manually add the tool mapping to ensure it's available (since mocking might not capture it properly) - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["add_tools"] = ( + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["add_tools"] = "zapier_gmail_server" + local_mcp_server_manager.tool_name_to_mcp_server_name_mapping["zapier_gmail_server-add_tools"] = ( "zapier_gmail_server" ) - local_mcp_server_manager.tool_name_to_mcp_server_name_mapping[ - "zapier_gmail_server-add_tools" - ] = "zapier_gmail_server" # Patch the global manager in both modules where it's used with ( @@ -422,7 +388,6 @@ async def test_mcp_tool_call_hook(): local_mcp_server_manager, ), ): - _set_authorized_user(local_mcp_server_manager.get_all_mcp_server_ids()) print( @@ -442,9 +407,7 @@ async def test_mcp_tool_call_hook(): # check logged standard logging payload logged_standard_logging_payload = test_logger.standard_logging_payload print("logged_standard_logging_payload", logged_standard_logging_payload) - assert ( - logged_standard_logging_payload is not None - ), "Standard logging payload should not be None" + assert logged_standard_logging_payload is not None, "Standard logging payload should not be None" assert logged_standard_logging_payload["response_cost"] == 1.42 @@ -452,6 +415,11 @@ async def test_mcp_tool_call_hook(): # Tests for #28929: standard_logging_object on /mcp/ JSON-RPC protocol-level # rejections (unknown method / malformed params). These exercise the additive # send-wrapper + helper added to handle_streamable_http_mcp's dispatch path. +# +# The load-bearing assertion is that a REAL standard_logging_object is emitted +# to a registered failure callback (not merely that an internal hook was +# called) — the rejection path builds a genuine LiteLLMLoggingObj and drives +# its async_failure_handler, mirroring list_mcp_tools / call_mcp_tool. # --------------------------------------------------------------------------- import json from datetime import datetime @@ -459,21 +427,34 @@ from datetime import datetime from litellm.proxy._experimental.mcp_server.server import ( _jsonrpc_rejection_reason, _log_mcp_protocol_rejection, + _parse_jsonrpc_request_for_logging, _wrap_send_for_protocol_error_logging, ) from litellm.types.utils import CallTypes +class _ProtocolRejectionLogger(CustomLogger): + """Captures the standard_logging_object dispatched on the failure path.""" + + def __init__(self): + self.failure_payloads = [] + super().__init__() + + async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): + slo = kwargs.get("standard_logging_object", None) + self.failure_payloads.append(slo) + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + # Should never fire for a protocol rejection. + self.failure_payloads.append(("UNEXPECTED_SUCCESS", kwargs.get("standard_logging_object"))) + + def _jsonrpc_error_body(code, message="boom", _id=1): - return json.dumps( - {"jsonrpc": "2.0", "id": _id, "error": {"code": code, "message": message}} - ).encode("utf-8") + return json.dumps({"jsonrpc": "2.0", "id": _id, "error": {"code": code, "message": message}}).encode("utf-8") def _jsonrpc_result_body(_id=1): - return json.dumps({"jsonrpc": "2.0", "id": _id, "result": {"ok": True}}).encode( - "utf-8" - ) + return json.dumps({"jsonrpc": "2.0", "id": _id, "result": {"ok": True}}).encode("utf-8") def test_mcp_jsonrpc_rejection_reason_mapping(): @@ -487,27 +468,21 @@ def test_mcp_jsonrpc_rejection_reason_mapping(): @pytest.mark.asyncio -async def test_mcp_protocol_rejection_emits_failure_record(): - """A JSON-RPC error response (no result) must produce exactly one - post_call_failure_hook call carrying a standard-logging-shaped request_data - with status-bearing failure semantics, call_type=call_mcp_tool, and the - parsed rejection metadata (#28929).""" - user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") +async def test_mcp_protocol_rejection_emits_standard_logging_object(): + """A JSON-RPC error response (no result) must cause a REAL + standard_logging_object to be dispatched to a registered failure callback, + carrying the rejection metadata (#28929).""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") sent_messages = [] async def fake_send(message): sent_messages.append(message) - captured = {} - - class _FakeProxyLogging: - async def post_call_failure_hook(self, **kwargs): - captured.update(kwargs) - - with patch( - "litellm.proxy.proxy_server.proxy_logging_obj", _FakeProxyLogging() - ): + try: wrapped = _wrap_send_for_protocol_error_logging( fake_send, request_method="tools/call", @@ -520,42 +495,47 @@ async def test_mcp_protocol_rejection_emits_failure_record(): body = _jsonrpc_error_body(-32602, "Invalid request parameters", _id=7) await wrapped({"type": "http.response.start", "status": 200, "headers": []}) await wrapped({"type": "http.response.body", "body": body}) + # async_failure_handler may dispatch on the loop; give it a tick. + await asyncio.sleep(1) + finally: + litellm.callbacks = [] # Response forwarded byte-for-byte and unchanged. - assert any( - m.get("type") == "http.response.body" and m.get("body") == body - for m in sent_messages - ), "original response bytes must be forwarded unchanged" + assert any(m.get("type") == "http.response.body" and m.get("body") == body for m in sent_messages), ( + "original response bytes must be forwarded unchanged" + ) - # Exactly one failure record produced with the expected shape. - assert captured, "post_call_failure_hook should have been called" - request_data = captured["request_data"] - assert request_data["call_type"] == CallTypes.call_mcp_tool.value - assert captured["route"] == "/mcp/protocol_error" - spend_meta = request_data["metadata"]["spend_logs_metadata"] - assert spend_meta["mcp_operation"] == "tools/call" - assert spend_meta["rejection_reason"] == "malformed_params" - assert spend_meta["jsonrpc_error_code"] == -32602 - assert spend_meta["mcp_tool_name"] == "totally_made_up" + # A genuine, non-None standard_logging_object reached the callback. + non_none = [p for p in capture.failure_payloads if isinstance(p, dict)] + assert non_none, ( + f"expected a non-None standard_logging_object on the failure callback, got: {capture.failure_payloads!r}" + ) + slo = non_none[0] + assert slo.get("call_type") == CallTypes.call_mcp_tool.value + # status is a failure-bearing value (StandardLoggingPayloadStatus). + assert slo.get("status") == "failure" + # Rejection metadata threaded through spend_logs_metadata. + spend_meta = (slo.get("metadata") or {}).get("spend_logs_metadata") or {} + assert spend_meta.get("mcp_operation") == "tools/call" + assert spend_meta.get("rejection_reason") == "malformed_params" + assert spend_meta.get("jsonrpc_error_code") == -32602 + assert spend_meta.get("mcp_tool_name") == "totally_made_up" @pytest.mark.asyncio -async def test_mcp_protocol_unknown_method_reason(): - """An unknown-method (-32601) rejection is logged with the right reason.""" +async def test_mcp_protocol_unknown_method_emits_slo(): + """An unknown-method (-32601) rejection emits an SLO with the right + reason and no tool name (empty params).""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") async def fake_send(message): pass - captured = {} - - class _FakeProxyLogging: - async def post_call_failure_hook(self, **kwargs): - captured.update(kwargs) - - with patch( - "litellm.proxy.proxy_server.proxy_logging_obj", _FakeProxyLogging() - ): + try: wrapped = _wrap_send_for_protocol_error_logging( fake_send, request_method="tools/totally_made_up", @@ -571,34 +551,75 @@ async def test_mcp_protocol_unknown_method_reason(): "body": _jsonrpc_error_body(-32601, "Method not found"), } ) + await asyncio.sleep(1) + finally: + litellm.callbacks = [] - assert captured, "unknown method should be logged" - spend_meta = captured["request_data"]["metadata"]["spend_logs_metadata"] - assert spend_meta["rejection_reason"] == "unknown_method" - # No tool name parsable from empty params -> key omitted. + non_none = [p for p in capture.failure_payloads if isinstance(p, dict)] + assert non_none, f"expected an SLO, got: {capture.failure_payloads!r}" + spend_meta = (non_none[0].get("metadata") or {}).get("spend_logs_metadata") or {} + assert spend_meta.get("rejection_reason") == "unknown_method" assert "mcp_tool_name" not in spend_meta @pytest.mark.asyncio -async def test_mcp_protocol_success_path_not_logged(): - """A JSON-RPC success response (has result) must NOT emit a failure - record, and the response must still be forwarded unchanged.""" +async def test_mcp_protocol_batch_rejection_emits_slo(): + """A JSON-RPC batch (array body) the MCP SDK rejects as one error must + still produce one SLO, attributed to mcp_operation='batch' (#28929, + Codex batch finding).""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") + async def fake_send(message): + pass + + try: + # Mirrors how handle_streamable_http_mcp maps a top-level array to a + # synthetic 'batch' operation before wrapping send. + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="batch", + params={}, + request_id=None, + user_api_key_auth=user_auth, + raw_headers={}, + start_time=datetime.now(), + ) + await wrapped( + { + "type": "http.response.body", + "body": _jsonrpc_error_body(-32602, "Validation error", _id="server-error"), + } + ) + await asyncio.sleep(1) + finally: + litellm.callbacks = [] + + non_none = [p for p in capture.failure_payloads if isinstance(p, dict)] + assert non_none, f"expected a batch SLO, got: {capture.failure_payloads!r}" + spend_meta = (non_none[0].get("metadata") or {}).get("spend_logs_metadata") or {} + assert spend_meta.get("mcp_operation") == "batch" + assert spend_meta.get("rejection_reason") == "malformed_params" + + +@pytest.mark.asyncio +async def test_mcp_protocol_success_path_emits_no_failure_record(): + """A JSON-RPC success response (has result) must NOT emit a failure + record, and the response must still be forwarded unchanged.""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") sent_messages = [] async def fake_send(message): sent_messages.append(message) - call_count = {"n": 0} - - class _FakeProxyLogging: - async def post_call_failure_hook(self, **kwargs): - call_count["n"] += 1 - - with patch( - "litellm.proxy.proxy_server.proxy_logging_obj", _FakeProxyLogging() - ): + try: wrapped = _wrap_send_for_protocol_error_logging( fake_send, request_method="tools/call", @@ -610,8 +631,13 @@ async def test_mcp_protocol_success_path_not_logged(): ) body = _jsonrpc_result_body(_id=1) await wrapped({"type": "http.response.body", "body": body}) + await asyncio.sleep(1) + finally: + litellm.callbacks = [] - assert call_count["n"] == 0, "success path must not be logged as a failure" + assert capture.failure_payloads == [], ( + f"success path must not dispatch any failure record, got: {capture.failure_payloads!r}" + ) assert any(m.get("body") == body for m in sent_messages) @@ -637,53 +663,49 @@ async def test_mcp_protocol_wrapper_noop_when_not_a_request(): @pytest.mark.asyncio async def test_mcp_protocol_logging_failure_does_not_break_response(): - """If the logging callback raises, the wrapper must swallow it and still - forward the client response (observability must never drop a response).""" - user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") + """If logging-object construction raises, the wrapper must swallow it and + still forward the client response (observability must never drop a + response).""" + litellm.logging_callback_manager._reset_all_callbacks() + litellm.callbacks = [] + user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") sent_messages = [] async def fake_send(message): sent_messages.append(message) - class _ExplodingProxyLogging: - async def post_call_failure_hook(self, **kwargs): - raise RuntimeError("logging backend down") + try: + with patch( + "litellm.proxy._experimental.mcp_server.server.function_setup", + side_effect=RuntimeError("logging init down"), + ): + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=user_auth, + raw_headers={}, + start_time=datetime.now(), + ) + body = _jsonrpc_error_body(-32602) + # Must not raise. + await wrapped({"type": "http.response.body", "body": body}) + finally: + litellm.callbacks = [] - with patch( - "litellm.proxy.proxy_server.proxy_logging_obj", _ExplodingProxyLogging() - ): - wrapped = _wrap_send_for_protocol_error_logging( - fake_send, - request_method="tools/call", - params={"name": "add"}, - request_id=1, - user_api_key_auth=user_auth, - raw_headers={}, - start_time=datetime.now(), - ) - body = _jsonrpc_error_body(-32602) - # Must not raise. - await wrapped({"type": "http.response.body", "body": body}) - - assert any(m.get("body") == body for m in sent_messages), ( - "response must be forwarded even when logging raises" - ) + assert any(m.get("body") == body for m in sent_messages), "response must be forwarded even when logging init raises" @pytest.mark.asyncio async def test_mcp_log_protocol_rejection_no_auth_is_noop(): - """The helper is a no-op (no exception) when there is no user auth.""" - - class _FakeProxyLogging: - def __init__(self): - self.called = False - - async def post_call_failure_hook(self, **kwargs): - self.called = True - - fake = _FakeProxyLogging() - with patch("litellm.proxy.proxy_server.proxy_logging_obj", fake): + """The helper is a no-op (no exception, no SLO) when there is no user + auth.""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + try: await _log_mcp_protocol_rejection( request_method="tools/call", params={"name": "x"}, @@ -693,4 +715,294 @@ async def test_mcp_log_protocol_rejection_no_auth_is_noop(): raw_headers={}, start_time=datetime.now(), ) - assert fake.called is False + await asyncio.sleep(0.2) + finally: + litellm.callbacks = [] + assert capture.failure_payloads == [] + + +@pytest.mark.asyncio +async def test_mcp_protocol_wrapper_skips_oversized_body(): + """V2 guard: a response body larger than the parse cap is never + deserialized (a permitted tool result), so no SLO is emitted even if it + happens to contain an error-shaped substring.""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + sent = [] + + async def fake_send(message): + sent.append(message) + + # Build a body > 64 KiB that still contains the error marker bytes. + big = b'{"result": {"x": "' + (b"a" * (70 * 1024)) + b'"}, "error": 1}' + try: + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + await wrapped({"type": "http.response.body", "body": big}) + await asyncio.sleep(0.3) + finally: + litellm.callbacks = [] + + assert capture.failure_payloads == [], "oversized body must not be parsed/logged" + assert any(m.get("body") == big for m in sent) + + +@pytest.mark.asyncio +async def test_mcp_protocol_wrapper_skips_non_error_body(): + """V2 guard: the cheap byte pre-check skips bodies that don't look like an + error envelope (no '"error"', or a '"result"' present) without json.loads.""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + async def fake_send(message): + pass + + try: + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + # No "error" marker at all. + await wrapped({"type": "http.response.body", "body": b'{"jsonrpc":"2.0","id":1}'}) + # Both "error" and "result" present -> not a pure rejection. + await wrapped( + { + "type": "http.response.body", + "body": b'{"jsonrpc":"2.0","id":1,"result":{},"error":null}', + } + ) + await asyncio.sleep(0.3) + finally: + litellm.callbacks = [] + + assert capture.failure_payloads == [], "non-error bodies must not be logged" + + +@pytest.mark.asyncio +async def test_mcp_protocol_wrapper_ignores_non_body_messages_and_empty(): + """The wrapper forwards http.response.start and empty bodies without + attempting to log.""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + sent = [] + + async def fake_send(message): + sent.append(message) + + try: + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + await wrapped({"type": "http.response.start", "status": 200, "headers": []}) + await wrapped({"type": "http.response.body", "body": b""}) + await asyncio.sleep(0.2) + finally: + litellm.callbacks = [] + + assert capture.failure_payloads == [] + assert len(sent) == 2 # both forwarded + + +@pytest.mark.asyncio +async def test_mcp_protocol_wrapper_handles_malformed_error_body(): + """An error-looking but invalid-JSON body trips the JSONDecodeError branch + and is swallowed (no SLO, response still forwarded).""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + sent = [] + + async def fake_send(message): + sent.append(message) + + try: + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + # passes the byte pre-check ('"error"' present, no '"result"') but is + # not valid JSON. + body = b'{"error": this-is-not-json' + await wrapped({"type": "http.response.body", "body": body}) + await asyncio.sleep(0.2) + finally: + litellm.callbacks = [] + + assert capture.failure_payloads == [] + assert any(m.get("body") == body for m in sent) + + +@pytest.mark.asyncio +async def test_mcp_log_protocol_rejection_logging_obj_none_is_noop(monkeypatch): + """If function_setup yields no logging object, the helper returns without + raising (covers the None-guard branch).""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + import litellm.proxy._experimental.mcp_server.server as mcp_server_mod + + def _fake_function_setup(*args, **kwargs): + return None, {} + + monkeypatch.setattr(mcp_server_mod, "function_setup", _fake_function_setup) + try: + await _log_mcp_protocol_rejection( + request_method="tools/call", + params={"name": "x"}, + jsonrpc_error={"code": -32602, "message": "bad"}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + await asyncio.sleep(0.2) + finally: + litellm.callbacks = [] + + assert capture.failure_payloads == [] + + +@pytest.mark.asyncio +async def test_mcp_protocol_wrapper_logs_only_once_per_request(): + """already_logged short-circuit: a second error body on the same request + does not produce a second SLO (covers the early-return guard).""" + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + async def fake_send(message): + pass + + try: + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + await wrapped({"type": "http.response.body", "body": _jsonrpc_error_body(-32602)}) + await wrapped({"type": "http.response.body", "body": _jsonrpc_error_body(-32601)}) + await asyncio.sleep(1) + finally: + litellm.callbacks = [] + + non_none = [p for p in capture.failure_payloads if isinstance(p, dict)] + assert len(non_none) == 1, f"exactly one SLO expected, got {len(non_none)}" + + +@pytest.mark.asyncio +async def test_mcp_protocol_wrapper_swallows_unexpected_error(monkeypatch): + """A non-JSON unexpected error raised inside the detection block is caught + by the broad guard and never breaks the response (covers the BLE001 + branch).""" + litellm.logging_callback_manager._reset_all_callbacks() + litellm.callbacks = [] + + import litellm.proxy._experimental.mcp_server.server as mcp_server_mod + + async def _boom(**kwargs): + raise RuntimeError("unexpected") + + monkeypatch.setattr(mcp_server_mod, "_log_mcp_protocol_rejection", _boom) + + sent = [] + + async def fake_send(message): + sent.append(message) + + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/call", + params={"name": "add"}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + body = _jsonrpc_error_body(-32602) + # Must not raise despite _log_mcp_protocol_rejection blowing up. + await wrapped({"type": "http.response.body", "body": body}) + assert any(m.get("body") == body for m in sent) + + +def test_parse_jsonrpc_request_for_logging_single(): + method, params, rid = _parse_jsonrpc_request_for_logging( + b'{"jsonrpc":"2.0","method":"tools/call","id":7,"params":{"name":"x"}}' + ) + assert method == "tools/call" + assert params == {"name": "x"} + assert rid == 7 + + +def test_parse_jsonrpc_request_for_logging_non_string_method_skipped(): + # method present but not a string -> not loggable as a request. + method, params, rid = _parse_jsonrpc_request_for_logging(b'{"jsonrpc":"2.0","method":123,"id":1}') + assert method is None + assert params == {} + assert rid is None + + +def test_parse_jsonrpc_request_for_logging_non_dict_params(): + # params present but not a dict -> params stays empty, method still parsed. + method, params, rid = _parse_jsonrpc_request_for_logging( + b'{"jsonrpc":"2.0","method":"tools/call","id":1,"params":[1,2]}' + ) + assert method == "tools/call" + assert params == {} + + +def test_parse_jsonrpc_request_for_logging_missing_jsonrpc_version(): + method, params, rid = _parse_jsonrpc_request_for_logging(b'{"method":"tools/call","id":1}') + assert method is None + + +def test_parse_jsonrpc_request_for_logging_batch(): + method, params, rid = _parse_jsonrpc_request_for_logging(b'[{"jsonrpc":"2.0","method":"tools/call","id":1}]') + assert method == "batch" + assert params == {} + assert rid is None + + +def test_parse_jsonrpc_request_for_logging_truncated_or_non_json(): + # Truncated/oversized peek or plain non-JSON -> no attribution. + assert _parse_jsonrpc_request_for_logging(b'{"jsonrpc":"2.0","meth') == ( + None, + {}, + None, + ) + assert _parse_jsonrpc_request_for_logging(b"not json at all") == (None, {}, None) + # A bare JSON scalar (neither dict nor list). + assert _parse_jsonrpc_request_for_logging(b'"a string"') == (None, {}, None) From c9979098bc2b4e73dadae3d70045ed1b3373d014 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sun, 13 Sep 2026 11:10:37 -0700 Subject: [PATCH 04/11] fix(mcp): log SSE protocol rejections --- .../proxy/_experimental/mcp_server/server.py | 26 +++++++++- tests/mcp_tests/test_mcp_logging.py | 48 +++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index 11a536d7f36..ccd02e0eef0 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -3791,6 +3791,26 @@ if MCP_AVAILABLE: except Exception as log_exc: # noqa: BLE001 # best-effort: logging must never break the MCP response verbose_logger.debug("MCP protocol-error logging failed (continuing): %s", log_exc) + def _parse_jsonrpc_error_response_for_logging( + raw_body: bytes, + ) -> _JSONRPCErrorEnvelopeForLogging | None: + stripped_body: Final = raw_body.lstrip() + if stripped_body.startswith(b"{"): + json_body: Final = stripped_body + else: + data_lines: Final = tuple( + line.removeprefix(b"data:").lstrip() + for line in raw_body.splitlines() + if line.startswith(b"data:") + ) + if len(data_lines) != 1: + return None + json_body = data_lines[0] + try: + return _JSONRPC_ERROR_FOR_LOGGING_ADAPTER.validate_json(json_body) + except ValidationError: + return None + def _wrap_send_for_protocol_error_logging( inner_send: Send, request_method: str | None, @@ -3834,7 +3854,9 @@ if MCP_AVAILABLE: return if b'"error"' not in raw_body or b'"result"' in raw_body: return - payload: Final = _JSONRPC_ERROR_FOR_LOGGING_ADAPTER.validate_json(raw_body) + payload: Final = _parse_jsonrpc_error_response_for_logging(raw_body) + if payload is None: + return already_logged["done"] = True await _log_mcp_protocol_rejection( request_method=request_method, @@ -3845,7 +3867,7 @@ if MCP_AVAILABLE: raw_headers=raw_headers, start_time=start_time, ) - except (ValidationError, TypeError, ValueError): + except (TypeError, ValueError): # Non-JSON / streaming chunk — nothing to log here. return except Exception as wrap_exc: # noqa: BLE001 # best-effort: never break the response on logging diff --git a/tests/mcp_tests/test_mcp_logging.py b/tests/mcp_tests/test_mcp_logging.py index e8327d5a39c..86e40235770 100644 --- a/tests/mcp_tests/test_mcp_logging.py +++ b/tests/mcp_tests/test_mcp_logging.py @@ -448,6 +448,7 @@ from datetime import datetime from litellm.proxy._experimental.mcp_server.server import ( _jsonrpc_rejection_reason, _log_mcp_protocol_rejection, + _parse_jsonrpc_error_response_for_logging, _parse_jsonrpc_request_for_logging, _wrap_send_for_protocol_error_logging, ) @@ -1029,6 +1030,53 @@ def test_parse_jsonrpc_request_for_logging_truncated_or_non_json(): assert _parse_jsonrpc_request_for_logging(b'"a string"') == (None, {}, None) +def test_parse_jsonrpc_error_response_for_logging_accepts_sse(): + payload = _parse_jsonrpc_error_response_for_logging( + b'event: message\r\ndata: {"jsonrpc":"2.0","id":1,"error":{"code":-32602,"message":"bad"}}\r\n\r\n' + ) + assert payload is not None + assert payload["error"]["code"] == -32602 + + +@pytest.mark.asyncio +async def test_mcp_protocol_sse_rejection_emits_slo(): + litellm.logging_callback_manager._reset_all_callbacks() + capture = _ProtocolRejectionLogger() + litellm.callbacks = [capture] + + async def fake_send(message): + pass + + try: + wrapped = _wrap_send_for_protocol_error_logging( + fake_send, + request_method="tools/not-real", + params={}, + request_id=1, + user_api_key_auth=UserAPIKeyAuth(api_key="test", user_id="u1"), + raw_headers={}, + start_time=datetime.now(), + ) + await wrapped( + { + "type": "http.response.body", + "body": ( + b'event: message\r\ndata: {"jsonrpc":"2.0","id":1,' + b'"error":{"code":-32602,"message":"Invalid request parameters"}}\r\n\r\n' + ), + } + ) + await asyncio.sleep(1) + finally: + litellm.callbacks = [] + + payloads = [payload for payload in capture.failure_payloads if isinstance(payload, dict)] + assert len(payloads) == 1 + spend_meta = (payloads[0].get("metadata") or {}).get("spend_logs_metadata") or {} + assert spend_meta.get("mcp_operation") == "tools/not-real" + assert spend_meta.get("rejection_reason") == "malformed_params" + + _QUEUED_LOGGING_OUTLIVES_TEST = ''' import time From d7e9146ba1e10ef9550b044e15c457b228a8ecc7 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sun, 13 Sep 2026 11:14:32 -0700 Subject: [PATCH 05/11] style(mcp): format protocol logging changes --- litellm/proxy/_experimental/mcp_server/server.py | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index ccd02e0eef0..fb0936aa602 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -139,14 +139,10 @@ class _JSONRPCErrorEnvelopeForLogging(TypedDict): error: ReadOnly[_JSONRPCErrorForLogging] -_JSONRPC_REQUEST_FOR_LOGGING_ADAPTER: Final[ - TypeAdapter[_JSONRPCRequestForLogging | tuple[object, ...]] -] = TypeAdapter( +_JSONRPC_REQUEST_FOR_LOGGING_ADAPTER: Final[TypeAdapter[_JSONRPCRequestForLogging | tuple[object, ...]]] = TypeAdapter( _JSONRPCRequestForLogging | tuple[object, ...] ) -_JSONRPC_PARAMS_FOR_LOGGING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter( - Mapping[str, object] -) +_JSONRPC_PARAMS_FOR_LOGGING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) _JSONRPC_ERROR_FOR_LOGGING_ADAPTER: Final[TypeAdapter[_JSONRPCErrorEnvelopeForLogging]] = TypeAdapter( _JSONRPCErrorEnvelopeForLogging ) @@ -3726,7 +3722,9 @@ if MCP_AVAILABLE: if tool_name is not None: spend_logs_metadata["mcp_tool_name"] = tool_name - request_data: Final[dict[str, object]] = { # mutable-ok: auth enrichment mutates this legacy request payload + request_data: Final[ + dict[str, object] + ] = { # mutable-ok: auth enrichment mutates this legacy request payload "model": "MCP: protocol_error", "call_type": CallTypes.call_mcp_tool.value, "litellm_call_id": str(uuid.uuid4()), @@ -3799,9 +3797,7 @@ if MCP_AVAILABLE: json_body: Final = stripped_body else: data_lines: Final = tuple( - line.removeprefix(b"data:").lstrip() - for line in raw_body.splitlines() - if line.startswith(b"data:") + line.removeprefix(b"data:").lstrip() for line in raw_body.splitlines() if line.startswith(b"data:") ) if len(data_lines) != 1: return None From 809c7b50eac2d81edd370174dba86005352dbeb8 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Mon, 14 Sep 2026 00:42:40 -0700 Subject: [PATCH 06/11] test(mcp): satisfy current quality gates --- tests/mcp_tests/test_mcp_logging.py | 60 ++++++++++--------- ui/litellm-dashboard/src/lib/http/schema.d.ts | 2 - 2 files changed, 32 insertions(+), 30 deletions(-) diff --git a/tests/mcp_tests/test_mcp_logging.py b/tests/mcp_tests/test_mcp_logging.py index 86e40235770..421441c5f6d 100644 --- a/tests/mcp_tests/test_mcp_logging.py +++ b/tests/mcp_tests/test_mcp_logging.py @@ -471,6 +471,10 @@ class _ProtocolRejectionLogger(CustomLogger): self.failure_payloads.append(("UNEXPECTED_SUCCESS", kwargs.get("standard_logging_object"))) +def _set_protocol_test_callbacks(callbacks): + litellm.callbacks = callbacks # test-quality-ok: integration tests require LiteLLM's public callback seam and every caller restores it + + def _jsonrpc_error_body(code, message="boom", _id=1): return json.dumps({"jsonrpc": "2.0", "id": _id, "error": {"code": code, "message": message}}).encode("utf-8") @@ -496,7 +500,7 @@ async def test_mcp_protocol_rejection_emits_standard_logging_object(): carrying the rejection metadata (#28929).""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") sent_messages = [] @@ -520,7 +524,7 @@ async def test_mcp_protocol_rejection_emits_standard_logging_object(): # async_failure_handler may dispatch on the loop; give it a tick. await asyncio.sleep(1) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) # Response forwarded byte-for-byte and unchanged. assert any(m.get("type") == "http.response.body" and m.get("body") == body for m in sent_messages), ( @@ -550,7 +554,7 @@ async def test_mcp_protocol_unknown_method_emits_slo(): reason and no tool name (empty params).""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") @@ -575,7 +579,7 @@ async def test_mcp_protocol_unknown_method_emits_slo(): ) await asyncio.sleep(1) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) non_none = [p for p in capture.failure_payloads if isinstance(p, dict)] assert non_none, f"expected an SLO, got: {capture.failure_payloads!r}" @@ -591,7 +595,7 @@ async def test_mcp_protocol_batch_rejection_emits_slo(): Codex batch finding).""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") @@ -618,7 +622,7 @@ async def test_mcp_protocol_batch_rejection_emits_slo(): ) await asyncio.sleep(1) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) non_none = [p for p in capture.failure_payloads if isinstance(p, dict)] assert non_none, f"expected a batch SLO, got: {capture.failure_payloads!r}" @@ -633,7 +637,7 @@ async def test_mcp_protocol_success_path_emits_no_failure_record(): record, and the response must still be forwarded unchanged.""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") sent_messages = [] @@ -655,7 +659,7 @@ async def test_mcp_protocol_success_path_emits_no_failure_record(): await wrapped({"type": "http.response.body", "body": body}) await asyncio.sleep(1) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert capture.failure_payloads == [], ( f"success path must not dispatch any failure record, got: {capture.failure_payloads!r}" @@ -689,7 +693,7 @@ async def test_mcp_protocol_logging_failure_does_not_break_response(): still forward the client response (observability must never drop a response).""" litellm.logging_callback_manager._reset_all_callbacks() - litellm.callbacks = [] + _set_protocol_test_callbacks([]) user_auth = UserAPIKeyAuth(api_key="test", user_id="u1") sent_messages = [] @@ -698,7 +702,7 @@ async def test_mcp_protocol_logging_failure_does_not_break_response(): sent_messages.append(message) try: - with patch( + with patch( # test-quality-ok: fault injection verifies logging failure cannot interrupt the already-forwarded MCP response "litellm.proxy._experimental.mcp_server.server.function_setup", side_effect=RuntimeError("logging init down"), ): @@ -715,7 +719,7 @@ async def test_mcp_protocol_logging_failure_does_not_break_response(): # Must not raise. await wrapped({"type": "http.response.body", "body": body}) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert any(m.get("body") == body for m in sent_messages), "response must be forwarded even when logging init raises" @@ -726,7 +730,7 @@ async def test_mcp_log_protocol_rejection_no_auth_is_noop(): auth.""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) try: await _log_mcp_protocol_rejection( request_method="tools/call", @@ -739,7 +743,7 @@ async def test_mcp_log_protocol_rejection_no_auth_is_noop(): ) await asyncio.sleep(0.2) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert capture.failure_payloads == [] @@ -750,7 +754,7 @@ async def test_mcp_protocol_wrapper_skips_oversized_body(): happens to contain an error-shaped substring.""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) sent = [] @@ -772,7 +776,7 @@ async def test_mcp_protocol_wrapper_skips_oversized_body(): await wrapped({"type": "http.response.body", "body": big}) await asyncio.sleep(0.3) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert capture.failure_payloads == [], "oversized body must not be parsed/logged" assert any(m.get("body") == big for m in sent) @@ -784,7 +788,7 @@ async def test_mcp_protocol_wrapper_skips_non_error_body(): error envelope (no '"error"', or a '"result"' present) without json.loads.""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) async def fake_send(message): pass @@ -810,7 +814,7 @@ async def test_mcp_protocol_wrapper_skips_non_error_body(): ) await asyncio.sleep(0.3) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert capture.failure_payloads == [], "non-error bodies must not be logged" @@ -821,7 +825,7 @@ async def test_mcp_protocol_wrapper_ignores_non_body_messages_and_empty(): attempting to log.""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) sent = [] @@ -842,7 +846,7 @@ async def test_mcp_protocol_wrapper_ignores_non_body_messages_and_empty(): await wrapped({"type": "http.response.body", "body": b""}) await asyncio.sleep(0.2) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert capture.failure_payloads == [] assert len(sent) == 2 # both forwarded @@ -854,7 +858,7 @@ async def test_mcp_protocol_wrapper_handles_malformed_error_body(): and is swallowed (no SLO, response still forwarded).""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) sent = [] @@ -877,7 +881,7 @@ async def test_mcp_protocol_wrapper_handles_malformed_error_body(): await wrapped({"type": "http.response.body", "body": body}) await asyncio.sleep(0.2) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert capture.failure_payloads == [] assert any(m.get("body") == body for m in sent) @@ -889,7 +893,7 @@ async def test_mcp_log_protocol_rejection_logging_obj_none_is_noop(monkeypatch): raising (covers the None-guard branch).""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) import litellm.proxy._experimental.mcp_server.server as mcp_server_mod @@ -909,7 +913,7 @@ async def test_mcp_log_protocol_rejection_logging_obj_none_is_noop(monkeypatch): ) await asyncio.sleep(0.2) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) assert capture.failure_payloads == [] @@ -920,7 +924,7 @@ async def test_mcp_protocol_wrapper_logs_only_once_per_request(): does not produce a second SLO (covers the early-return guard).""" litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) async def fake_send(message): pass @@ -939,7 +943,7 @@ async def test_mcp_protocol_wrapper_logs_only_once_per_request(): await wrapped({"type": "http.response.body", "body": _jsonrpc_error_body(-32601)}) await asyncio.sleep(1) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) non_none = [p for p in capture.failure_payloads if isinstance(p, dict)] assert len(non_none) == 1, f"exactly one SLO expected, got {len(non_none)}" @@ -951,7 +955,7 @@ async def test_mcp_protocol_wrapper_swallows_unexpected_error(monkeypatch): by the broad guard and never breaks the response (covers the BLE001 branch).""" litellm.logging_callback_manager._reset_all_callbacks() - litellm.callbacks = [] + _set_protocol_test_callbacks([]) import litellm.proxy._experimental.mcp_server.server as mcp_server_mod @@ -1042,7 +1046,7 @@ def test_parse_jsonrpc_error_response_for_logging_accepts_sse(): async def test_mcp_protocol_sse_rejection_emits_slo(): litellm.logging_callback_manager._reset_all_callbacks() capture = _ProtocolRejectionLogger() - litellm.callbacks = [capture] + _set_protocol_test_callbacks([capture]) async def fake_send(message): pass @@ -1068,7 +1072,7 @@ async def test_mcp_protocol_sse_rejection_emits_slo(): ) await asyncio.sleep(1) finally: - litellm.callbacks = [] + _set_protocol_test_callbacks([]) payloads = [payload for payload in capture.failure_payloads if isinstance(payload, dict)] assert len(payloads) == 1 diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 7eadaa6c991..839aa52fa84 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -16781,7 +16781,6 @@ export interface paths { * - permissions: Optional[dict] - [Not Implemented Yet] User-specific permissions, eg. turning off pii masking. * - metadata: Optional[dict] - Metadata for user, store information for user. Example metadata = {"team": "core-infra", "app": "app2", "email": "ishaan@berri.ai" } * - max_parallel_requests: Optional[int] - Rate limit a user based on the number of parallel requests. Raises 429 error, if user's parallel requests > x. - * - soft_budget: Optional[float] - Get alerts when user crosses given budget, doesn't block requests. * - model_max_budget: Optional[dict] - Model-specific max budget for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-budgets-to-keys) * - budget_fallbacks: Optional[Dict[str, List[str]]] - Per-model fallback chain tried in order when that model's own `model_max_budget` is exceeded, e.g. {"gpt-4o": ["gpt-4o-mini"]}. * - model_rpm_limit: Optional[float] - Model-specific rpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys) @@ -16887,7 +16886,6 @@ export interface paths { * - permissions: Optional[dict] - [Not Implemented Yet] User-specific permissions, eg. turning off pii masking. * - metadata: Optional[dict] - Metadata for user, store information for user. Example metadata = {"team": "core-infra", "app": "app2", "email": "ishaan@berri.ai" } * - max_parallel_requests: Optional[int] - Rate limit a user based on the number of parallel requests. Raises 429 error, if user's parallel requests > x. - * - soft_budget: Optional[float] - Get alerts when user crosses given budget, doesn't block requests. * - model_max_budget: Optional[dict] - Model-specific max budget for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-budgets-to-keys) * - budget_fallbacks: Optional[Dict[str, List[str]]] - Per-model fallback chain tried in order when that model's own `model_max_budget` is exceeded, e.g. {"gpt-4o": ["gpt-4o-mini"]}. * - model_rpm_limit: Optional[float] - Model-specific rpm limit for user. [Docs](https://docs.litellm.ai/docs/proxy/users#add-model-specific-limits-to-keys) From 6bd971bad8a988221c09f7f89095e576e444261a Mon Sep 17 00:00:00 2001 From: Tarshann Date: Mon, 14 Sep 2026 01:02:47 -0700 Subject: [PATCH 07/11] fix(mcp): assign parsed response once --- litellm/proxy/_experimental/mcp_server/server.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index fb0936aa602..0b3633cf54e 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -3794,14 +3794,15 @@ if MCP_AVAILABLE: ) -> _JSONRPCErrorEnvelopeForLogging | None: stripped_body: Final = raw_body.lstrip() if stripped_body.startswith(b"{"): - json_body: Final = stripped_body + candidate_body = stripped_body else: data_lines: Final = tuple( line.removeprefix(b"data:").lstrip() for line in raw_body.splitlines() if line.startswith(b"data:") ) if len(data_lines) != 1: return None - json_body = data_lines[0] + candidate_body = data_lines[0] + json_body: Final = candidate_body try: return _JSONRPC_ERROR_FOR_LOGGING_ADAPTER.validate_json(json_body) except ValidationError: From 2d82df967b558cbf30462d394129d292df756e99 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sat, 31 Oct 2026 14:50:35 -0500 Subject: [PATCH 08/11] test(mcp): point protocol-rejection tests at protocol_error_logging and cover remaining branches --- .../mcp_server/protocol_error_logging.py | 9 ++-- .../unit/proxy/mcp_server/test_mcp_logging.py | 43 +++++++++---------- 2 files changed, 27 insertions(+), 25 deletions(-) diff --git a/litellm/proxy/mcp_server/protocol_error_logging.py b/litellm/proxy/mcp_server/protocol_error_logging.py index f513ec12ac0..0f80038421c 100644 --- a/litellm/proxy/mcp_server/protocol_error_logging.py +++ b/litellm/proxy/mcp_server/protocol_error_logging.py @@ -27,6 +27,7 @@ from litellm.proxy.litellm_pre_call_utils import ( ) from litellm.types.utils import CallTypes from litellm.utils import Rules, function_setup + # Upper bound on a response body the send wrapper will parse to detect a # JSON-RPC rejection. Rejection envelopes are tiny; anything larger is a # permitted tool result and must not be re-deserialized after it was sent. @@ -78,6 +79,7 @@ def _jsonrpc_rejection_reason(error_code: int | None) -> str: case _: return "protocol_error" + def parse_jsonrpc_request_for_logging( body: bytes, ) -> tuple[str | None, Mapping[str, object], object]: @@ -112,6 +114,7 @@ def parse_jsonrpc_request_for_logging( return method, EMPTY_MCP_PROTOCOL_PARAMS, request.get("id") return method, params, request.get("id") + async def _log_mcp_protocol_rejection( request_method: str | None, params: Mapping[str, object], @@ -164,9 +167,7 @@ async def _log_mcp_protocol_rejection( if tool_name is not None: spend_logs_metadata["mcp_tool_name"] = tool_name - request_data: Final[ - dict[str, object] - ] = { # mutable-ok: auth enrichment mutates this legacy request payload + request_data: Final[dict[str, object]] = { # mutable-ok: auth enrichment mutates this legacy request payload "model": "MCP: protocol_error", "call_type": CallTypes.call_mcp_tool.value, "litellm_call_id": str(uuid.uuid4()), @@ -231,6 +232,7 @@ async def _log_mcp_protocol_rejection( except Exception as log_exc: # noqa: BLE001 # best-effort: logging must never break the MCP response verbose_logger.debug("MCP protocol-error logging failed (continuing): %s", log_exc) + def _parse_jsonrpc_error_response_for_logging( raw_body: bytes, ) -> _JSONRPCErrorEnvelopeForLogging | None: @@ -250,6 +252,7 @@ def _parse_jsonrpc_error_response_for_logging( except ValidationError: return None + def wrap_send_for_protocol_error_logging( inner_send: Send, request_method: str | None, diff --git a/tests/unit/proxy/mcp_server/test_mcp_logging.py b/tests/unit/proxy/mcp_server/test_mcp_logging.py index ede8a2189d9..6025e0a7d25 100644 --- a/tests/unit/proxy/mcp_server/test_mcp_logging.py +++ b/tests/unit/proxy/mcp_server/test_mcp_logging.py @@ -186,6 +186,7 @@ async def test_sdk_rejection_sync_logging_allows_concurrent_ping( loop: Final = asyncio.get_running_loop() ping_responses: Final = SimpleQueue[EmptyResult]() async with Client(server, mode="legacy") as client: + def sync_failure( kwargs: dict[str, object], response_obj: object, start_time: datetime, end_time: datetime ) -> None: @@ -326,6 +327,7 @@ def _call_tool_params(name, arguments=None): return CallToolRequestParams(name=name, arguments=arguments) + class TestMCPLogger(CustomLogger): def __init__(self): self.standard_logging_payload = None @@ -447,9 +449,7 @@ async def test_mcp_cost_tracking(): await asyncio.sleep(2) logged_standard_logging_payload = ( - test_logger.mcp_tool_call_payloads[-1] - if test_logger.mcp_tool_call_payloads - else None + test_logger.mcp_tool_call_payloads[-1] if test_logger.mcp_tool_call_payloads else None ) print("logged_standard_logging_payload", logged_standard_logging_payload) @@ -575,13 +575,9 @@ async def test_mcp_cost_tracking_per_tool(): await asyncio.sleep(2) logged_standard_logging_payload_1 = ( - test_logger.mcp_tool_call_payloads[-1] - if test_logger.mcp_tool_call_payloads - else None - ) - print( - "logged_standard_logging_payload_1", logged_standard_logging_payload_1 + test_logger.mcp_tool_call_payloads[-1] if test_logger.mcp_tool_call_payloads else None ) + print("logged_standard_logging_payload_1", logged_standard_logging_payload_1) # Verify expensive tool cost assert logged_standard_logging_payload_1 is not None, "Standard logging payload 1 should not be None" @@ -601,13 +597,9 @@ async def test_mcp_cost_tracking_per_tool(): await asyncio.sleep(2) logged_standard_logging_payload_2 = ( - test_logger.mcp_tool_call_payloads[-1] - if test_logger.mcp_tool_call_payloads - else None - ) - print( - "logged_standard_logging_payload_2", logged_standard_logging_payload_2 + test_logger.mcp_tool_call_payloads[-1] if test_logger.mcp_tool_call_payloads else None ) + print("logged_standard_logging_payload_2", logged_standard_logging_payload_2) # Verify cheap tool cost assert logged_standard_logging_payload_2 is not None, "Standard logging payload 2 should not be None" @@ -725,9 +717,7 @@ async def test_mcp_tool_call_hook(): # check logged standard logging payload logged_standard_logging_payload = ( - test_logger.mcp_tool_call_payloads[-1] - if test_logger.mcp_tool_call_payloads - else None + test_logger.mcp_tool_call_payloads[-1] if test_logger.mcp_tool_call_payloads else None ) print("logged_standard_logging_payload", logged_standard_logging_payload) assert logged_standard_logging_payload is not None, "Standard logging payload should not be None" @@ -1201,7 +1191,7 @@ async def test_mcp_log_protocol_rejection_logging_obj_none_is_noop(monkeypatch): capture = _ProtocolRejectionLogger() _set_protocol_test_callbacks([capture]) - import litellm.proxy._experimental.mcp_server.server as mcp_server_mod + import litellm.proxy.mcp_server.protocol_error_logging as mcp_server_mod def _fake_function_setup(*args, **kwargs): return None, {} @@ -1263,7 +1253,7 @@ async def test_mcp_protocol_wrapper_swallows_unexpected_error(monkeypatch): litellm.logging_callback_manager._reset_all_callbacks() _set_protocol_test_callbacks([]) - import litellm.proxy._experimental.mcp_server.server as mcp_server_mod + import litellm.proxy.mcp_server.protocol_error_logging as mcp_server_mod async def _boom(**kwargs): raise RuntimeError("unexpected") @@ -1387,7 +1377,7 @@ async def test_mcp_protocol_sse_rejection_emits_slo(): assert spend_meta.get("rejection_reason") == "malformed_params" -_QUEUED_LOGGING_OUTLIVES_TEST = ''' +_QUEUED_LOGGING_OUTLIVES_TEST = """ import time from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER @@ -1410,7 +1400,7 @@ async def test_2_starts_after_the_previous_tests_logging_ran(): GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(_record_run()) await GLOBAL_LOGGING_WORKER.flush() assert [t < started_at for t in ran_at] == [True, False] -''' +""" def test_logging_queued_by_one_test_is_drained_before_the_next(tmp_path: Path): @@ -1427,3 +1417,12 @@ def test_logging_queued_by_one_test_is_drained_before_the_next(tmp_path: Path): timeout=120, ) assert result.returncode == 0, result.stdout + result.stderr + + +def test_jsonrpc_rejection_reason_maps_internal_error(): + assert _jsonrpc_rejection_reason(-32603) == "internal_error" + + +def test_parse_jsonrpc_error_response_rejects_multiple_sse_data_lines(): + body = b'data: {"error":{"code":-32602}}\ndata: {"error":{"code":-32601}}\n' + assert _parse_jsonrpc_error_response_for_logging(body) is None From 3bd6ad6ceaaa999a40ea74a68eec0ce1afba7752 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sat, 31 Oct 2026 14:51:31 -0500 Subject: [PATCH 09/11] test(mcp): drop duplicate json import left by the upstream merge --- tests/unit/proxy/mcp_server/test_mcp_logging.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/unit/proxy/mcp_server/test_mcp_logging.py b/tests/unit/proxy/mcp_server/test_mcp_logging.py index 6025e0a7d25..b269eb53603 100644 --- a/tests/unit/proxy/mcp_server/test_mcp_logging.py +++ b/tests/unit/proxy/mcp_server/test_mcp_logging.py @@ -734,7 +734,6 @@ async def test_mcp_tool_call_hook(): # called) — the rejection path builds a genuine LiteLLMLoggingObj and drives # its async_failure_handler, mirroring list_mcp_tools / call_mcp_tool. # --------------------------------------------------------------------------- -import json from datetime import datetime from litellm.proxy.mcp_server.protocol_error_logging import ( From 21bc41ee90c1f91ded8cb8fad201af047dcbf88f Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sat, 31 Oct 2026 14:54:12 -0500 Subject: [PATCH 10/11] fix(mcp): satisfy type-discipline gate in protocol-rejection logging --- .../mcp_server/protocol_error_logging.py | 30 ++++++------------- litellm/proxy/mcp_server/server.py | 2 +- 2 files changed, 10 insertions(+), 22 deletions(-) diff --git a/litellm/proxy/mcp_server/protocol_error_logging.py b/litellm/proxy/mcp_server/protocol_error_logging.py index 0f80038421c..b29ed99bc0f 100644 --- a/litellm/proxy/mcp_server/protocol_error_logging.py +++ b/litellm/proxy/mcp_server/protocol_error_logging.py @@ -32,9 +32,7 @@ from litellm.utils import Rules, function_setup # JSON-RPC rejection. Rejection envelopes are tiny; anything larger is a # permitted tool result and must not be re-deserialized after it was sent. _MCP_PROTOCOL_ERROR_PEEK_MAX_BYTES: Final = 64 * 1024 -EMPTY_MCP_PROTOCOL_PARAMS: Final[Mapping[str, object]] = types.MappingProxyType( - {} # mutable-ok: frozen immediately and reused as the empty request-parameter value -) +EMPTY_MCP_PROTOCOL_PARAMS: Final[Mapping[str, object]] = types.MappingProxyType({}) class _JSONRPCRequestForLogging(TypedDict, total=False): @@ -171,11 +169,7 @@ async def _log_mcp_protocol_rejection( "model": "MCP: protocol_error", "call_type": CallTypes.call_mcp_tool.value, "litellm_call_id": str(uuid.uuid4()), - "litellm_trace_id": get_chain_id_from_headers( - dict(raw_headers) # mutable-ok: legacy trace helper requires a concrete dict but does not mutate it - if raw_headers is not None - else None - ), + "litellm_trace_id": get_chain_id_from_headers(dict(raw_headers) if raw_headers is not None else None), "metadata": { "spend_logs_metadata": spend_logs_metadata, }, @@ -237,16 +231,12 @@ def _parse_jsonrpc_error_response_for_logging( raw_body: bytes, ) -> _JSONRPCErrorEnvelopeForLogging | None: stripped_body: Final = raw_body.lstrip() - if stripped_body.startswith(b"{"): - candidate_body = stripped_body - else: - data_lines: Final = tuple( - line.removeprefix(b"data:").lstrip() for line in raw_body.splitlines() if line.startswith(b"data:") - ) - if len(data_lines) != 1: - return None - candidate_body = data_lines[0] - json_body: Final = candidate_body + data_lines: Final = tuple( + line.removeprefix(b"data:").lstrip() for line in raw_body.splitlines() if line.startswith(b"data:") + ) + if not stripped_body.startswith(b"{") and len(data_lines) != 1: + return None + json_body: Final = stripped_body if stripped_body.startswith(b"{") else data_lines[0] try: return _JSONRPC_ERROR_FOR_LOGGING_ADAPTER.validate_json(json_body) except ValidationError: @@ -273,9 +263,7 @@ def wrap_send_for_protocol_error_logging( if request_method is None: return inner_send - already_logged: Final = { # mutable-ok: closure state prevents duplicate records across response chunks - "done": False - } + already_logged: Final = {"done": False} async def _logging_send(message: Message) -> None: await inner_send(message) diff --git a/litellm/proxy/mcp_server/server.py b/litellm/proxy/mcp_server/server.py index 4830b49592f..a53c774fe20 100644 --- a/litellm/proxy/mcp_server/server.py +++ b/litellm/proxy/mcp_server/server.py @@ -2787,7 +2787,7 @@ if MCP_AVAILABLE: # Additively observe protocol-level JSON-RPC rejections (#28929) # without touching the success path. Applied first so the # stateful-session wrapper (below) still wraps the outermost send. - local_send = wrap_send_for_protocol_error_logging( + local_send = wrap_send_for_protocol_error_logging( # rebind-ok: layers the logging wrapper onto the per-request send local_send, request_method=_protocol_log_method, params=_protocol_log_params, From c3baad008eefea177b9ce294bc4bcd3e175e8c41 Mon Sep 17 00:00:00 2001 From: Tarshann Date: Sat, 31 Oct 2026 14:57:47 -0500 Subject: [PATCH 11/11] fix(mcp): annotate forwarded logging kwargs for basedpyright --- litellm/proxy/mcp_server/protocol_error_logging.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/proxy/mcp_server/protocol_error_logging.py b/litellm/proxy/mcp_server/protocol_error_logging.py index b29ed99bc0f..026626109bc 100644 --- a/litellm/proxy/mcp_server/protocol_error_logging.py +++ b/litellm/proxy/mcp_server/protocol_error_logging.py @@ -208,7 +208,7 @@ async def _log_mcp_protocol_rejection( original_function="call_mcp_tool", rules_obj=Rules(), start_time=start_time, - **request_data, # pyright: ignore[reportAny] # function_setup erases its forwarded keyword types + **request_data, # pyright: ignore[reportAny, reportArgumentType] # function_setup erases its forwarded keyword types ) if litellm_logging_obj is None: # pyright: ignore[reportUnnecessaryComparison] # keep best-effort guard for runtime substitutes