mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-04 02:31:27 +00:00
fix(mcp): log upstream request method, body and response on MCP tool-list and OAuth2 token failures
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
parent
328a5f5d60
commit
1425c71c10
5 changed files with 165 additions and 3 deletions
|
|
@ -85,11 +85,14 @@ Usage with curl::
|
|||
http://localhost:4000/mcp/atlassian_mcp
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
import httpx
|
||||
from starlette.types import Message, Send
|
||||
|
||||
from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
|
||||
from litellm.proxy._experimental.mcp_server.faults.traversal import iter_exception_tree
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.types.mcp_server.mcp_server_manager import MCPServer
|
||||
|
|
@ -127,6 +130,10 @@ class MCPDebug:
|
|||
@staticmethod
|
||||
def _mask(value: str | None) -> str:
|
||||
"""Mask a single value for safe display in headers."""
|
||||
return MCPDebug.mask_secret(value)
|
||||
|
||||
@staticmethod
|
||||
def mask_secret(value: str | None) -> str:
|
||||
if not value:
|
||||
return "(none)"
|
||||
return MCPDebug._masker._mask_value(value)
|
||||
|
|
@ -311,3 +318,61 @@ class MCPDebug:
|
|||
server_url=server_url,
|
||||
server_auth_type=server_auth_type,
|
||||
)
|
||||
|
||||
|
||||
_BODY_PREVIEW_CHARS: Final = 512
|
||||
_SENSITIVE_HEADER_NAMES: Final = frozenset({"authorization", "proxy-authorization", "cookie", "x-api-key"})
|
||||
_SENSITIVE_BODY_FIELD: Final = re.compile(
|
||||
r'(?P<key>"?(?:client_secret|client_assertion|refresh_token|access_token|id_token|password|code)"?\s*[=:]\s*"?)'
|
||||
r'(?P<value>[^&"\s,}]+)'
|
||||
)
|
||||
|
||||
|
||||
def _mask_body_match(match: re.Match[str]) -> str:
|
||||
return f"{match.group('key')}{MCPDebug.mask_secret(match.group('value'))}"
|
||||
|
||||
|
||||
def _preview(raw: bytes) -> str:
|
||||
text: Final = _SENSITIVE_BODY_FIELD.sub(_mask_body_match, raw.decode("utf-8", errors="replace"))
|
||||
return (
|
||||
text
|
||||
if len(text) <= _BODY_PREVIEW_CHARS
|
||||
else f"{text[:_BODY_PREVIEW_CHARS]}...(+{len(text) - _BODY_PREVIEW_CHARS} chars)"
|
||||
)
|
||||
|
||||
|
||||
def _masked_headers(headers: httpx.Headers) -> str:
|
||||
return ", ".join(
|
||||
f"{name}={MCPDebug.mask_secret(value) if name.lower() in _SENSITIVE_HEADER_NAMES else value}"
|
||||
for name, value in headers.items()
|
||||
)
|
||||
|
||||
|
||||
def _request_body_preview(request: httpx.Request) -> str:
|
||||
try:
|
||||
return _preview(request.content) or "(empty)"
|
||||
except httpx.RequestNotRead:
|
||||
return "(streamed, not captured)"
|
||||
|
||||
|
||||
def _response_body_preview(response: httpx.Response) -> str:
|
||||
try:
|
||||
return _preview(response.content) or "(empty)"
|
||||
except httpx.ResponseNotRead:
|
||||
return "(not read)"
|
||||
|
||||
|
||||
def describe_upstream_http_failure(exc: BaseException) -> str | None:
|
||||
"""One line per upstream ``httpx.Response`` in the exception tree: the request method, URL,
|
||||
masked request headers and JSON-RPC body that were sent, plus the status and body that came back.
|
||||
``None`` when the failure never reached an HTTP response (DNS, refused connection, timeout)."""
|
||||
lines: Final = tuple(
|
||||
f"{response.request.method} {response.request.url} -> HTTP {response.status_code} {response.reason_phrase}"
|
||||
f" | request headers: {_masked_headers(response.request.headers)}"
|
||||
f" | request body: {_request_body_preview(response.request)}"
|
||||
f" | response body: {_response_body_preview(response)}"
|
||||
for current in iter_exception_tree(exc)
|
||||
for response in (getattr(current, "response", None),)
|
||||
if isinstance(response, httpx.Response)
|
||||
)
|
||||
return "\n".join(lines) or None
|
||||
|
|
|
|||
|
|
@ -80,6 +80,7 @@ from litellm.proxy._experimental.mcp_server.faults.list_outcomes import (
|
|||
raise_classified_list_failure,
|
||||
upstream_auth_challenge,
|
||||
)
|
||||
from litellm.proxy._experimental.mcp_server.mcp_debug import describe_upstream_http_failure
|
||||
from litellm.proxy._experimental.mcp_server.oauth2_token_cache import (
|
||||
MCPPerUserTokenCache,
|
||||
mcp_per_user_token_cache,
|
||||
|
|
@ -1353,6 +1354,11 @@ def _extract_upstream_auth_failure(
|
|||
return upstream_auth_challenge(exc)
|
||||
|
||||
|
||||
def _upstream_failure_suffix(exc: BaseException) -> str:
|
||||
detail: Final = describe_upstream_http_failure(exc)
|
||||
return f"\n upstream exchange: {detail}" if detail else ""
|
||||
|
||||
|
||||
def _obo_retry_applies(server: MCPServer, subject_token: str | None) -> bool:
|
||||
"""Whether an upstream 401/403 should invalidate the minted credential and retry once.
|
||||
|
||||
|
|
@ -4259,7 +4265,9 @@ class MCPServerManager:
|
|||
except MCPServerListError:
|
||||
raise
|
||||
except Exception as e:
|
||||
verbose_logger.warning("Failed to get tools from server %s: %s", server.name, e)
|
||||
verbose_logger.warning(
|
||||
"Failed to get tools from server %s: %s%s", server.name, e, _upstream_failure_suffix(e)
|
||||
)
|
||||
raise_classified_list_failure(e, server.name, suppress_challenge=server.is_dcr_bridge)
|
||||
|
||||
async def get_prompts_from_server(
|
||||
|
|
@ -5004,7 +5012,7 @@ class MCPServerManager:
|
|||
verbose_logger.warning("Connection error while listing tools from %s: %s", server_name, e)
|
||||
raise MCPServerListError(ServerListFault(tag="unreachable"), server_name) from e
|
||||
except Exception as e:
|
||||
verbose_logger.warning("Error listing tools from %s: %s", server_name, e)
|
||||
verbose_logger.warning("Error listing tools from %s: %s%s", server_name, e, _upstream_failure_suffix(e))
|
||||
raise_classified_list_failure(e, server_name)
|
||||
|
||||
_SHORT_PREFIX_MAX_REHASH_ATTEMPTS = 1024
|
||||
|
|
|
|||
|
|
@ -37,6 +37,8 @@ import httpx
|
|||
from pydantic import BaseModel, ConfigDict, Field, SecretStr, TypeAdapter, ValidationError
|
||||
from typing_extensions import assert_never
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.proxy._experimental.mcp_server.mcp_debug import describe_upstream_http_failure
|
||||
from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import (
|
||||
InMemoryTokenCacheBackend,
|
||||
OAuthToken,
|
||||
|
|
@ -110,6 +112,10 @@ async def post_client_credentials_grant(
|
|||
)
|
||||
except httpx.HTTPStatusError as status_err:
|
||||
status_code: Final = status_err.response.status_code
|
||||
verbose_logger.warning(
|
||||
"OAuth2 client_credentials token request denied:\n upstream exchange: %s",
|
||||
describe_upstream_http_failure(status_err),
|
||||
)
|
||||
return TokenEndpointDenied(status_code=status_code, detail=f"token endpoint returned HTTP {status_code}")
|
||||
except Exception as exc: # noqa: BLE001 # any transport failure is the same outcome: unreachable
|
||||
return TokenEndpointUnreachable(detail=str(exc))
|
||||
|
|
|
|||
|
|
@ -5,9 +5,12 @@ Tests for MCPDebug — MCP OAuth2 debug response headers.
|
|||
import asyncio
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import httpx
|
||||
|
||||
from litellm.proxy._experimental.mcp_server.mcp_debug import (
|
||||
MCP_DEBUG_REQUEST_HEADER,
|
||||
MCPDebug,
|
||||
describe_upstream_http_failure,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -269,3 +272,58 @@ class TestWrapSendWithDebugHeaders:
|
|||
asyncio.run(wrapped(body_msg))
|
||||
|
||||
assert captured[0] == body_msg
|
||||
|
||||
|
||||
class TestDescribeUpstreamHttpFailure:
|
||||
@staticmethod
|
||||
def _status_error(*, body: bytes, response_body: bytes | None = None) -> httpx.HTTPStatusError:
|
||||
request = httpx.Request(
|
||||
"POST",
|
||||
"https://upstream.example/apis/mcp",
|
||||
headers={"Authorization": "Bearer secret-token-abcdef0123456789", "Content-Type": "application/json"},
|
||||
content=body,
|
||||
)
|
||||
response = (
|
||||
httpx.Response(500, request=request, content=response_body)
|
||||
if response_body is not None
|
||||
else httpx.Response(500, request=request, stream=httpx.ByteStream(b'{"error":"boom"}'))
|
||||
)
|
||||
return httpx.HTTPStatusError("500", request=request, response=response)
|
||||
|
||||
def test_includes_method_url_status_and_request_body(self):
|
||||
exc = self._status_error(
|
||||
body=b'{"method":"initialize","jsonrpc":"2.0","id":0}',
|
||||
response_body=b'{"error":"boom"}',
|
||||
)
|
||||
described = describe_upstream_http_failure(exc)
|
||||
assert described is not None
|
||||
assert "POST https://upstream.example/apis/mcp -> HTTP 500" in described
|
||||
assert '{"method":"initialize"' in described
|
||||
assert 'response body: {"error":"boom"}' in described
|
||||
|
||||
def test_masks_authorization_header_and_secret_body_fields(self):
|
||||
exc = self._status_error(
|
||||
body=b"grant_type=client_credentials&client_id=abc&client_secret=super-secret-value-1234",
|
||||
response_body=b"{}",
|
||||
)
|
||||
described = describe_upstream_http_failure(exc)
|
||||
assert described is not None
|
||||
assert "secret-token-abcdef0123456789" not in described
|
||||
assert "super-secret-value-1234" not in described
|
||||
assert "client_id=abc" in described
|
||||
assert "client_secret=" in described
|
||||
|
||||
def test_reports_unread_streamed_response_body(self):
|
||||
described = describe_upstream_http_failure(self._status_error(body=b"{}"))
|
||||
assert described is not None
|
||||
assert "response body: (not read)" in described
|
||||
|
||||
def test_finds_response_behind_cause_chain(self):
|
||||
wrapper = RuntimeError("token minting failed")
|
||||
wrapper.__cause__ = self._status_error(body=b"{}", response_body=b'{"error":"invalid_client"}')
|
||||
described = describe_upstream_http_failure(wrapper)
|
||||
assert described is not None
|
||||
assert "invalid_client" in described
|
||||
|
||||
def test_returns_none_without_http_response(self):
|
||||
assert describe_upstream_http_failure(ConnectionError("refused")) is None
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""Unit tests for MCP OAuth passthrough tool-fetch behavior."""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
|
|
@ -11,7 +12,7 @@ if sys.version_info < (3, 11):
|
|||
from exceptiongroup import ExceptionGroup
|
||||
|
||||
|
||||
from litellm.proxy._experimental.mcp_server.exceptions import MCPUpstreamAuthError
|
||||
from litellm.proxy._experimental.mcp_server.exceptions import MCPServerListError, MCPUpstreamAuthError
|
||||
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
|
||||
MCPServerManager,
|
||||
_extract_upstream_auth_failure,
|
||||
|
|
@ -434,3 +435,27 @@ async def test_aggregate_with_single_accessible_server_still_absorbs():
|
|||
|
||||
assert listing.tools == []
|
||||
assert listing.outcomes["delegate_docs"].tag == "auth_required"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_tools_logs_upstream_request_details_on_500(caplog):
|
||||
manager = MCPServerManager()
|
||||
request = httpx.Request(
|
||||
"POST",
|
||||
"https://upstream/apis/mcp",
|
||||
headers={"Authorization": "Bearer upstream-token-0123456789"},
|
||||
content=b'{"method":"initialize","jsonrpc":"2.0","id":0}',
|
||||
)
|
||||
response = httpx.Response(500, request=request)
|
||||
mock_client = MagicMock()
|
||||
mock_client.list_tools = AsyncMock(
|
||||
side_effect=httpx.HTTPStatusError("500", request=request, response=response)
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
|
||||
with pytest.raises(MCPServerListError):
|
||||
await manager._fetch_tools_with_timeout(mock_client, "sample_docs")
|
||||
|
||||
assert "POST https://upstream/apis/mcp -> HTTP 500" in caplog.text
|
||||
assert '"method":"initialize"' in caplog.text
|
||||
assert "upstream-token-0123456789" not in caplog.text
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue