scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-01 07:03:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 129

fix(a2a): log not-granted and unknown agent rejections as failures

Browse source 
A2A calls rejected inside invoke_agent_a2a after auth (agent not granted 403, unknown agent 404) now go through post_call_failure_hook with the authenticated identity and the target agent, so failure callbacks and SpendLogs see them. Client responses are unchanged and nothing is dispatched upstream

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Moe Khalil 2026-10-31 19:14:55 +00:00
parent aea0e4f2d4
commit 0eacdb31ec
2 changed files with 227 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
litellm/proxy/agent_endpoints/a2a_endpoints.py
View file

@ -201,6 +201,35 @@ def _forwarding_headers(
return merged or None
async def _log_agent_rejection(
*,
proxy_logging_obj: ProxyLogging,
user_api_key_dict: UserAPIKeyAuth,
request_id: object,
method: str | None,
rejection: HTTPException,
target_agent_id: str,
model: str,
) -> None:
request_data: Final[dict[str, object]] = { # mutable-ok: the failure hook stamps its logging state onto it
"jsonrpc": "2.0",
"id": request_id,
"method": method,
"model": model,
"agent_id": target_agent_id,
"metadata": {"agent_id": target_agent_id},
}
try:
await proxy_logging_obj.post_call_failure_hook( # pyright: ignore[reportUnknownMemberType] # bare dict in sig
request_data=request_data,
original_exception=rejection,
user_api_key_dict=user_api_key_dict,
route="/a2a/{agent_id}",
)
except Exception: # noqa: BLE001 # a logging failure must not change the rejection the caller gets
verbose_proxy_logger.exception("Failed to log rejected A2A request for agent '%s'", target_agent_id)
def _jsonrpc_error(
request_id: object,
code: int,
@ -722,6 +751,15 @@ async def invoke_agent_a2a(
# Find the agent
agent: Final = await _get_agent(agent_id)
if agent is None:
await _log_agent_rejection(
proxy_logging_obj=proxy_logging_obj,
user_api_key_dict=user_api_key_dict,
request_id=request_id,
method=method,
rejection=HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found"),
target_agent_id=agent_id,
model=f"a2a_agent/{agent_id}",
)
return _jsonrpc_error(request_id, -32000, f"Agent '{agent_id}' not found", 404)
served_version: Final = _served_version(agent, request, original_method)
@ -731,10 +769,20 @@ async def invoke_agent_a2a(
user_api_key_auth=user_api_key_dict,
)
if not is_allowed:
raise HTTPException(
denial: Final = HTTPException(
status_code=403,
detail=f"Agent '{agent_id}' is not allowed for your key/team. Contact proxy admin for access.",
)
await _log_agent_rejection(
proxy_logging_obj=proxy_logging_obj,
user_api_key_dict=user_api_key_dict,
request_id=request_id,
method=method,
rejection=denial,
target_agent_id=agent.agent_id,
model=f"a2a_agent/{(as_str_mapping(agent.agent_card_params or {}) or {}).get('name', agent_id)}",
)
raise denial
user_api_key_dict.invoked_agent_id = agent.agent_id
strict_policy: Final = agent.identity_managed or bool(user_api_key_dict.requires_fresh_policy)

178
tests/unit/proxy/agent_endpoints/test_a2a_endpoints.py
View file

@ -12,12 +12,14 @@ import sys
from collections.abc import AsyncIterator, Awaitable, Callable, Mapping
from contextlib import AbstractContextManager, ExitStack
from dataclasses import dataclass
from types import MappingProxyType
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
import respx
from fastapi import HTTPException
from pydantic import BaseModel, ConfigDict
import litellm
@ -2941,3 +2943,179 @@ async def test_failed_upstream_stream_relays_its_body_only_when_that_is_a_json_o
"id": "req-1",
"method": "tasks/resubscribe",
}
class _RejectionRecorder(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.failure_payloads: Final[list[Mapping[str, object]]] = [] # mutable-ok: collects callback events
self.hook_calls: Final[list[tuple[Mapping[str, object], Exception]]] = [] # mutable-ok: collects hook calls
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:
self.failure_payloads.append(kwargs["standard_logging_object"])
async def async_post_call_failure_hook(
self, request_data, original_exception, user_api_key_dict, traceback_str=None
) -> HTTPException:
self.hook_calls.append((dict(request_data), original_exception))
return HTTPException(status_code=418, detail="transformed")
_SPOOFED_BODY_FIELDS: Final = MappingProxyType(
{
"agent_id": "spoof",
"litellm_call_id": "spoofed-call-id",
"metadata": {"agent_id": "spoof", "user_api_key_alias": "spoof", "user_api_key_team_id": "spoof"},
}
)
def _rejection_request(path_agent_id: str, request_id: object) -> MagicMock:
req: Final = MagicMock()
req.headers = {}
req.json = AsyncMock(
return_value={
"jsonrpc": "2.0",
"id": request_id,
"method": "SendMessage",
"params": {**_HELLO_MESSAGE_PARAMS, "litellm_call_id": "spoofed-param-call-id"},
**_SPOOFED_BODY_FIELDS,
}
)
return req
async def _invoke_rejected(
monkeypatch: pytest.MonkeyPatch,
*,
registered: bool,
proxy_logging_obj: object,
request_id: object = "rej-1",
) -> tuple[int, object, AsyncMock]:
from litellm.proxy import proxy_server
from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a
path_agent_id: Final = "denied-agent" if registered else "no-such-agent"
agent: Final = _make_agent_mock()
agent.agent_id = "registered-denied-id"
upstream: Final = AsyncMock()
monkeypatch.setattr(proxy_server, "proxy_logging_obj", proxy_logging_obj)
auth: Final = UserAPIKeyAuth(
api_key="hashed-caller",
key_alias="caller-alias",
user_id="caller-user",
team_id="caller-team",
request_route=f"/a2a/{path_agent_id}",
)
with ExitStack() as stack:
stack.enter_context(
patch("litellm.proxy.agent_endpoints.a2a_endpoints._get_agent", return_value=agent if registered else None)
)
stack.enter_context(
patch(
"litellm.proxy.agent_endpoints.auth.agent_permission_handler.AgentRequestHandler.is_agent_allowed",
new=AsyncMock(return_value=False),
)
)
stack.enter_context(patch("litellm.a2a_protocol.asend_message", new=upstream))
stack.enter_context(patch("litellm.proxy.agent_endpoints.a2a_endpoints._handle_stream_message", new=upstream))
try:
response: Final = await invoke_agent_a2a(
agent_id=path_agent_id,
request=_rejection_request(path_agent_id, request_id),
fastapi_response=MagicMock(),
user_api_key_dict=auth,
)
except HTTPException as e:
return e.status_code, e.detail, upstream
return response.status_code, json.loads(response.body.decode()), upstream
_EXPECTED_REJECTIONS: Final = (
pytest.param(
True,
403,
"Agent 'denied-agent' is not allowed for your key/team. Contact proxy admin for access.",
"registered-denied-id",
"a2a_agent/Test Agent",
id="not-granted",
),
pytest.param(
False,
404,
{"jsonrpc": "2.0", "id": "rej-1", "error": {"code": -32000, "message": "Agent 'no-such-agent' not found"}},
"no-such-agent",
"a2a_agent/no-such-agent",
id="unknown",
),
)
@pytest.mark.asyncio
@pytest.mark.parametrize(("registered", "status", "body", "target", "model"), _EXPECTED_REJECTIONS)
async def test_rejected_agent_call_is_logged_once_as_the_caller_without_dispatch(
monkeypatch: pytest.MonkeyPatch, registered: bool, status: int, body: object, target: str, model: str
) -> None:
from litellm.caching.caching import DualCache
from litellm.proxy.utils import ProxyLogging
recorder: Final = _RejectionRecorder()
for name in ("callbacks", "failure_callback", "_async_failure_callback"):
monkeypatch.setattr(litellm, name, [recorder])
got_status, got_body, upstream = await _invoke_rejected(
monkeypatch, registered=registered, proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache())
)
assert (got_status, got_body) == (status, body)
upstream.assert_not_awaited()
assert len(recorder.failure_payloads) == 1
payload: Final = recorder.failure_payloads[0]
metadata: Final = payload["metadata"]
assert (payload["status"], payload["call_type"], payload["model"], payload["response_cost"]) == (
"failure",
"asend_message",
model,
0,
)
assert (
metadata["user_api_key_alias"],
metadata["user_api_key_user_id"],
metadata["user_api_key_team_id"],
metadata["user_api_key_hash"],
) == ("caller-alias", "caller-user", "caller-team", "hashed-caller")
assert payload["id"] not in {"spoofed-call-id", "spoofed-param-call-id", "rej-1"}
assert len(recorder.hook_calls) == 1
hook_data, hook_exception = recorder.hook_calls[0]
assert (hook_data["agent_id"], hook_data["metadata"]["agent_id"]) == (target, target)
assert isinstance(hook_exception, HTTPException) and hook_exception.status_code == status
@pytest.mark.asyncio
@pytest.mark.parametrize(("registered", "status", "body", "target", "model"), _EXPECTED_REJECTIONS)
async def test_rejected_agent_call_response_survives_a_failing_failure_hook(
monkeypatch: pytest.MonkeyPatch, registered: bool, status: int, body: object, target: str, model: str
) -> None:
proxy_logging_obj: Final = MagicMock()
proxy_logging_obj.post_call_failure_hook = AsyncMock(side_effect=RuntimeError("logging down"))
got_status, got_body, upstream = await _invoke_rejected(
monkeypatch, registered=registered, proxy_logging_obj=proxy_logging_obj
)
assert (got_status, got_body) == (status, body)
upstream.assert_not_awaited()
proxy_logging_obj.post_call_failure_hook.assert_awaited_once()
@pytest.mark.asyncio
@pytest.mark.parametrize("request_id", [0, 7, "abc"])
async def test_unknown_agent_rejection_echoes_the_jsonrpc_id(monkeypatch: pytest.MonkeyPatch, request_id: object) -> None:
proxy_logging_obj: Final = MagicMock()
proxy_logging_obj.post_call_failure_hook = AsyncMock(return_value=None)
_, got_body, _ = await _invoke_rejected(
monkeypatch, registered=False, proxy_logging_obj=proxy_logging_obj, request_id=request_id
)
assert isinstance(got_body, dict) and got_body["id"] == request_id and type(got_body["id"]) is type(request_id)