scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 123

fix(mcp): write failure spend log for guardrail-blocked /mcp-rest/tools/call (#40555)

Browse source 
* fix(mcp): write failure spend log for guardrail-blocked /mcp-rest/tools/call

call_tool_rest_api only translated exceptions to HTTP responses, so a pre_mcp_call
guardrail block never reached failure_handler / async_failure_handler /
post_call_failure_hook and no LiteLLM_SpendLogs failure row was written. Extract
the failure logging from call_mcp_tool into _fire_mcp_tool_call_failure_logging
and run it in the REST route for anything raised between
common_processing_pre_call_logic and execute_mcp_tool

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(mcp): keep the original REST tool error when failure logging raises

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(mcp): log virtual mcp_tool_call failures and keep REST success latency scoped to tool execution

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: yucheng <yucheng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-11 12:54:03 -07:00 • committed by GitHub
parent f22f9bc461
commit e073cd3aeb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 407 additions and 108 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

184
litellm/proxy/_experimental/mcp_server/rest_endpoints.py
View file

@ -191,6 +191,7 @@ if MCP_AVAILABLE:
execute_mcp_tool,
filter_tools_by_allowed_tools,
filter_tools_by_key_team_permissions,
fire_mcp_tool_call_failure_logging,
)
########################################################
@ -232,6 +233,20 @@ if MCP_AVAILABLE:
return result
return outcome
async def _safe_fire_mcp_tool_call_failure_logging(
logging_obj: "LiteLLMLoggingObj | None",
exception: Exception,
start_time: datetime,
user_api_key_auth: UserAPIKeyAuth,
request_data: Mapping[str, object],
) -> None:
try:
await fire_mcp_tool_call_failure_logging(
logging_obj, exception, start_time, user_api_key_auth, request_data
)
except Exception as logging_error:
verbose_logger.warning("MCP tool call failure logging failed (continuing): %s", logging_error)
def _relay_upstream_auth_http_exception(e: MCPUpstreamAuthError, request: Request) -> HTTPException:
"""Convert a client-forwarded pass-through upstream 401 into an HTTPException that preserves the
upstream WWW-Authenticate, so a standards-compliant MCP client can run the upstream OAuth flow
@ -310,26 +325,39 @@ if MCP_AVAILABLE:
)
# MCP_TOOL_CALL_TOOL_NAME: run the same pre-call pipeline as the normal path so the tool
# execution is spend-logged and guardrail-checked.
(_, virtual_logging_obj) = await ProxyBaseLLMRequestProcessing(data=data).common_processing_pre_call_logic(
request=request,
user_api_key_dict=user_api_key_dict,
proxy_config=proxy_config,
route_type=CallTypes.call_mcp_tool.value,
proxy_logging_obj=proxy_logging_obj,
general_settings=general_settings,
)
_tool_start_time: Final = datetime.now()
result: Final = await handle_mcp_tool_call(
tool_name=tool_arguments.get("tool_name", ""),
arguments=tool_arguments.get("arguments") or {},
user_api_key_dict=user_api_key_dict,
client_ip=rest_client_ip,
mcp_auth_header=virtual_mcp_auth_header,
mcp_server_auth_headers=virtual_mcp_server_auth_headers,
oauth2_headers=virtual_oauth2_headers,
raw_headers=virtual_raw_headers,
litellm_logging_obj=virtual_logging_obj,
)
virtual_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
_request_start_time: Final = datetime.now() # noqa: DTZ005 # naive to match the tool start time below
try:
(_, virtual_logging_obj) = await virtual_processor.common_processing_pre_call_logic(
request=request,
user_api_key_dict=user_api_key_dict,
proxy_config=proxy_config,
route_type=CallTypes.call_mcp_tool.value,
proxy_logging_obj=proxy_logging_obj,
general_settings=general_settings,
)
_tool_start_time: Final = datetime.now()
result: Final = await handle_mcp_tool_call(
tool_name=tool_arguments.get("tool_name", ""),
arguments=tool_arguments.get("arguments") or {},
user_api_key_dict=user_api_key_dict,
client_ip=rest_client_ip,
mcp_auth_header=virtual_mcp_auth_header,
mcp_server_auth_headers=virtual_mcp_server_auth_headers,
oauth2_headers=virtual_oauth2_headers,
raw_headers=virtual_raw_headers,
litellm_logging_obj=virtual_logging_obj,
)
except Exception as e:
virtual_request_data: Final = virtual_processor.data
await _safe_fire_mcp_tool_call_failure_logging(
virtual_request_data.get("litellm_logging_obj"),
e,
_request_start_time,
user_api_key_dict,
virtual_request_data,
)
raise
return await _safe_fire_mcp_tool_call_logging(
virtual_logging_obj,
result,
@ -1081,65 +1109,73 @@ if MCP_AVAILABLE:
)
proxy_base_llm_response_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
(
data,
logging_obj,
) = await proxy_base_llm_response_processor.common_processing_pre_call_logic(
request=request,
user_api_key_dict=user_api_key_dict,
proxy_config=proxy_config,
route_type=CallTypes.call_mcp_tool.value,
proxy_logging_obj=proxy_logging_obj,
general_settings=general_settings,
)
_request_start_time: Final = datetime.now() # noqa: DTZ005 # naive to match the tool start time below
try:
(
data,
logging_obj,
) = await proxy_base_llm_response_processor.common_processing_pre_call_logic(
request=request,
user_api_key_dict=user_api_key_dict,
proxy_config=proxy_config,
route_type=CallTypes.call_mcp_tool.value,
proxy_logging_obj=proxy_logging_obj,
general_settings=general_settings,
)
# Extract MCP auth headers from request and add to data dict
(
mcp_auth_header,
mcp_server_auth_headers,
raw_headers_from_request,
) = _extract_mcp_headers_from_request(request, MCPRequestHandler)
if mcp_auth_header:
data["mcp_auth_header"] = mcp_auth_header
if mcp_server_auth_headers:
data["mcp_server_auth_headers"] = mcp_server_auth_headers
data["raw_headers"] = raw_headers_from_request
# Extract MCP auth headers from request and add to data dict
(
mcp_auth_header,
mcp_server_auth_headers,
raw_headers_from_request,
) = _extract_mcp_headers_from_request(request, MCPRequestHandler)
if mcp_auth_header:
data["mcp_auth_header"] = mcp_auth_header
if mcp_server_auth_headers:
data["mcp_server_auth_headers"] = mcp_server_auth_headers
data["raw_headers"] = raw_headers_from_request
# Extract user_api_key_auth from metadata and add to top level
# call_mcp_tool expects user_api_key_auth as a top-level parameter
if "metadata" in data and "user_api_key_auth" in data["metadata"]:
data["user_api_key_auth"] = data["metadata"]["user_api_key_auth"]
# Extract user_api_key_auth from metadata and add to top level
# call_mcp_tool expects user_api_key_auth as a top-level parameter
if "metadata" in data and "user_api_key_auth" in data["metadata"]:
data["user_api_key_auth"] = data["metadata"]["user_api_key_auth"]
# Resolve allowed MCP servers with IP filtering
(
allowed_mcp_servers,
canonical_server_id,
) = await _resolve_allowed_mcp_servers_with_ip_filter(request, user_api_key_dict, server_id)
# Resolve allowed MCP servers with IP filtering
(
allowed_mcp_servers,
canonical_server_id,
) = await _resolve_allowed_mcp_servers_with_ip_filter(request, user_api_key_dict, server_id)
# Look up per-user OAuth headers for this server (mirrors list_tool_rest_api).
user_oauth_extra_headers: dict[str, str] | None = None
target_server: Final = next(
(s for s in allowed_mcp_servers if s.server_id == canonical_server_id),
None,
)
if target_server is not None:
user_oauth_extra_headers = await _get_user_oauth_extra_headers(target_server, user_api_key_dict)
# Look up per-user OAuth headers for this server (mirrors list_tool_rest_api).
user_oauth_extra_headers: dict[str, str] | None = None
target_server: Final = next(
(s for s in allowed_mcp_servers if s.server_id == canonical_server_id),
None,
)
if target_server is not None:
user_oauth_extra_headers = await _get_user_oauth_extra_headers(target_server, user_api_key_dict)
# Call execute_mcp_tool directly (permission checks already done)
_tool_start_time: Final = datetime.now()
result: Final = await execute_mcp_tool(
name=tool_name,
arguments=tool_arguments,
allowed_mcp_servers=allowed_mcp_servers,
start_time=_tool_start_time,
user_api_key_auth=data.get("user_api_key_auth"),
mcp_auth_header=data.get("mcp_auth_header"),
mcp_server_auth_headers=data.get("mcp_server_auth_headers"),
oauth2_headers=user_oauth_extra_headers or data.get("oauth2_headers"),
raw_headers=data.get("raw_headers"),
litellm_logging_obj=data.get("litellm_logging_obj"),
requested_server_id=canonical_server_id,
)
# Call execute_mcp_tool directly (permission checks already done)
_tool_start_time: Final = datetime.now()
result: Final = await execute_mcp_tool(
name=tool_name,
arguments=tool_arguments,
allowed_mcp_servers=allowed_mcp_servers,
start_time=_tool_start_time,
user_api_key_auth=data.get("user_api_key_auth"),
mcp_auth_header=data.get("mcp_auth_header"),
mcp_server_auth_headers=data.get("mcp_server_auth_headers"),
oauth2_headers=user_oauth_extra_headers or data.get("oauth2_headers"),
raw_headers=data.get("raw_headers"),
litellm_logging_obj=data.get("litellm_logging_obj"),
requested_server_id=canonical_server_id,
)
except Exception as e:
request_data: Final = proxy_base_llm_response_processor.data
await _safe_fire_mcp_tool_call_failure_logging(
request_data.get("litellm_logging_obj"), e, _request_start_time, user_api_key_dict, request_data
)
raise
return await _safe_fire_mcp_tool_call_logging(
logging_obj,
result,

71
litellm/proxy/_experimental/mcp_server/server.py
View file

@ -3339,6 +3339,43 @@ if MCP_AVAILABLE:
)
return result
async def fire_mcp_tool_call_failure_logging(
logging_obj: LiteLLMLoggingObj | None,
exception: Exception,
start_time: datetime,
user_api_key_auth: UserAPIKeyAuth | None,
request_data: Mapping[str, object],
) -> None:
"""Failure logging shared by the ``/mcp`` path and the REST endpoint. Call from
inside the ``except`` block so the traceback is still available.
The failure handlers run first because ``_ProxyDBLogger.async_post_call_failure_hook``
builds the failure spend-log row from the ``standard_logging_object`` they produce;
both gate on ``should_run_logging``, so the ``@client`` wrapper does not log twice.
A relayed upstream 401 (``MCPUpstreamAuthError``) is an expected caller-must-reauth
signal and skips ``post_call_failure_hook``, which fires the ``llm_exceptions`` alert.
"""
from litellm.proxy.proxy_server import proxy_logging_obj
traceback_str: Final = traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG)
if logging_obj is not None:
end_time: Final = datetime.now() # noqa: DTZ005 # naive to match `start_time`, which it is subtracted from
logging_obj.failure_handler(exception, traceback_str, start_time, end_time)
await logging_obj.async_failure_handler(exception, traceback_str, start_time, end_time)
if isinstance(exception, MCPUpstreamAuthError) or not proxy_logging_obj or user_api_key_auth is None:
return
sanitized_request_data: Final = {
key: value for key, value in request_data.items() if key not in _MCP_CREDENTIAL_REQUEST_FIELDS
}
await proxy_logging_obj.post_call_failure_hook(
request_data=sanitized_request_data,
original_exception=exception,
user_api_key_dict=user_api_key_auth,
route="/mcp/call_tool",
traceback_str=traceback_str,
)
@client
async def call_mcp_tool(
name: str,
@ -3405,40 +3442,8 @@ if MCP_AVAILABLE:
raw_headers=raw_headers,
**kwargs,
)
except MCPUpstreamAuthError:
# A client-forwarded pass-through upstream 401 is an expected caller-must-reauth signal, so
# re-raise it without post_call_failure_hook, which fires the proxy's llm_exceptions alert.
# mcp_server_tool_call then downgrades it to an informational isError result for the
# streamable client. Note: this function is @client-decorated, so the decorator's standard
# failure logging still records the event (spend log / OTel); only the extra alert sink is
# skipped here.
raise
except Exception as e:
traceback_str: Final = traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG)
from litellm.proxy.proxy_server import proxy_logging_obj
# Ordering is load-bearing. ``_ProxyDBLogger.async_post_call_failure_hook``,
# reached below, writes the failure spend-log row from this logger's
# ``standard_logging_object``, which only exists once the failure handlers
# have run. Flush them first or the row lands with
# ``guardrail_information=None`` and a guardrail block is never counted.
#
# Not double-logged: both handlers gate on ``should_run_logging`` and then
# mark it, so the ``@client`` wrapper's own post-raise logging no-ops on this
# logger, same as ``_fire_mcp_tool_call_logging`` does for ``isError=True``.
if litellm_logging_obj is not None:
end_time: Final = datetime.now() # noqa: DTZ005 # naive to match `start_time`, which it is subtracted from
litellm_logging_obj.failure_handler(e, traceback_str, start_time, end_time)
await litellm_logging_obj.async_failure_handler(e, traceback_str, start_time, end_time)
if proxy_logging_obj and user_api_key_auth:
await proxy_logging_obj.post_call_failure_hook(
request_data=kwargs,
original_exception=e,
user_api_key_dict=user_api_key_auth,
route="/mcp/call_tool",
traceback_str=traceback_str,
)
await fire_mcp_tool_call_failure_logging(litellm_logging_obj, e, start_time, user_api_key_auth, kwargs)
raise
if litellm_logging_obj:

260
tests/test_litellm/proxy/_experimental/mcp_server/test_rest_endpoints.py
View file

@ -2595,6 +2595,79 @@ class TestCallToolRestAPI:
assert result == masked_result
async def test_success_logging_start_time_excludes_pre_call_processing(self, monkeypatch):
"""Pre-call hook latency (guardrails, header resolution) must not inflate the tool call's
logged duration on success."""
from litellm.proxy import proxy_server
async def fake_contexts(user_api_key_auth):
return [user_api_key_auth]
async def fake_get_allowed_mcp_servers(*args, **kwargs):
return ["server-1"]
class StubServer:
server_id = "server-1"
alias = "server-1"
server_name = "server-1"
name = "stub"
allowed_tools = None
mcp_info = {"server_name": "stub"}
available_on_public_internet = True
auth_type = None
stub_server = StubServer()
async def fake_add_litellm_data_to_request(**kwargs):
return kwargs.get("data", {})
pre_call_finished_at = {}
async def slow_pre_call_hook(user_api_key_dict, data, call_type):
await asyncio.sleep(0.05)
pre_call_finished_at["value"] = datetime.now()
return data
captured = {}
async def fake_execute_mcp_tool(**kwargs):
captured.update(kwargs)
return {"result": "ok"}
fire_logging = AsyncMock(return_value={"result": "ok"})
monkeypatch.setattr(rest_endpoints, "build_effective_auth_contexts", fake_contexts, raising=False)
monkeypatch.setattr(
rest_endpoints.global_mcp_server_manager,
"get_allowed_mcp_servers",
fake_get_allowed_mcp_servers,
raising=False,
)
monkeypatch.setattr(
rest_endpoints.global_mcp_server_manager,
"get_mcp_server_by_id",
lambda server_id: stub_server if server_id == "server-1" else None,
raising=False,
)
monkeypatch.setattr(
proxy_server, "add_litellm_data_to_request", fake_add_litellm_data_to_request, raising=False
)
monkeypatch.setattr(proxy_server, "proxy_config", {}, raising=False)
monkeypatch.setattr(proxy_server.proxy_logging_obj, "pre_call_hook", slow_pre_call_hook)
monkeypatch.setattr(rest_endpoints, "execute_mcp_tool", fake_execute_mcp_tool, raising=False)
monkeypatch.setattr(rest_endpoints, "_fire_mcp_tool_call_logging", fire_logging, raising=False)
request = _build_request(
path="/mcp-rest/tools/call",
method="POST",
json_body={"server_id": "server-1", "name": "demo-tool", "arguments": {}},
)
await rest_endpoints.call_tool_rest_api(request, user_api_key_dict=UserAPIKeyAuth())
logged_start_time = fire_logging.await_args.args[2]
assert captured["start_time"] >= pre_call_finished_at["value"]
assert logged_start_time == captured["start_time"]
async def test_success_logging_guardrail_rejection_propagates(self, monkeypatch):
"""A guardrail rejecting the tool result must not be swallowed as a logging failure,
otherwise the unguarded result would still be returned to the caller."""
@ -2765,6 +2838,139 @@ class TestCallToolRestAPI:
info_messages = [_rendered_log_message(c) for c in mock_logger.info.call_args_list if c.args]
assert not any("relaying upstream" in m for m in info_messages)
@pytest.mark.parametrize("raise_site", ["pre_call_hook", "execute_mcp_tool"])
async def test_guardrail_block_runs_failure_logging_before_http_translation(self, monkeypatch, raise_site):
"""A pre_mcp_call guardrail block, whether raised by the pre-call hook or from inside
execute_mcp_tool, must reach proxy_logging_obj.post_call_failure_hook (the only path that
writes the failure spend-log row) with the logging object's failure payload already built,
and the REST caller must still get the same 400 it got before."""
from litellm.proxy import proxy_server
async def fake_contexts(user_api_key_auth):
return [user_api_key_auth]
async def fake_get_allowed_mcp_servers(*args, **kwargs):
return ["server-1"]
class StubServer:
server_id = "server-1"
alias = "server-1"
server_name = "server-1"
name = "stub"
allowed_tools = None
mcp_info = {"server_name": "stub"}
available_on_public_internet = True
auth_type = None
async def fake_add_litellm_data_to_request(**kwargs):
return kwargs.get("data", {})
guardrail_error = HTTPException(
status_code=400,
detail={"error": "Content blocked: keyword 'confidential' detected", "keyword": "confidential"},
)
async def passthrough_pre_call_hook(user_api_key_dict, data, call_type):
return data
async def blocking_pre_call_hook(user_api_key_dict, data, call_type):
raise guardrail_error
async def fake_execute_mcp_tool(**kwargs):
raise guardrail_error
async def passthrough_execute_mcp_tool(**kwargs):
return []
monkeypatch.setattr(rest_endpoints, "build_effective_auth_contexts", fake_contexts, raising=False)
monkeypatch.setattr(
rest_endpoints.global_mcp_server_manager,
"get_allowed_mcp_servers",
fake_get_allowed_mcp_servers,
raising=False,
)
monkeypatch.setattr(
rest_endpoints.global_mcp_server_manager,
"get_mcp_server_by_id",
lambda server_id: StubServer() if server_id == "server-1" else None,
raising=False,
)
monkeypatch.setattr(
proxy_server, "add_litellm_data_to_request", fake_add_litellm_data_to_request, raising=False
)
monkeypatch.setattr(proxy_server, "proxy_config", {}, raising=False)
monkeypatch.setattr(
proxy_server.proxy_logging_obj,
"pre_call_hook",
blocking_pre_call_hook if raise_site == "pre_call_hook" else passthrough_pre_call_hook,
)
monkeypatch.setattr(
rest_endpoints,
"execute_mcp_tool",
fake_execute_mcp_tool if raise_site == "execute_mcp_tool" else passthrough_execute_mcp_tool,
raising=False,
)
post_call_failure_hook = AsyncMock(return_value=None)
monkeypatch.setattr(proxy_server.proxy_logging_obj, "post_call_failure_hook", post_call_failure_hook)
user_api_key_dict = UserAPIKeyAuth(api_key="hashed-key", request_route="/mcp-rest/tools/call")
request = _build_request(
headers={"x-mcp-deepwiki-authorization": "Bearer upstream-secret"},
path="/mcp-rest/tools/call",
method="POST",
json_body={"server_id": "server-1", "name": "demo-tool", "arguments": {"q": "confidential"}},
)
with pytest.raises(HTTPException) as exc_info:
await rest_endpoints.call_tool_rest_api(request, user_api_key_dict=user_api_key_dict)
assert exc_info.value is guardrail_error
post_call_failure_hook.assert_awaited_once()
hook_kwargs = post_call_failure_hook.await_args.kwargs
assert hook_kwargs["original_exception"] is guardrail_error
assert hook_kwargs["user_api_key_dict"] is user_api_key_dict
assert hook_kwargs["route"] == "/mcp/call_tool"
request_data = hook_kwargs["request_data"]
assert "raw_headers" not in request_data
assert "mcp_server_auth_headers" not in request_data
standard_logging_object = request_data["litellm_logging_obj"].model_call_details["standard_logging_object"]
assert standard_logging_object["status"] == "failure"
assert standard_logging_object["error_str"] == str(guardrail_error)
async def test_failure_logging_error_does_not_replace_guardrail_error(self, monkeypatch):
from litellm.proxy import proxy_server
guardrail_error = HTTPException(status_code=400, detail={"error": "Content blocked"})
async def fake_add_litellm_data_to_request(**kwargs):
return kwargs.get("data", {})
async def blocking_pre_call_hook(user_api_key_dict, data, call_type):
raise guardrail_error
failure_logging = AsyncMock(side_effect=RuntimeError("spend log db down"))
monkeypatch.setattr(
proxy_server, "add_litellm_data_to_request", fake_add_litellm_data_to_request, raising=False
)
monkeypatch.setattr(proxy_server, "proxy_config", {}, raising=False)
monkeypatch.setattr(proxy_server.proxy_logging_obj, "pre_call_hook", blocking_pre_call_hook)
monkeypatch.setattr(rest_endpoints, "fire_mcp_tool_call_failure_logging", failure_logging, raising=False)
request = _build_request(
path="/mcp-rest/tools/call",
method="POST",
json_body={"server_id": "server-1", "name": "demo-tool", "arguments": {"q": "confidential"}},
)
with pytest.raises(HTTPException) as exc_info:
await rest_endpoints.call_tool_rest_api(
request, user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key", request_route="/mcp-rest/tools/call")
)
assert exc_info.value is guardrail_error
failure_logging.assert_awaited_once()
async def test_success_logging_cancellation_propagates(self, monkeypatch):
fire_logging = AsyncMock(side_effect=asyncio.CancelledError())
monkeypatch.setattr(
@ -2801,7 +3007,7 @@ class TestCallToolRestAPI:
class _FakePreCall:
def __init__(self, data):
pass
self.data = data
async def common_processing_pre_call_logic(self, **kwargs):
return None, MagicMock()
@ -2835,6 +3041,58 @@ class TestCallToolRestAPI:
assert exc_info.value.headers is not None
assert exc_info.value.headers.get("www-authenticate") == challenge
async def test_virtual_mcp_tool_call_guardrail_block_runs_failure_logging(self, monkeypatch):
"""A pre_mcp_call guardrail block on the virtual mcp_tool_call branch must write a failure
spend log, same as the direct tool call branch, and still raise the original error."""
from litellm.proxy import proxy_server
from litellm.proxy._types import LiteLLM_ObjectPermissionTable
guardrail_error = HTTPException(status_code=400, detail={"error": "Content blocked"})
async def fake_contexts(user_api_key_auth):
return [user_api_key_auth]
async def fake_add_litellm_data_to_request(**kwargs):
return kwargs.get("data", {})
async def blocking_pre_call_hook(user_api_key_dict, data, call_type):
raise guardrail_error
failure_logging = AsyncMock()
monkeypatch.setattr(rest_endpoints, "build_effective_auth_contexts", fake_contexts, raising=False)
monkeypatch.setattr(
proxy_server, "add_litellm_data_to_request", fake_add_litellm_data_to_request, raising=False
)
monkeypatch.setattr(proxy_server, "proxy_config", {}, raising=False)
monkeypatch.setattr(proxy_server, "general_settings", {}, raising=False)
monkeypatch.setattr(proxy_server.proxy_logging_obj, "pre_call_hook", blocking_pre_call_hook)
monkeypatch.setattr(rest_endpoints, "fire_mcp_tool_call_failure_logging", failure_logging, raising=False)
user_api_key_dict = UserAPIKeyAuth(
api_key="hashed-key",
request_route="/mcp-rest/tools/call",
object_permission=LiteLLM_ObjectPermissionTable(
object_permission_id="search-scope",
mcp_tool_search_enabled=True,
),
)
request = _build_request(
path="/mcp-rest/tools/call",
method="POST",
json_body={"name": "mcp_tool_call", "arguments": {"tool_name": "x", "arguments": {"q": "confidential"}}},
)
with pytest.raises(HTTPException) as exc_info:
await rest_endpoints.call_tool_rest_api(request, user_api_key_dict=user_api_key_dict)
assert exc_info.value is guardrail_error
failure_logging.assert_awaited_once()
logging_obj, exception, _start_time, user_api_key_auth, request_data = failure_logging.await_args.args
assert exception is guardrail_error
assert user_api_key_auth is user_api_key_dict
assert logging_obj is request_data.get("litellm_logging_obj")
assert logging_obj is not None
class TestGetToolsForSingleServer:
"""Test _get_tools_for_single_server with object_permission filtering"""