diff --git a/litellm/llms/litellm_proxy/skills/skill_search.py b/litellm/llms/litellm_proxy/skills/skill_search.py index 22d3f5b6dd1..d10f35680f5 100644 --- a/litellm/llms/litellm_proxy/skills/skill_search.py +++ b/litellm/llms/litellm_proxy/skills/skill_search.py @@ -131,7 +131,7 @@ async def search_skills( user_api_key_dict: UserAPIKeyAuth, proxy_logging_obj: ProxyLogging, ) -> SkillSearchOutcome: - if embedding_model is None: + if embedding_model is None or not embedding_model.strip(): return SkillSearchNotConfigured( reason="skill search needs litellm_settings.skill_search_embedding_model set to an embedding model from model_list" ) diff --git a/litellm/proxy/_lazy_features.py b/litellm/proxy/_lazy_features.py index dc6575f1e1b..b14f43635b2 100644 --- a/litellm/proxy/_lazy_features.py +++ b/litellm/proxy/_lazy_features.py @@ -105,10 +105,10 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = ( LazyFeature( name="a2a", module_path="litellm.proxy.agent_endpoints.a2a_endpoints", - # ``/v1/a2a/{agent_id}/message/send`` is caught via the suffix so the + # ``/v1/a2a/{agent_id}/message/send`` and ``/message/stream`` are caught via suffixes so the # ``/v1/a2a`` prefix doesn't subsume the discover prefix below. path_prefixes=("/a2a",), - path_suffixes=("/message/send",), + path_suffixes=("/message/send", "/message/stream"), ), LazyFeature( name="a2a_registration", diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index 546022bc0c4..2bbc7f6500a 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -54,6 +54,57 @@ ], "title": "ValidationError", "type": "object" + }, + "_A2AJsonRpcResponse": { + "additionalProperties": true, + "properties": { + "error": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "title": "Error" + }, + "id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Id" + }, + "jsonrpc": { + "const": "2.0", + "title": "Jsonrpc", + "type": "string" + }, + "result": { + "anyOf": [ + {}, + { + "type": "null" + } + ], + "title": "Result" + } + }, + "required": [ + "jsonrpc", + "id" + ], + "title": "_A2AJsonRpcResponse", + "type": "object" } } }, @@ -242,6 +293,119 @@ ] } }, + "/a2a/{agent_id}/message/stream": { + "post": { + "description": "Invoke an agent using the A2A protocol (JSON-RPC 2.0).\n\nSupported methods:\n- message/send: Send a message and get a response\n- message/stream: Send a message and stream the response", + "operationId": "invoke_agent_a2a_a2a__agent_id__message_stream_post", + "parameters": [ + { + "in": "path", + "name": "agent_id", + "required": true, + "schema": { + "title": "Agent Id", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "example": { + "id": "stream-1", + "jsonrpc": "2.0", + "method": "message/stream", + "params": { + "message": { + "kind": "message", + "messageId": "message-1", + "parts": [ + { + "kind": "text", + "text": "Hello" + } + ], + "role": "user" + } + } + }, + "schema": { + "additionalProperties": true, + "properties": { + "id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Id" + }, + "jsonrpc": { + "const": "2.0", + "title": "Jsonrpc", + "type": "string" + }, + "method": { + "title": "Method", + "type": "string" + }, + "params": { + "additionalProperties": true, + "title": "Params", + "type": "object" + } + }, + "required": [ + "jsonrpc", + "method", + "params" + ], + "title": "_A2AJsonRpcRequest", + "type": "object" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/_A2AJsonRpcResponse" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "security": [ + { + "APIKeyHeader": [] + } + ], + "summary": "Invoke Agent A2A", + "tags": [ + "a2a" + ] + } + }, "/v1/a2a/{agent_id}/message/send": { "post": { "description": "Invoke an agent using the A2A protocol (JSON-RPC 2.0).\n\nSupported methods:\n- message/send: Send a message and get a response\n- message/stream: Send a message and stream the response", @@ -287,6 +451,119 @@ "a2a" ] } + }, + "/v1/a2a/{agent_id}/message/stream": { + "post": { + "description": "Invoke an agent using the A2A protocol (JSON-RPC 2.0).\n\nSupported methods:\n- message/send: Send a message and get a response\n- message/stream: Send a message and stream the response", + "operationId": "invoke_agent_a2a_v1_a2a__agent_id__message_stream_post", + "parameters": [ + { + "in": "path", + "name": "agent_id", + "required": true, + "schema": { + "title": "Agent Id", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "example": { + "id": "stream-1", + "jsonrpc": "2.0", + "method": "message/stream", + "params": { + "message": { + "kind": "message", + "messageId": "message-1", + "parts": [ + { + "kind": "text", + "text": "Hello" + } + ], + "role": "user" + } + } + }, + "schema": { + "additionalProperties": true, + "properties": { + "id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Id" + }, + "jsonrpc": { + "const": "2.0", + "title": "Jsonrpc", + "type": "string" + }, + "method": { + "title": "Method", + "type": "string" + }, + "params": { + "additionalProperties": true, + "title": "Params", + "type": "object" + } + }, + "required": [ + "jsonrpc", + "method", + "params" + ], + "title": "_A2AJsonRpcRequest", + "type": "object" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/_A2AJsonRpcResponse" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "security": [ + { + "APIKeyHeader": [] + } + ], + "summary": "Invoke Agent A2A", + "tags": [ + "a2a" + ] + } } } }, diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 7b7bfc55ade..c9960d794c6 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -501,6 +501,7 @@ class LiteLLMRoutes(enum.Enum): "/a2a/{agent_id}/message/send", "/v1/a2a/{agent_id}/message/send", "/a2a/{agent_id}/message/stream", + "/v1/a2a/{agent_id}/message/stream", "/a2a/{agent_id}/.well-known/agent-card.json", ) diff --git a/litellm/proxy/agent_endpoints/a2a_endpoints.py b/litellm/proxy/agent_endpoints/a2a_endpoints.py index 227a1b2bdf0..fb5e9e392f8 100644 --- a/litellm/proxy/agent_endpoints/a2a_endpoints.py +++ b/litellm/proxy/agent_endpoints/a2a_endpoints.py @@ -14,7 +14,7 @@ import json from collections.abc import AsyncGenerator, Callable, Coroutine, Mapping from copy import deepcopy from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Final, Protocol +from typing import TYPE_CHECKING, Any, Final, Literal, Protocol from urllib.parse import urlparse from fastapi import APIRouter, Depends, HTTPException, Request, Response @@ -43,6 +43,7 @@ from litellm.proxy.agent_endpoints.databricks_oauth import ( ) from litellm.proxy.agent_endpoints.utils import merge_agent_headers from litellm.proxy.auth.user_api_key_auth import user_api_key_auth +from litellm.proxy.common_utils.custom_openapi_spec import inline_request_body from litellm.proxy.common_utils.openai_error_payload import status_typed_proxy_exception from litellm.proxy.common_utils.sse_keepalive import ( SSE_COMMENT_PING, @@ -59,6 +60,42 @@ if TYPE_CHECKING: router: Final = APIRouter() + +class _A2AJsonRpcRequest(BaseModel): + model_config = ConfigDict(extra="allow", frozen=True) + + jsonrpc: Literal["2.0"] + id: str | int | None = None + method: str + params: Mapping[str, object] + + +class _A2AJsonRpcResponse(BaseModel): + model_config = ConfigDict(extra="allow", frozen=True) + + jsonrpc: Literal["2.0"] + id: str | int | None + result: object | None = None + error: Mapping[str, object] | None = None + + +_A2A_MESSAGE_STREAM_REQUEST_BODY: Final = inline_request_body( + _A2AJsonRpcRequest, + { + "jsonrpc": "2.0", + "id": "stream-1", + "method": "message/stream", + "params": { + "message": { + "messageId": "message-1", + "role": "user", + "parts": [{"kind": "text", "text": "Hello"}], + "kind": "message", + } + }, + }, +) + # Mirrors the native seam's own headers: a reverse proxy that batches the whole # stream would swallow the keepalives this route sends to defeat idle timeouts. _SSE_KEEPALIVE_HEADERS: Final[Mapping[str, str]] = MappingProxyType( @@ -206,6 +243,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") + + def _jsonrpc_error( request_id: object, code: int, @@ -717,6 +783,20 @@ async def get_agent_card( tags=["[beta] A2A Agents"], dependencies=[Depends(user_api_key_auth)], ) +@router.post( + "/a2a/{agent_id}/message/stream", + tags=["[beta] A2A Agents"], + dependencies=[Depends(user_api_key_auth)], + openapi_extra={"requestBody": _A2A_MESSAGE_STREAM_REQUEST_BODY}, + responses={200: {"model": _A2AJsonRpcResponse}}, +) +@router.post( + "/v1/a2a/{agent_id}/message/stream", + tags=["[beta] A2A Agents"], + dependencies=[Depends(user_api_key_auth)], + openapi_extra={"requestBody": _A2A_MESSAGE_STREAM_REQUEST_BODY}, + responses={200: {"model": _A2AJsonRpcResponse}}, +) async def invoke_agent_a2a( agent_id: str, request: Request, @@ -778,6 +858,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) @@ -787,10 +876,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) diff --git a/litellm/proxy/agent_endpoints/agent_search.py b/litellm/proxy/agent_endpoints/agent_search.py index af2baa2bd89..0443ec1f045 100644 --- a/litellm/proxy/agent_endpoints/agent_search.py +++ b/litellm/proxy/agent_endpoints/agent_search.py @@ -140,7 +140,7 @@ async def search_agents( user_api_key_dict: UserAPIKeyAuth, proxy_logging_obj: ProxyLogging, ) -> AgentSearchOutcome: - if embedding_model is None: + if embedding_model is None or not embedding_model.strip(): return AgentSearchNotConfigured( reason="agent search needs litellm_settings.agent_search_embedding_model set to an embedding model from model_list" ) diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index f7f0cf2ab3b..c7983f40aec 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -3614,7 +3614,9 @@ async def deepgram_listen_websocket_route( if not deepgram_listen_is_priced(target): await websocket.close( code=1008, - reason=_DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target)), + reason=truncated_close_reason( + _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target)) + ), ) return diff --git a/litellm/router_strategy/auto_router/litellm_encoder.py b/litellm/router_strategy/auto_router/litellm_encoder.py index 340efb509eb..743a0b97fd2 100644 --- a/litellm/router_strategy/auto_router/litellm_encoder.py +++ b/litellm/router_strategy/auto_router/litellm_encoder.py @@ -14,13 +14,13 @@ else: RoutingContext = Any -def litellm_to_list(embeds: litellm.EmbeddingResponse) -> list[list[float]]: +def litellm_to_list(embeds: object) -> list[list[float]]: """Convert a LiteLLM embedding response to a list of embeddings. :param embeds: The LiteLLM embedding response. :return: A list of embeddings. """ - if not embeds or not embeds.data: + if not embeds or not isinstance(embeds, litellm.EmbeddingResponse) or not embeds.data: raise ValueError("No embeddings found in LiteLLM embedding response.") return [x["embedding"] for x in embeds.data] diff --git a/litellm/router_strategy/complexity_router/jev_classifier.py b/litellm/router_strategy/complexity_router/jev_classifier.py index 89e9dec96f1..bb7d2e91cd3 100644 --- a/litellm/router_strategy/complexity_router/jev_classifier.py +++ b/litellm/router_strategy/complexity_router/jev_classifier.py @@ -68,6 +68,7 @@ class JevBudgetedRequest: telemetry: StandardLoggingRoutingDecision +@renamed_router_param("router", "context") def jev_input_targets(context: "RoutingContext", config: "OpenSourceClassifierConfig") -> tuple[JevInputTarget, ...]: from litellm.types.router import Deployment diff --git a/litellm/router_strategy/tag_based_routing.py b/litellm/router_strategy/tag_based_routing.py index dca717100bf..f2aafa5b7df 100644 --- a/litellm/router_strategy/tag_based_routing.py +++ b/litellm/router_strategy/tag_based_routing.py @@ -392,9 +392,11 @@ def _resolve_constraint_only_pool( def _all_deployments_or_fallback( ctx: "RoutingContext", - model: str, + model: str | None, fallback: _DeploymentPool, ) -> Sequence[_DeploymentLike | DeploymentTypedDict] | Mapping[_DeploymentLike, object]: + if model is None: # a group of no name has no deployments, it is not a failed lookup + return () try: return ctx.registry().deployments(model, source="all") except Exception: # noqa: BLE001 # fail safe toward today's healthy-only behavior on lookup errors diff --git a/litellm/router_utils/handle_error.py b/litellm/router_utils/handle_error.py index ad502fe68aa..6033024123f 100644 --- a/litellm/router_utils/handle_error.py +++ b/litellm/router_utils/handle_error.py @@ -23,7 +23,7 @@ else: @renamed_router_param("litellm_router_instance") async def send_llm_exception_alert( - ctx: "RoutingContext", + ctx: "RoutingContext | None", request_kwargs: dict, error_traceback_str: str, original_exception, @@ -39,6 +39,9 @@ async def send_llm_exception_alert( Returns: None """ + if ctx is None: + return + slack_alerting_logger: Final = ctx.observers().slack_alerting_logger if slack_alerting_logger is None: return diff --git a/litellm/types/utils.py b/litellm/types/utils.py index f4763acaeda..da1d99957a7 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -1151,6 +1151,8 @@ API_ROUTE_TO_CALL_TYPES: Final[Mapping[str, Sequence[CallTypes]]] = { # A2A (Agent-to-Agent) "/a2a/{agent_id}": [CallTypes.asend_message, CallTypes.send_message], "/a2a/{agent_id}/message/send": [CallTypes.asend_message, CallTypes.send_message], + "/a2a/{agent_id}/message/stream": [CallTypes.asend_message, CallTypes.send_message], + "/v1/a2a/{agent_id}/message/stream": [CallTypes.asend_message, CallTypes.send_message], # Passthrough endpoints "/llm_passthrough": [ CallTypes.allm_passthrough_route, diff --git a/tests/integration/_support/a2a_denial_probe.py b/tests/integration/_support/a2a_denial_probe.py new file mode 100644 index 00000000000..adfebbb8b36 --- /dev/null +++ b/tests/integration/_support/a2a_denial_probe.py @@ -0,0 +1,145 @@ +import json +import os +from collections.abc import Mapping +from datetime import datetime +from typing import Final + +import httpx +from fastapi import HTTPException + +from litellm.integrations.custom_logger import CustomLogger +from litellm.proxy._types import UserAPIKeyAuth + + +def _mapping(value: object) -> Mapping[str, object]: + return value if isinstance(value, Mapping) else {} + + +def _logged(hook: str, kwargs: Mapping[str, object]) -> Mapping[str, object]: + logged: Final = _mapping(kwargs.get("standard_logging_object")) + metadata: Final = _mapping(logged.get("metadata")) + return { + "hook": hook, + "call_type": kwargs.get("call_type"), + "litellm_call_id": kwargs.get("litellm_call_id"), + "model": logged.get("model"), + "status": logged.get("status"), + "key_alias": metadata.get("user_api_key_alias"), + "key_hash": metadata.get("user_api_key_hash"), + "user_id": metadata.get("user_api_key_user_id"), + "team_id": metadata.get("user_api_key_team_id"), + } + + +def _status(exception: Exception) -> int | None: + status: Final = getattr(exception, "status_code", None) + if isinstance(status, int): + return status + code: Final = getattr(exception, "code", None) + return int(code) if isinstance(code, str) and code.isdigit() else None + + +def _hooked( + request_data: Mapping[str, object], user_api_key_dict: UserAPIKeyAuth, original_exception: Exception +) -> Mapping[str, object]: + return { + "hook": "post_call_failure", + "call_type": request_data.get("call_type"), + "litellm_call_id": request_data.get("litellm_call_id"), + "key_alias": user_api_key_dict.key_alias, + "key_hash": user_api_key_dict.api_key, + "user_id": user_api_key_dict.user_id, + "team_id": user_api_key_dict.team_id, + "agent_id": request_data.get("agent_id"), + "metadata_agent_id": _mapping(request_data.get("metadata")).get("agent_id"), + "status_code": _status(original_exception), + } + + +def _send(record: Mapping[str, object]) -> None: + with httpx.Client(trust_env=False) as client: + client.post(os.environ["A2A_DENIAL_PROBE_URL"], content=json.dumps(record, default=str)).raise_for_status() + + +async def _send_async(record: Mapping[str, object]) -> None: + async with httpx.AsyncClient(trust_env=False) as client: + response: Final = await client.post(os.environ["A2A_DENIAL_PROBE_URL"], content=json.dumps(record, default=str)) + response.raise_for_status() + + +class A2ADenialProbe(CustomLogger): + """Reports the identity, target and outcome every logging and failure hook receives, one record per call.""" + + async def async_log_success_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime + ) -> None: + await _send_async(_logged("async_success", kwargs)) + + def log_failure_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime + ) -> None: + _send(_logged("sync_failure", kwargs)) + + async def async_log_failure_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime + ) -> None: + await _send_async(_logged("async_failure", kwargs)) + + async def async_post_call_failure_hook( + self, + request_data: dict[str, object], + original_exception: Exception, + user_api_key_dict: UserAPIKeyAuth, + traceback_str: str | None = None, + ) -> None: + await _send_async(_hooked(request_data, user_api_key_dict, original_exception)) + + +class RaisingFailureLogger(CustomLogger): + """Fails inside both failure hooks.""" + + async def async_log_failure_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime + ) -> None: + raise RuntimeError("failure logger is broken") + + async def async_post_call_failure_hook( + self, + request_data: dict[str, object], + original_exception: Exception, + user_api_key_dict: UserAPIKeyAuth, + traceback_str: str | None = None, + ) -> None: + raise RuntimeError("failure hook is broken") + + +class ReturningTeapotLogger(CustomLogger): + """Hands a replacement 418 back from the failure hook.""" + + async def async_post_call_failure_hook( + self, + request_data: dict[str, object], + original_exception: Exception, + user_api_key_dict: UserAPIKeyAuth, + traceback_str: str | None = None, + ) -> HTTPException: + return HTTPException(status_code=418, detail="transformed") + + +class RaisingTeapotLogger(CustomLogger): + """Raises a replacement 418 from the failure hook.""" + + async def async_post_call_failure_hook( + self, + request_data: dict[str, object], + original_exception: Exception, + user_api_key_dict: UserAPIKeyAuth, + traceback_str: str | None = None, + ) -> None: + raise HTTPException(status_code=418, detail="transformed") + + +probe: Final = A2ADenialProbe() +raising_failure_logger: Final = RaisingFailureLogger() +returning_teapot_logger: Final = ReturningTeapotLogger() +raising_teapot_logger: Final = RaisingTeapotLogger() diff --git a/tests/integration/authorization/test_agent_denial_logging.py b/tests/integration/authorization/test_agent_denial_logging.py new file mode 100644 index 00000000000..8e990b07889 --- /dev/null +++ b/tests/integration/authorization/test_agent_denial_logging.py @@ -0,0 +1,809 @@ +import hashlib +import json +import uuid +from collections.abc import Iterator, Mapping +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType +from typing import Final + +import httpx +import pytest +import yaml +from integration._support.client import ( + JSON_OBJECT, + Gateway, + Scenario, + eventually, + gateway_from_environment, + held, + object_value, + string_value, +) +from integration._support.process import owned_proxy +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import BaseModel, ConfigDict, JsonValue + +pytestmark = pytest.mark.timeout(900) + +_PROBE: Final = "integration._support.a2a_denial_probe." +_SETTLE: Final = 3.0 +_CONVERGE: Final = 30.0 +_NO_SETTINGS: Final[Mapping[str, JsonValue]] = MappingProxyType({}) +_ROWS: Final = ( + 'SELECT request_id, call_type, status, model, agent_id, spend, api_key, "user", team_id, metadata ' + 'FROM "LiteLLM_SpendLogs" WHERE api_key = %s' +) +_ONE_FAILURE: Final = ("async_failure", "post_call_failure") +_STREAM_METHODS: Final = frozenset({"message/stream", "SendStreamingMessage"}) + + +class _Event(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + hook: str + call_type: str | None = None + litellm_call_id: str | None = None + model: str | None = None + status: str | None = None + key_alias: str | None = None + key_hash: str | None = None + user_id: str | None = None + team_id: str | None = None + agent_id: str | None = None + metadata_agent_id: str | None = None + status_code: int | None = None + + +class _Events: + def __init__(self, sink: Wire) -> None: + self.sink: Final = sink + self.seen: tuple[_Event, ...] = () + + def of(self, alias: str) -> tuple[_Event, ...]: + self.seen = (*self.seen, *(_Event.model_validate_json(request.body) for request in self.sink.drain())) + return tuple(event for event in self.seen if event.key_alias == alias) + + +@dataclass(frozen=True, slots=True) +class _Agent: + agent_id: str + name: str + + +@dataclass(frozen=True, slots=True) +class _Rig: + proxy: Gateway + events: _Events + upstream: Wire + allowed: _Agent + denied: _Agent + denied_legacy: _Agent + bridge: _Agent + + +@dataclass(frozen=True, slots=True) +class _Caller: + key: str + alias: str + user_id: str + team_id: str | None + + @property + def hashed(self) -> str: + return hashlib.sha256(self.key.encode()).hexdigest() + + +def _compact(content: Mapping[str, JsonValue]) -> bytes: + return json.dumps(content, separators=(",", ":")).encode() + + +def _refusal(agent: str, response: httpx.Response | None = None) -> bytes: + message: Final = f"Agent '{agent}' is not allowed for your key/team. Contact proxy admin for access." + if response is None: + return _compact({"detail": message}) + return _compact( + { + "jsonrpc": "2.0", + "id": JSON_OBJECT.validate_json(response.request.content)["id"], + "error": { + "code": -32000, + "message": message, + "data": {"type": "permission_error", "param": None, "code": "403"}, + }, + } + ) + + +def _not_found(request_id: JsonValue, agent: str) -> bytes: + return _compact( + {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32000, "message": f"Agent '{agent}' not found"}} + ) + + +def _card(url: str, name: str, version: str) -> dict[str, JsonValue]: + card: Final[dict[str, JsonValue]] = { + "protocolVersion": version, + "name": name, + "description": "Synthetic peer", + "version": "1.0.0", + "url": url + "/", + "capabilities": {"streaming": True}, + "defaultInputModes": ["text"], + "defaultOutputModes": ["text"], + "skills": [], + } + if version == "0.3": + return card + return {**card, "supportedInterfaces": [{"url": url + "/", "protocolBinding": "JSONRPC", "protocolVersion": "1.0"}]} + + +def _peer(request: Request) -> Reply: + if request.method == "GET": + return Reply(body=_compact(_card("http://" + request.headers["host"], "peer", "0.3"))) + body: Final = JSON_OBJECT.validate_json(request.body) + if body.get("method") == "message/stream": + frame: Final = { + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "kind": "status-update", + "taskId": "t1", + "contextId": "c1", + "status": {"state": "completed"}, + "final": True, + }, + } + return Reply(content_type="text/event-stream", chunks=(f"data: {json.dumps(frame)}\n\n".encode(),)) + return Reply( + body=_compact( + { + "jsonrpc": "2.0", + "id": body.get("id"), + "result": { + "kind": "message", + "role": "agent", + "messageId": "peer-message", + "parts": [{"kind": "text", "text": "granted"}], + }, + } + ) + ) + + +def _register(proxy: Gateway, scenario: Scenario, body: Mapping[str, JsonValue]) -> _Agent: + created: Final = proxy.request("POST", "/v1/agents", body) + assert created.status_code == 200, created.text + agent: Final = _Agent(string_value(JSON_OBJECT.validate_json(created.content)["agent_id"]), str(body["agent_name"])) + + def cleanup() -> None: + deleted: Final = proxy.request("DELETE", f"/v1/agents/{agent.agent_id}") + assert deleted.status_code == 200, deleted.text + + scenario.cleanups.callback(cleanup) + return agent + + +def _config(directory: Path, callbacks: tuple[str, ...], general: Mapping[str, JsonValue]) -> Path: + base: Final = JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())) + config: Final = { + **base, + "general_settings": {**object_value(base["general_settings"]), **general}, + "litellm_settings": { + **object_value(base["litellm_settings"]), + "callbacks": [_PROBE + "probe", *(_PROBE + name for name in callbacks)], + }, + } + path: Final = directory / "a2a-denials.yaml" + path.write_text(yaml.safe_dump(config)) + return path + + +@contextmanager +def _owned_rig( + directory: Path, + callbacks: tuple[str, ...] = (), + general: Mapping[str, JsonValue] = _NO_SETTINGS, + environment: Mapping[str, str] = MappingProxyType({}), +) -> Iterator[_Rig]: + marker: Final = "a2adeny" + uuid.uuid4().hex[:10] + with ( + wire_server(_peer) as upstream, + wire_server(lambda request: Reply()) as sink, + gateway_from_environment() as shared, + owned_proxy( + shared, + directory, + {"A2A_DENIAL_PROBE_URL": sink.url, **environment}, + config=_config(directory, callbacks, general), + ) as proxy, + proxy.scenario() as scenario, + ): + + def agent(suffix: str, version: str, **extra: JsonValue) -> _Agent: + name: Final = f"{marker}-{suffix}" + return _register( + proxy, scenario, {"agent_name": name, "agent_card_params": _card(upstream.url, name, version), **extra} + ) + + yield _Rig( + proxy, + _Events(sink), + upstream, + allowed=agent("allowed", "0.3"), + denied=agent("denied", "1.0"), + denied_legacy=agent("legacy", "0.3"), + bridge=agent( + "bridge", + "0.3", + litellm_params={ + "custom_llm_provider": "openai", + "model": "openai/gpt-5.4-mini", + "api_base": upstream.url + "/v1", + "api_key": "synthetic-bridge-key", + }, + ), + ) + + +@pytest.fixture(scope="module") +def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]: + with _owned_rig(tmp_path_factory.mktemp("a2a-denials")) as owned: + yield owned + + +@pytest.fixture(scope="module", params=["raising_failure_logger,returning_teapot_logger", "raising_teapot_logger"]) +def hostile(request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]: + callbacks: Final = tuple(str(request.param).split(",")) + with _owned_rig(tmp_path_factory.mktemp("a2a-denials-hostile"), callbacks) as owned: + yield owned + + +@pytest.fixture(scope="module", params=["disable_pre_routing_error_logs", "disable_error_logs"]) +def flagged(request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]: + with _owned_rig(tmp_path_factory.mktemp("a2a-denials-flagged"), general={str(request.param): True}) as owned: + yield owned + + +@pytest.fixture(scope="module") +def alerting(tmp_path_factory: pytest.TempPathFactory) -> Iterator[tuple[_Rig, Wire]]: + with wire_server(lambda request: Reply(body=b"ok", content_type="text/plain")) as alerts: + with _owned_rig( + tmp_path_factory.mktemp("a2a-denials-alerting"), + general={"alerting": ["slack"], "alert_types": ["llm_exceptions"]}, + environment={"SLACK_WEBHOOK_URL": alerts.url, "DEFAULT_FLUSH_INTERVAL_SECONDS": "1"}, + ) as owned: + yield owned, alerts + + +def _caller(rig: _Rig, scenario: Scenario, grant: str = "team") -> _Caller: + alias: Final = "a2adeny-" + uuid.uuid4().hex[:12] + user: Final = scenario.user() + allowed: Final[list[JsonValue]] = [rig.allowed.agent_id] + if grant == "personal": + return _Caller( + scenario.key(user_id=user, key_alias=alias, object_permission={"agents": allowed}), alias, user, None + ) + if grant == "access group": + created: Final = rig.proxy.request( + "POST", "/v1/access_group", {"access_group_name": alias, "access_agent_ids": allowed} + ) + assert created.status_code == 201, created.text + group: Final = string_value(JSON_OBJECT.validate_json(created.content)["access_group_id"]) + scenario.cleanups.callback(rig.proxy.request, "DELETE", f"/v1/access_group/{group}") + return _Caller(scenario.key(user_id=user, key_alias=alias, access_group_ids=[group]), alias, user, None) + team: Final = scenario.team(object_permission={"agents": allowed}) + return _Caller(scenario.key(user_id=user, team_id=team, key_alias=alias), alias, user, team) + + +def _params(method: str, message_id: str, version: str | None) -> dict[str, JsonValue]: + if method.startswith("tasks/pushNotificationConfig"): + return {"taskId": "t1", "pushNotificationConfig": {"url": "https://callback.example.com/hook"}} + if method in {"tasks/get", "tasks/cancel", "GetTask", "CancelTask"}: + return {"id": "t1"} + if version == "1.0": + return {"message": {"messageId": message_id, "role": "ROLE_USER", "parts": [{"text": "x"}]}} + return { + "message": { + "kind": "message", + "messageId": message_id, + "role": "user", + "parts": [{"kind": "text", "text": "x"}], + } + } + + +def _send( + rig: _Rig, + key: str, + path: str, + *, + method: str = "SendMessage", + version: str | None = "1.0", + request_id: JsonValue = None, + message_id: str | None = None, + extra: Mapping[str, JsonValue] = _NO_SETTINGS, +) -> httpx.Response: + headers: Final = { + "Authorization": f"Bearer {key}", + **({"A2A-Version": version} if version is not None else {}), + **({"Accept": "text/event-stream"} if method in _STREAM_METHODS else {}), + } + body: Final = { + "jsonrpc": "2.0", + "id": "r-" + uuid.uuid4().hex if request_id is None else request_id, + "method": method, + "params": _params(method, message_id or uuid.uuid4().hex, version), + **extra, + } + return rig.proxy.client.post(path, headers=headers, json=body) + + +def _assert_never_dispatched(rig: _Rig, scenario: Scenario, denied_message: str) -> None: + control_key: Final = scenario.key(object_permission={"agents": [rig.allowed.agent_id]}) + control_message: Final = "control-" + uuid.uuid4().hex + granted: Final = _send( + rig, + control_key, + f"/a2a/{rig.allowed.agent_id}", + method="message/send", + version=None, + message_id=control_message, + ) + assert granted.status_code == 200, granted.text + received: Final = tuple(request.body.decode() for request in rig.upstream.drain()) + assert any(control_message in body for body in received), received + assert not any(denied_message in body for body in received), received + + +def _metadata(row: Mapping[str, JsonValue]) -> dict[str, JsonValue]: + value: Final = row["metadata"] + return JSON_OBJECT.validate_json(value) if isinstance(value, str) else object_value(value) + + +def _rows(rig: _Rig, caller: _Caller) -> list[dict[str, JsonValue]]: + return rig.proxy.rows(_ROWS, (caller.hashed,)) + + +def _hooks(events: tuple[_Event, ...]) -> tuple[str, ...]: + return tuple(sorted(event.hook for event in events)) + + +def _spend(rig: _Rig, caller: _Caller) -> tuple[JsonValue, ...]: + key: Final = object_value(rig.proxy.get("/key/info", {"key": caller.key})["info"])["spend"] + user: Final = object_value(rig.proxy.get("/user/info", {"user_id": caller.user_id})["user_info"])["spend"] + if caller.team_id is None: + return key, user + team: Final = object_value(rig.proxy.get("/team/info", {"team_id": caller.team_id})["team_info"])["spend"] + return key, user, team + + +def _assert_one_failure(rig: _Rig, caller: _Caller, *, status: int, agent_id: str, model: str) -> dict[str, JsonValue]: + events: Final = held( + lambda: rig.events.of(caller.alias), + lambda found: _hooks(found) == _ONE_FAILURE, + holding=_SETTLE, + seconds=_CONVERGE, + ) + logged: Final = next(event for event in events if event.hook == "async_failure") + assert (logged.call_type, logged.status, logged.model) == ("asend_message", "failure", model), logged + assert (logged.key_hash, logged.user_id, logged.team_id) == (caller.hashed, caller.user_id, caller.team_id), logged + hooked: Final = next(event for event in events if event.hook == "post_call_failure") + assert (hooked.status_code, hooked.agent_id, hooked.key_hash) == (status, agent_id, caller.hashed), hooked + assert (hooked.user_id, hooked.team_id) == (caller.user_id, caller.team_id), hooked + rows: Final = held(lambda: _rows(rig, caller), lambda found: len(found) == 1, holding=_SETTLE, seconds=_CONVERGE) + row: Final = rows[0] + observed: Final = (row["status"], row["call_type"], row["agent_id"], row["api_key"], row["user"], row["spend"]) + assert observed == ("failure", "asend_message", agent_id, caller.hashed, caller.user_id, 0.0), row + assert (row["team_id"] or None) == caller.team_id, row + assert object_value(_metadata(row)["error_information"])["error_code"] == str(status), row + assert all(spend == 0.0 for spend in _spend(rig, caller)), _spend(rig, caller) + return row + + +@pytest.mark.parametrize( + ("route", "method", "version", "legacy"), + [ + pytest.param("/a2a/{}", "SendMessage", "1.0", False, id="regression-send-1.0"), + pytest.param("/a2a/{}/message/send", "SendMessage", "1.0", False, id="regression-send-alias"), + pytest.param("/v1/a2a/{}/message/send", "SendMessage", "1.0", False, id="regression-send-v1-alias"), + pytest.param("/a2a/{}", "SendStreamingMessage", "1.0", False, id="regression-stream-1.0"), + pytest.param("/a2a/{}", "GetTask", "1.0", False, id="regression-get-task-1.0"), + pytest.param("/a2a/{}", "CancelTask", "1.0", False, id="regression-cancel-task-1.0"), + pytest.param("/a2a/{}", "message/send", None, True, id="regression-send-0.3"), + pytest.param("/a2a/{}", "message/stream", None, True, id="regression-stream-0.3"), + pytest.param("/a2a/{}", "tasks/get", None, True, id="regression-tasks-get-0.3"), + pytest.param("/a2a/{}", "tasks/cancel", None, True, id="regression-tasks-cancel-0.3"), + pytest.param("/a2a/{}", "tasks/pushNotificationConfig/set", None, True, id="regression-push-set-0.3"), + ], +) +def test_denied_agent_keeps_its_403_logs_one_zero_spend_failure_for_the_caller_and_never_dispatches( + rig: _Rig, route: str, method: str, version: str | None, legacy: bool +) -> None: + target: Final = rig.denied_legacy if legacy else rig.denied + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + message: Final = "denied-" + uuid.uuid4().hex + response: Final = _send( + rig, caller.key, route.format(target.agent_id), method=method, version=version, message_id=message + ) + assert (response.status_code, response.content) == (403, _refusal(target.agent_id, response)) + assert response.headers["content-type"] == "application/json" + _assert_never_dispatched(rig, scenario, message) + _assert_one_failure(rig, caller, status=403, agent_id=target.agent_id, model=f"a2a_agent/{target.name}") + + +@pytest.mark.parametrize("grant", ["personal", "team", "access group"], ids=lambda grant: "regression-" + grant) +def test_denied_agent_failure_is_attributed_to_the_authenticated_key_for_every_grant_kind( + rig: _Rig, grant: str +) -> None: + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario, grant) + granted: Final = _send(rig, caller.key, f"/a2a/{rig.allowed.agent_id}", method="message/send", version=None) + assert granted.status_code == 200, granted.text + response: Final = _send(rig, caller.key, f"/a2a/{rig.denied.agent_id}") + assert (response.status_code, response.content) == (403, _refusal(rig.denied.agent_id, response)) + events: Final = held( + lambda: rig.events.of(caller.alias), + lambda found: _hooks(found) == tuple(sorted((*_ONE_FAILURE, "async_success"))), + holding=_SETTLE, + seconds=_CONVERGE, + ) + failure: Final = next(event for event in events if event.hook == "async_failure") + assert (failure.key_hash, failure.user_id, failure.team_id) == (caller.hashed, caller.user_id, caller.team_id) + rows: Final = held( + lambda: _rows(rig, caller), lambda found: len(found) == 2, holding=_SETTLE, seconds=_CONVERGE + ) + failed: Final = tuple(row for row in rows if row["status"] == "failure") + assert [(row["agent_id"], row["user"], row["spend"]) for row in failed] == [ + (rig.denied.agent_id, caller.user_id, 0.0) + ], rows + + +def test_denied_completion_bridge_agent_logs_one_failure_and_never_reaches_the_bridged_model(rig: _Rig) -> None: + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + message: Final = "bridge-" + uuid.uuid4().hex + response: Final = _send( + rig, caller.key, f"/a2a/{rig.bridge.agent_id}", method="message/send", version=None, message_id=message + ) + assert (response.status_code, response.content) == (403, _refusal(rig.bridge.agent_id, response)) + _assert_never_dispatched(rig, scenario, message) + _assert_one_failure(rig, caller, status=403, agent_id=rig.bridge.agent_id, model=f"a2a_agent/{rig.bridge.name}") + + +@pytest.mark.parametrize( + ("route", "method", "version"), + [ + pytest.param("/a2a/{}", "SendMessage", "1.0", id="regression-send"), + pytest.param("/a2a/{}/message/send", "SendMessage", "1.0", id="regression-send-alias"), + pytest.param("/v1/a2a/{}/message/send", "SendMessage", "1.0", id="regression-send-v1-alias"), + pytest.param("/a2a/{}", "message/stream", None, id="regression-stream-0.3"), + ], +) +def test_unknown_agent_keeps_its_jsonrpc_404_and_logs_one_zero_spend_failure_for_the_caller( + rig: _Rig, route: str, method: str, version: str | None +) -> None: + unknown: Final = "no-such-" + uuid.uuid4().hex + request_id: Final = "u-" + uuid.uuid4().hex + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + message: Final = "unknown-" + uuid.uuid4().hex + response: Final = _send( + rig, + caller.key, + route.format(unknown), + method=method, + version=version, + request_id=request_id, + message_id=message, + ) + assert (response.status_code, response.content) == (404, _not_found(request_id, unknown)) + assert response.headers["content-type"] == "application/json" + _assert_never_dispatched(rig, scenario, message) + _assert_one_failure(rig, caller, status=404, agent_id=unknown, model=f"a2a_agent/{unknown}") + + +@pytest.mark.parametrize("request_id", [0, 7, "abc"], ids=lambda value: f"regression-id-{value!r}") +def test_unknown_agent_echoes_the_jsonrpc_id_with_its_type_and_logs_one_failure( + rig: _Rig, request_id: JsonValue +) -> None: + unknown: Final = "no-such-" + uuid.uuid4().hex + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + response: Final = _send(rig, caller.key, f"/a2a/{unknown}", request_id=request_id) + assert (response.status_code, response.content) == (404, _not_found(request_id, unknown)) + assert type(JSON_OBJECT.validate_json(response.content)["id"]) is type(request_id) + _assert_one_failure(rig, caller, status=404, agent_id=unknown, model=f"a2a_agent/{unknown}") + + +@pytest.mark.parametrize("outcome", ["denied", "unknown"], ids=lambda outcome: "regression-" + outcome) +def test_client_supplied_attribution_cannot_rewrite_a_denial_failure_row(rig: _Rig, outcome: str) -> None: + unknown: Final = "no-such-" + uuid.uuid4().hex + target: Final = rig.denied.agent_id if outcome == "denied" else unknown + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + other: Final = _caller(rig, scenario) + success_id: Final = "s-" + uuid.uuid4().hex + granted: Final = _send( + rig, other.key, f"/a2a/{rig.allowed.agent_id}", method="message/send", version=None, request_id=success_id + ) + assert granted.status_code == 200, granted.text + success_rows: Final = eventually(lambda: _rows(rig, other), lambda found: len(found) == 1, seconds=_CONVERGE) + success_request: Final = success_rows[0]["request_id"] + request_id: Final = "spoof-" + uuid.uuid4().hex + spoof: Final[dict[str, JsonValue]] = { + "agent_id": "spoof", + "metadata": { + "agent_id": "spoof", + "user_api_key_alias": "spoof", + "user_api_key_team_id": "spoof", + "tags": ["x"], + }, + "litellm_call_id": success_request, + } + body: Final = { + "jsonrpc": "2.0", + "id": request_id, + "method": "SendMessage", + "params": {**_params("SendMessage", uuid.uuid4().hex, "1.0"), "litellm_call_id": "spoof2"}, + **spoof, + } + response: Final = rig.proxy.client.post( + f"/a2a/{target}", headers={"Authorization": f"Bearer {caller.key}", "A2A-Version": "1.0"}, json=body + ) + expected: Final = _refusal(target, response) if outcome == "denied" else _not_found(request_id, unknown) + assert (response.status_code, response.content) == (403 if outcome == "denied" else 404, expected) + model: Final = f"a2a_agent/{rig.denied.name}" if outcome == "denied" else f"a2a_agent/{unknown}" + row: Final = _assert_one_failure(rig, caller, status=response.status_code, agent_id=target, model=model) + assert row["request_id"] not in {success_request, "spoof2", request_id}, row + metadata: Final = _metadata(row) + assert (metadata["user_api_key_alias"], metadata["user_api_key_team_id"]) == (caller.alias, caller.team_id) + assert rig.events.of("spoof") == () + after: Final = _rows(rig, other) + assert [(item["request_id"], item["status"]) for item in after] == [(success_request, "success")], after + + +def test_two_keys_reusing_one_jsonrpc_id_get_one_denial_row_each(rig: _Rig) -> None: + shared_id: Final = "same-" + uuid.uuid4().hex + with rig.proxy.scenario() as scenario: + callers: Final = (_caller(rig, scenario), _caller(rig, scenario)) + for caller in callers: + response = _send(rig, caller.key, f"/a2a/{rig.denied.agent_id}", request_id=shared_id) + assert (response.status_code, response.content) == (403, _refusal(rig.denied.agent_id, response)) + rows: Final = tuple( + _assert_one_failure( + rig, caller, status=403, agent_id=rig.denied.agent_id, model=f"a2a_agent/{rig.denied.name}" + ) + for caller in callers + ) + assert rows[0]["request_id"] != rows[1]["request_id"], rows + + +@pytest.mark.parametrize("outcome", ["denied", "unknown"]) +def test_failing_or_transforming_failure_callbacks_never_change_the_denial_response( + hostile: _Rig, outcome: str +) -> None: + unknown: Final = "no-such-" + uuid.uuid4().hex + request_id: Final = "h-" + uuid.uuid4().hex + target: Final = hostile.denied.agent_id if outcome == "denied" else unknown + with hostile.proxy.scenario() as scenario: + caller: Final = _caller(hostile, scenario) + response: Final = _send(hostile, caller.key, f"/a2a/{target}", request_id=request_id) + expected: Final = (403, _refusal(target, response)) if outcome == "denied" else (404, _not_found(request_id, unknown)) + assert (response.status_code, response.content) == expected + + +@pytest.mark.parametrize("outcome", ["denied", "unknown"], ids=lambda outcome: "regression-" + outcome) +def test_failing_or_transforming_failure_callbacks_do_not_stop_the_denial_from_being_logged( + hostile: _Rig, outcome: str +) -> None: + unknown: Final = "no-such-" + uuid.uuid4().hex + target: Final = hostile.denied.agent_id if outcome == "denied" else unknown + with hostile.proxy.scenario() as scenario: + caller: Final = _caller(hostile, scenario) + response: Final = _send(hostile, caller.key, f"/a2a/{target}") + assert response.status_code == (403 if outcome == "denied" else 404), response.text + model: Final = f"a2a_agent/{hostile.denied.name}" if outcome == "denied" else f"a2a_agent/{unknown}" + _assert_one_failure(hostile, caller, status=response.status_code, agent_id=target, model=model) + + +def test_denials_follow_the_model_denied_chat_logging_under_error_log_flags(flagged: _Rig) -> None: + unknown: Final = "no-such-" + uuid.uuid4().hex + request_id: Final = "f-" + uuid.uuid4().hex + with flagged.proxy.scenario() as scenario: + chat_caller: Final = _caller(flagged, scenario) + denied_caller: Final = _caller(flagged, scenario) + unknown_caller: Final = _caller(flagged, scenario) + chat_alias: Final = "a2adeny-chat-" + uuid.uuid4().hex[:8] + chat_key: Final = scenario.key( + user_id=chat_caller.user_id, team_id=chat_caller.team_id, key_alias=chat_alias, models=["openai/gpt-5-mini"] + ) + chat: Final = flagged.proxy.request( + "POST", + "/v1/chat/completions", + {"model": "openai/gpt-5.4-mini", "messages": [{"role": "user", "content": "x"}]}, + key=chat_key, + ) + assert chat.status_code in {401, 403}, chat.text + denied: Final = _send(flagged, denied_caller.key, f"/a2a/{flagged.denied.agent_id}") + assert (denied.status_code, denied.content) == (403, _refusal(flagged.denied.agent_id, denied)) + missing: Final = _send(flagged, unknown_caller.key, f"/a2a/{unknown}", request_id=request_id) + assert (missing.status_code, missing.content) == (404, _not_found(request_id, unknown)) + _assert_never_dispatched(flagged, scenario, "never-sent") + chat_hooks: Final = held( + lambda: _hooks(flagged.events.of(chat_alias)), + lambda found: len(found) > 0, + holding=_SETTLE, + seconds=_CONVERGE, + ) + assert _hooks(flagged.events.of(denied_caller.alias)) == chat_hooks + assert _hooks(flagged.events.of(unknown_caller.alias)) == chat_hooks + control: Final = _caller(flagged, scenario) + granted: Final = _send( + flagged, control.key, f"/a2a/{flagged.allowed.agent_id}", method="message/send", version=None + ) + assert granted.status_code == 200, granted.text + eventually(lambda: _rows(flagged, control), lambda found: len(found) == 1, seconds=_CONVERGE) + assert (_rows(flagged, denied_caller), _rows(flagged, unknown_caller)) == ([], []) + + +def test_client_error_denials_never_raise_an_llm_exceptions_alert(alerting: tuple[_Rig, Wire]) -> None: + rig, alerts = alerting + unknown: Final = "no-such-" + uuid.uuid4().hex + marker: Final = "boom-" + uuid.uuid4().hex + + def failing(request: Request) -> Reply: + return Reply( + status=500, body=_compact({"error": {"message": f"upstream exploded {marker}", "type": "server_error"}}) + ) + + with rig.proxy.scenario() as scenario, wire_server(failing) as broken: + caller: Final = _caller(rig, scenario) + assert _send(rig, caller.key, f"/a2a/{rig.denied.agent_id}").status_code == 403 + assert _send(rig, caller.key, f"/a2a/{unknown}").status_code == 404 + model: Final = scenario.model(api_base=broken.url + "/v1", num_retries=0) + failed: Final = rig.proxy.request( + "POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": "x"}]} + ) + assert failed.status_code >= 500, failed.text + texts: Final = eventually( + lambda: tuple(JSON_OBJECT.validate_json(request.body).get("text") for request in alerts.drain()), + lambda found: any(marker in str(text) for text in found), + seconds=_CONVERGE, + ) + assert not any(rig.denied.agent_id in str(text) or unknown in str(text) for text in texts), texts + + +def test_guard_model_denied_chat_logs_one_acompletion_failure(rig: _Rig) -> None: + with rig.proxy.scenario() as scenario: + user: Final = scenario.user() + alias: Final = "a2adeny-chat-" + uuid.uuid4().hex[:8] + key: Final = scenario.key(user_id=user, key_alias=alias, models=["openai/gpt-5-mini"]) + caller: Final = _Caller(key, alias, user, None) + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + {"model": "openai/gpt-5.4-mini", "messages": [{"role": "user", "content": "x"}]}, + key=key, + ) + assert response.status_code == 403, response.text + assert "key_model_access_denied" in response.text + events: Final = held( + lambda: rig.events.of(alias), + lambda found: _hooks(found) == _ONE_FAILURE, + holding=_SETTLE, + seconds=_CONVERGE, + ) + assert {event.status_code for event in events if event.hook == "post_call_failure"} == {403} + rows: Final = held( + lambda: _rows(rig, caller), lambda found: len(found) == 1, holding=_SETTLE, seconds=_CONVERGE + ) + assert (rows[0]["status"], rows[0]["call_type"], rows[0]["spend"]) == ("failure", "acompletion", 0.0), rows + + +@pytest.mark.parametrize( + ("method", "call_type"), [("message/send", "asend_message"), ("message/stream", "asend_message_streaming")] +) +def test_guard_granted_agent_succeeds_with_one_success_row_and_no_failure( + rig: _Rig, method: str, call_type: str +) -> None: + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + request_id: Final = "g-" + uuid.uuid4().hex + response: Final = _send( + rig, caller.key, f"/a2a/{rig.allowed.agent_id}", method=method, version=None, request_id=request_id + ) + assert response.status_code == 200, response.text + if method == "message/stream": + assert response.text.startswith("data: ") and request_id in response.text, response.text + else: + assert object_value(JSON_OBJECT.validate_json(response.content)["result"])["parts"] == [ + {"kind": "text", "text": "granted"} + ] + held( + lambda: rig.events.of(caller.alias), + lambda found: _hooks(found) == ("async_success",), + holding=_SETTLE, + seconds=_CONVERGE, + ) + rows: Final = held( + lambda: _rows(rig, caller), lambda found: len(found) == 1, holding=_SETTLE, seconds=_CONVERGE + ) + assert (rows[0]["status"], rows[0]["call_type"], rows[0]["agent_id"]) == ( + "success", + call_type, + rig.allowed.agent_id, + ), rows + + +def test_guard_zero_budget_key_is_refused_once_without_double_logging(rig: _Rig) -> None: + with rig.proxy.scenario() as scenario: + user: Final = scenario.user() + alias: Final = "a2adeny-budget-" + uuid.uuid4().hex[:8] + key: Final = scenario.key( + user_id=user, key_alias=alias, max_budget=0, object_permission={"agents": [rig.allowed.agent_id]} + ) + caller: Final = _Caller(key, alias, user, None) + message: Final = "budget-" + uuid.uuid4().hex + response: Final = _send( + rig, key, f"/a2a/{rig.allowed.agent_id}", method="message/send", version=None, message_id=message + ) + assert response.status_code == 422, response.text + assert "Budget has been exceeded" in response.text + _assert_never_dispatched(rig, scenario, message) + held( + lambda: rig.events.of(alias), + lambda found: _hooks(found) == _ONE_FAILURE, + holding=_SETTLE, + seconds=_CONVERGE, + ) + rows: Final = held( + lambda: _rows(rig, caller), lambda found: len(found) == 1, holding=_SETTLE, seconds=_CONVERGE + ) + assert (rows[0]["status"], rows[0]["call_type"]) == ("failure", "asend_message"), rows + + +def test_guard_agent_card_discovery_keeps_its_refusals_and_writes_no_failure_row(rig: _Rig) -> None: + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + headers: Final = {"Authorization": f"Bearer {caller.key}"} + denied_card: Final = rig.proxy.client.get( + f"/a2a/{rig.denied.agent_id}/.well-known/agent-card.json", headers=headers + ) + assert (denied_card.status_code, denied_card.content) == (403, _refusal(rig.denied.agent_id)) + missing: Final = rig.proxy.client.get("/a2a/no-such-card/.well-known/agent-card.json", headers=headers) + assert (missing.status_code, missing.content) == (404, _compact({"detail": "Agent 'no-such-card' not found"})) + granted: Final = _send(rig, caller.key, f"/a2a/{rig.allowed.agent_id}", method="message/send", version=None) + assert granted.status_code == 200, granted.text + rows: Final = held( + lambda: _rows(rig, caller), lambda found: len(found) == 1, holding=_SETTLE, seconds=_CONVERGE + ) + assert [row["status"] for row in rows] == ["success"], rows + assert _hooks(rig.events.of(caller.alias)) == ("async_success",) + + +def test_guard_jsonrpc_1_0_envelope_keeps_its_invalid_request_body(rig: _Rig) -> None: + with rig.proxy.scenario() as scenario: + caller: Final = _caller(rig, scenario) + request_id: Final = "v-" + uuid.uuid4().hex + response: Final = _send( + rig, + caller.key, + f"/a2a/{rig.allowed.agent_id}", + method="message/send", + version=None, + request_id=request_id, + extra={"jsonrpc": "1.0"}, + ) + assert (response.status_code, response.content) == ( + 400, + _compact( + { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": -32600, "message": "Invalid Request: jsonrpc must be '2.0'"}, + } + ), + ) diff --git a/tests/integration/compatibility/test_a2a_wire_versions.py b/tests/integration/compatibility/test_a2a_wire_versions.py index 981700080e4..38b3d6b4648 100644 --- a/tests/integration/compatibility/test_a2a_wire_versions.py +++ b/tests/integration/compatibility/test_a2a_wire_versions.py @@ -452,8 +452,13 @@ def test_guard_no_agent_key_or_proxy_caller_key_is_sent(gateway: Gateway) -> Non (("0.3", "message/stream", _MESSAGE_V03), ("1.0", "SendStreamingMessage", _MESSAGE_V10)), ids=("v0.3", "v1.0"), ) +@pytest.mark.parametrize( + "path", + ("/a2a/{identity}", "/a2a/{identity}/message/stream", "/v1/a2a/{identity}/message/stream"), + ids=("bare", "message_stream", "v1_message_stream"), +) def test_a2a_stream_relays_each_upstream_event_as_its_own_sse_frame_in_the_pinned_dialect( - gateway: Gateway, version: str, method: str, message: dict[str, JsonValue] + gateway: Gateway, version: str, method: str, message: dict[str, JsonValue], path: str ) -> None: marker: Final = "a2astream" + uuid.uuid4().hex first_frame_seen: Final = threading.Event() @@ -482,7 +487,7 @@ def test_a2a_stream_relays_each_upstream_event_as_its_own_sse_frame_in_the_pinne ) with gateway.client.stream( "POST", - f"/a2a/{identity}", + path.format(identity=identity), headers={ "Authorization": f"Bearer {key}", "a2a-version": version, diff --git a/tests/integration/configuration/test_search_embedding_model_settings.py b/tests/integration/configuration/test_search_embedding_model_settings.py new file mode 100644 index 00000000000..9e7b2994a17 --- /dev/null +++ b/tests/integration/configuration/test_search_embedding_model_settings.py @@ -0,0 +1,299 @@ +from __future__ import annotations + +import json +import uuid +from collections.abc import Iterator, Sequence +from dataclasses import dataclass +from io import BytesIO +from pathlib import Path +from typing import Final +from zipfile import ZipFile + +import pytest +from integration._support.client import ( + JSON_OBJECT, + Gateway, + gateway_from_environment, + list_value, + object_value, + string_value, +) +from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue, TypeAdapter + +pytestmark: Final = pytest.mark.timeout(int(2 * graceful_stop_seconds() + 120)) + +_MODEL_NAME: Final = "search-embedding" +_AGENT_NOT_CONFIGURED: Final = ( + "agent search needs litellm_settings.agent_search_embedding_model set to an embedding model from model_list" +) +_SKILL_NOT_CONFIGURED: Final = ( + "skill search needs litellm_settings.skill_search_embedding_model set to an embedding model from model_list" +) +_SKILL_MARKDOWN: Final = """--- +name: echo-sample +description: Echo text for search validation +--- + +# Echo sample + +Return the supplied text unchanged +""" +_JSON_ARRAY: Final = TypeAdapter(list[JsonValue]) + + +@dataclass(frozen=True, slots=True) +class _SearchProxy: + owned: OwnedProxy + wire: Wire + skill_id: str + agent_id: str + + +def _embedding_reply(request: Request) -> Reply: + if request.method != "POST" or request.target != "/v1/embeddings": + return Reply(body=b'{"object":"list","data":[]}') + body: Final = JSON_OBJECT.validate_json(request.body) + input_value: Final = body.get("input") + input_count: Final = len(input_value) if isinstance(input_value, list) else 1 + embeddings: Final = [ + {"object": "embedding", "index": index, "embedding": [0.25, 0.75]} for index in range(input_count) + ] + return Reply( + body=json.dumps( + { + "object": "list", + "data": embeddings, + "model": "text-embedding-3-small", + "usage": {"prompt_tokens": input_count, "total_tokens": input_count}, + } + ).encode() + ) + + +def _write_config(directory: Path, upstream_url: str, agent_model: str, skill_model: str) -> Path: + config: Final = directory / f"search_embedding_{uuid.uuid4().hex}.yaml" + contents: Final[dict[str, JsonValue]] = { + "model_list": [ + { + "model_name": _MODEL_NAME, + "litellm_params": { + "model": "openai/text-embedding-3-small", + "api_base": f"{upstream_url}/v1", + "api_key": "synthetic-openai-key", + }, + } + ], + "litellm_settings": { + "agent_search_embedding_model": agent_model, + "skill_search_embedding_model": skill_model, + }, + "general_settings": { + "master_key": "os.environ/LITELLM_MASTER_KEY", + "database_url": "os.environ/DATABASE_URL", + "store_model_in_db": True, + }, + "agents": [ + { + "agent_name": "plain", + "agent_card_params": { + "protocolVersion": "1.0", + "name": "plain", + "description": "echo agent", + "url": "http://agent:9000/plain/", + "supportedInterfaces": [ + { + "url": "http://agent:9000/plain/", + "protocolBinding": "JSONRPC", + "protocolVersion": "1.0", + } + ], + "version": "1.0.0", + "capabilities": {"streaming": True}, + "defaultInputModes": ["text/plain"], + "defaultOutputModes": ["text/plain"], + "skills": [], + }, + } + ], + } + config.write_text(json.dumps(contents)) + return config + + +def _create_skill(candidate: OwnedProxy) -> str: + archive_data: Final = BytesIO() + with ZipFile(archive_data, "w") as archive: + archive.writestr("echo-sample/SKILL.md", _SKILL_MARKDOWN) + response: Final = candidate.gateway.request_multipart( + "/v1/skills?beta=true&custom_llm_provider=litellm_proxy", + { + "display_title": "Echo search fixture", + "description": "Echo sample skill for search validation", + "instructions": "Return the supplied text unchanged", + }, + {"files[]": ("echo-skill.zip", archive_data.getvalue(), "application/zip")}, + ) + assert response.status_code == 200, response.text + return string_value(JSON_OBJECT.validate_json(response.content)["id"]) + + +def _agent_id(candidate: OwnedProxy) -> str: + response: Final = candidate.gateway.request("GET", "/v1/agents") + assert response.status_code == 200, response.text + agents: Final = _JSON_ARRAY.validate_json(response.content) + assert len(agents) == 1 + return string_value(object_value(agents[0])["agent_id"]) + + +@pytest.fixture(scope="module") +def empty_search_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_SearchProxy]: + directory: Final = tmp_path_factory.mktemp("search-empty-model") + with wire_server(_embedding_reply) as upstream, gateway_from_environment() as gateway: + config: Final = _write_config( + directory, + upstream.url, + "os.environ/AGENT_SEARCH_EMBEDDING_MODEL", + "os.environ/SKILL_SEARCH_EMBEDDING_MODEL", + ) + with owned_proxy_process( + gateway, + directory, + {"AGENT_SEARCH_EMBEDDING_MODEL": "", "SKILL_SEARCH_EMBEDDING_MODEL": ""}, + config=config, + ) as owned: + yield _SearchProxy(owned, upstream, _create_skill(owned), _agent_id(owned)) + + +@pytest.fixture(scope="module") +def configured_search_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_SearchProxy]: + directory: Final = tmp_path_factory.mktemp("search-configured-model") + with wire_server(_embedding_reply) as upstream, gateway_from_environment() as gateway: + config: Final = _write_config(directory, upstream.url, _MODEL_NAME, _MODEL_NAME) + with owned_proxy_process( + gateway, + directory, + {"AGENT_SEARCH_EMBEDDING_MODEL": _MODEL_NAME, "SKILL_SEARCH_EMBEDDING_MODEL": _MODEL_NAME}, + config=config, + ) as owned: + yield _SearchProxy(owned, upstream, _create_skill(owned), _agent_id(owned)) + + +def _assert_not_configured(response_body: JsonValue, expected_error: str, reason: str) -> None: + detail: Final = object_value(object_value(response_body)["detail"]) + assert detail == {"error": expected_error, "message": reason} + + +def _assert_no_embedding_requests(candidate: _SearchProxy) -> None: + embedding_requests: Final = tuple( + request for request in candidate.wire.drain() if request.method == "POST" and request.target == "/v1/embeddings" + ) + assert embedding_requests == () + + +def _assert_mcp_search_not_configured(gateway: Gateway, key: str, tool_name: str, reason: str) -> None: + response: Final = gateway.request( + "POST", + "/mcp-rest/tools/call", + {"name": tool_name, "arguments": {"query": "echo"}}, + key=key, + ) + assert response.status_code == 200, response.text + result: Final = JSON_OBJECT.validate_json(response.content) + assert result.get("isError") is True, response.text + content: Final = object_value(list_value(result["content"])[0]) + text: Final = string_value(content["text"]) + assert reason in text + assert "You passed in model=" not in text + + +def test_empty_agent_embedding_setting_returns_not_configured(empty_search_proxy: _SearchProxy) -> None: + response: Final = empty_search_proxy.owned.gateway.request("GET", "/v1/agents?query=echo") + assert response.status_code == 400, response.text + _assert_not_configured( + JSON_OBJECT.validate_json(response.content), + "agent_search_not_configured", + _AGENT_NOT_CONFIGURED, + ) + _assert_no_embedding_requests(empty_search_proxy) + + +def test_empty_skill_embedding_setting_returns_not_configured(empty_search_proxy: _SearchProxy) -> None: + response: Final = empty_search_proxy.owned.gateway.request( + "GET", "/v1/skills?custom_llm_provider=litellm_proxy&query=echo" + ) + assert response.status_code == 400, response.text + _assert_not_configured( + JSON_OBJECT.validate_json(response.content), + "skill_search_not_configured", + _SKILL_NOT_CONFIGURED, + ) + _assert_no_embedding_requests(empty_search_proxy) + + +def test_empty_agent_embedding_setting_returns_not_configured_through_mcp(empty_search_proxy: _SearchProxy) -> None: + with empty_search_proxy.owned.gateway.scenario() as scenario: + key: Final = scenario.key( + object_permission={ + "agents": [empty_search_proxy.agent_id], + "mcp_tool_search_enabled": True, + }, + ) + _assert_mcp_search_not_configured( + empty_search_proxy.owned.gateway, + key, + "agent_search", + _AGENT_NOT_CONFIGURED, + ) + _assert_no_embedding_requests(empty_search_proxy) + + +def test_empty_skill_embedding_setting_returns_not_configured_through_mcp(empty_search_proxy: _SearchProxy) -> None: + with empty_search_proxy.owned.gateway.scenario() as scenario: + user_id: Final = scenario.user(user_role="proxy_admin") + key: Final = scenario.key(user_id=user_id, object_permission={"mcp_tool_search_enabled": True}) + _assert_mcp_search_not_configured( + empty_search_proxy.owned.gateway, + key, + "skill_search", + _SKILL_NOT_CONFIGURED, + ) + _assert_no_embedding_requests(empty_search_proxy) + + +def test_configured_agent_embedding_model_returns_search_score(configured_search_proxy: _SearchProxy) -> None: + configured_search_proxy.wire.drain() + response: Final = configured_search_proxy.owned.gateway.request("GET", "/v1/agents?query=echo") + assert response.status_code == 200, response.text + agents: Final = _JSON_ARRAY.validate_json(response.content) + assert len(agents) == 1 + agent: Final = object_value(agents[0]) + assert string_value(agent["agent_name"]) == "plain" + assert isinstance(agent["search_score"], (int, float)) + embedding_requests: Final = tuple( + request + for request in configured_search_proxy.wire.drain() + if request.method == "POST" and request.target == "/v1/embeddings" + ) + assert embedding_requests + + +def test_configured_skill_embedding_model_returns_search_score(configured_search_proxy: _SearchProxy) -> None: + configured_search_proxy.wire.drain() + response: Final = configured_search_proxy.owned.gateway.request( + "GET", "/v1/skills?custom_llm_provider=litellm_proxy&query=echo" + ) + assert response.status_code == 200, response.text + skills: Final[Sequence[JsonValue]] = list_value(JSON_OBJECT.validate_json(response.content)["data"]) + matching_skills: Final = tuple( + object_value(skill) for skill in skills if object_value(skill)["id"] == configured_search_proxy.skill_id + ) + assert len(matching_skills) == 1 + assert isinstance(matching_skills[0]["search_score"], (int, float)) + embedding_requests: Final = tuple( + request + for request in configured_search_proxy.wire.drain() + if request.method == "POST" and request.target == "/v1/embeddings" + ) + assert embedding_requests diff --git a/tests/integration/translation/chat_completions/streaming/test_chat_completions_streaming_anthropic.py b/tests/integration/translation/chat_completions/streaming/test_chat_completions_streaming_anthropic.py index b2c0d39f9aa..fca8de69f00 100644 --- a/tests/integration/translation/chat_completions/streaming/test_chat_completions_streaming_anthropic.py +++ b/tests/integration/translation/chat_completions/streaming/test_chat_completions_streaming_anthropic.py @@ -152,15 +152,14 @@ def _place_delta(partial_json: str) -> SSEFrame: ) -def _place_chunk(delta: dict[str, JsonValue], *, finish_reason: str | None = None) -> SSEFrame: - finish: Final = {} if finish_reason is None else {"finish_reason": finish_reason} +def _place_chunk(delta: dict[str, JsonValue]) -> SSEFrame: return SSEFrame( data={ "id": ANY, "object": "chat.completion.chunk", "created": ANY, "model": "anthropic/claude-haiku-5-5", - "choices": [{"index": 0, "delta": delta, **finish}], + "choices": [{"index": 0, "delta": delta}], } ) @@ -245,7 +244,19 @@ CLAUDE_HAIKU_5_5_JSON_SCHEMA_CUT_STREAM_TEST_CASE: Final = StreamingTranslationT _place_chunk({"content": 'try": "'}), _place_chunk({"content": 'France"'}), _place_chunk({"content": "}"}), - _place_chunk({}, finish_reason="stop"), + SSEFrame( + data={ + "error": { + "message": ( + "litellm.APIConnectionError: AnthropicException - " + "Stream ended before a finish_reason was received" + ), + "type": "internal_server_error", + "param": None, + "code": "500", + } + } + ), SSEFrame(data="[DONE]"), ), expected_litellm_response={ @@ -257,7 +268,6 @@ CLAUDE_HAIKU_5_5_JSON_SCHEMA_CUT_STREAM_TEST_CASE: Final = StreamingTranslationT { "index": 0, "message": {"role": "assistant", "content": '{"city": "Paris", "country": "France"}'}, - "finish_reason": "stop", } ], }, diff --git a/tests/integration/translation/runner.py b/tests/integration/translation/runner.py index a5470df7e2d..65fd8dbbb8e 100644 --- a/tests/integration/translation/runner.py +++ b/tests/integration/translation/runner.py @@ -129,8 +129,9 @@ def _join_stream_frames(endpoint: str, frames: tuple[SSEFrame, ...]) -> Mapping[ case "/v1/chat/completions": state: Final = ChatCompletionStreamState() for frame in frames: - if frame.data != "[DONE]": - state.handle_chunk(ChatCompletionChunk.model_validate(frame.data)) + if frame.data == "[DONE]" or (isinstance(frame.data, Mapping) and "error" in frame.data): + continue + state.handle_chunk(ChatCompletionChunk.model_validate(frame.data)) return JSON_OBJECT.validate_python(state.get_final_completion().model_dump(mode="json", exclude_none=True)) case _: raise AssertionError(f"no stream accumulator for {endpoint}") diff --git a/tests/unit/caching/test_request_redis_batch_pre_call.py b/tests/unit/caching/test_request_redis_batch_pre_call.py index 37169e13a0b..94f19dbdc7c 100644 --- a/tests/unit/caching/test_request_redis_batch_pre_call.py +++ b/tests/unit/caching/test_request_redis_batch_pre_call.py @@ -290,7 +290,11 @@ def _router(redis_cache: FakeRedisCache, routing_strategy: str = "usage-based-ro @pytest.mark.asyncio -async def test_armed_routing_read_rides_the_admission_pipeline_and_routing_issues_no_read_of_its_own(): +async def test_armed_routing_read_rides_the_admission_pipeline_and_routing_issues_no_read_of_its_own(monkeypatch): + monkeypatch.setattr( + "litellm.router_strategy.lowest_tpm_rpm_v2.get_utc_datetime", + lambda: datetime(2026, 1, 1, 12, 0, 30, tzinfo=timezone.utc), + ) client = FakeClient(_lua_ok_replies) redis_cache = FakeRedisCache(client) router = _router(redis_cache) diff --git a/tests/unit/litellm_core_utils/test_api_route_to_call_types.py b/tests/unit/litellm_core_utils/test_api_route_to_call_types.py index 42ca91bfd8f..c355ef65344 100644 --- a/tests/unit/litellm_core_utils/test_api_route_to_call_types.py +++ b/tests/unit/litellm_core_utils/test_api_route_to_call_types.py @@ -8,6 +8,8 @@ Regression coverage for the guardrail route table bugs: take call_types[0] to a handler-less type """ +from typing import Final + from litellm.litellm_core_utils.api_route_to_call_types import ( get_call_types_for_route, ) @@ -103,6 +105,16 @@ class TestExistingRouteResolutionUnchanged: assert call_types is not None assert list(call_types) == [CallTypes.asend_message, CallTypes.send_message] + def test_single_segment_placeholder_message_stream_resolves(self): + call_types: Final = get_call_types_for_route("/a2a/my-agent/message/stream") + assert call_types is not None + assert list(call_types) == [CallTypes.asend_message, CallTypes.send_message] + + def test_v1_single_segment_placeholder_message_stream_resolves(self): + call_types: Final = get_call_types_for_route("/v1/a2a/my-agent/message/stream") + assert call_types is not None + assert list(call_types) == [CallTypes.asend_message, CallTypes.send_message] + def test_longer_route_does_not_collapse_into_bare_placeholder_pattern(self): call_types = get_call_types_for_route("/responses/resp_123/input_items") assert call_types is not None diff --git a/tests/unit/litellm_core_utils/test_json_fragment_accumulator.py b/tests/unit/litellm_core_utils/test_json_fragment_accumulator.py index f27a5fd2307..212dff79daf 100644 --- a/tests/unit/litellm_core_utils/test_json_fragment_accumulator.py +++ b/tests/unit/litellm_core_utils/test_json_fragment_accumulator.py @@ -1,6 +1,5 @@ import json import time -import tracemalloc from collections.abc import Iterator from typing import Final from unittest.mock import patch @@ -193,34 +192,15 @@ def _streamed(fragments: tuple[str, ...]) -> JSONFragmentAccumulator: return accumulator -def _drain_peak_bytes(fragments: tuple[str, ...]) -> int: - accumulator: Final = _streamed(fragments) - first_found, _ = accumulator.pop_next_value() - assert first_found is True - already_tracing: Final = tracemalloc.is_tracing() - if not already_tracing: - tracemalloc.start() - tracemalloc.reset_peak() - baseline, _ = tracemalloc.get_traced_memory() - try: - drained: Final = sum(1 for _ in _popped_values(accumulator)) - _, peak = tracemalloc.get_traced_memory() - finally: - if not already_tracing: - tracemalloc.stop() - assert drained == len(fragments) - 1 - return peak - baseline - - -def test_draining_many_concatenated_values_never_copies_the_remainder(): +def test_draining_many_concatenated_values_yields_every_value_in_order(): value: Final = '{"text": "' + "x" * 20 + '"}' fragments: Final = tuple(value if index % 2 else f" {value}" for index in range(40_000)) - payload_bytes: Final = sum(len(fragment) for fragment in fragments) - peak_bytes: Final = _drain_peak_bytes(fragments) - assert peak_bytes < payload_bytes // 4, ( - f"draining {len(fragments)} values allocated a peak of {peak_bytes} bytes over a {payload_bytes} byte payload, " - "copying the remainder on every pop?" - ) + accumulator: Final = _streamed(fragments) + + drained: Final = tuple(_popped_values(accumulator)) + + assert drained == (json.loads(value),) * len(fragments) + assert accumulator.pop_next_value() == (False, None) def test_could_close_json_after_many_blank_fragments_is_not_quadratic(): diff --git a/tests/unit/llms/litellm_proxy/skills/test_skill_search.py b/tests/unit/llms/litellm_proxy/skills/test_skill_search.py index 0647b203aff..e3bf6d904a6 100644 --- a/tests/unit/llms/litellm_proxy/skills/test_skill_search.py +++ b/tests/unit/llms/litellm_proxy/skills/test_skill_search.py @@ -281,6 +281,36 @@ class TestSearchSkills: assert isinstance(outcome, SkillSearchNotConfigured) assert "skill_search_embedding_model" in outcome.reason + @pytest.mark.parametrize("embedding_model", ("", " "), ids=("empty", "whitespace")) + @pytest.mark.asyncio + async def test_empty_embedding_model_is_not_configured(self, embedding_model: str) -> None: + router: Final = _embedding_router() + not_configured: Final = await search_skills( + "language translation", + SKILLS, + 5, + router=router, + embedding_model=None, + index=SkillSearchIndex(), + user_api_key_dict=CALLER, + proxy_logging_obj=_pass_through_key_limits(), + ) + outcome: Final = await search_skills( + "language translation", + SKILLS, + 5, + router=router, + embedding_model=embedding_model, + index=SkillSearchIndex(), + user_api_key_dict=CALLER, + proxy_logging_obj=_pass_through_key_limits(), + ) + + assert isinstance(not_configured, SkillSearchNotConfigured) + assert isinstance(outcome, SkillSearchNotConfigured) + assert outcome.reason == not_configured.reason + router.aembedding.assert_not_awaited() + @pytest.mark.asyncio async def test_no_router_is_not_configured(self) -> None: outcome = await search_skills( diff --git a/tests/unit/proxy/agent_endpoints/test_a2a_endpoints.py b/tests/unit/proxy/agent_endpoints/test_a2a_endpoints.py index 016a5151e16..88da07c612b 100644 --- a/tests/unit/proxy/agent_endpoints/test_a2a_endpoints.py +++ b/tests/unit/proxy/agent_endpoints/test_a2a_endpoints.py @@ -12,19 +12,22 @@ import sys from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from contextlib import AbstractContextManager, ExitStack from dataclasses import dataclass +from datetime import datetime +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 fastapi.testclient import TestClient from pydantic import BaseModel, ConfigDict import litellm from litellm.integrations.custom_logger import CustomLogger -from litellm.proxy._types import UserAPIKeyAuth -from litellm.types.agents import AgentCaller +from litellm.proxy._types import LiteLLM_ObjectPermissionTable, UserAPIKeyAuth +from litellm.types.agents import AgentCaller, AgentResponse AddLiteLLMData = Callable[..., Awaitable[dict[str, object]]] @@ -2782,7 +2785,9 @@ async def test_forward_jsonrpc_sse_is_untouched_while_keepalives_are_unconfigure @pytest.mark.asyncio -async def test_forward_jsonrpc_sse_keeps_unicode_line_separators_inside_an_event(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_forward_jsonrpc_sse_keeps_unicode_line_separators_inside_an_event( + monkeypatch: pytest.MonkeyPatch, +) -> None: monkeypatch.setattr(litellm, "sse_keepalive_ping_interval_seconds", None) separated_text: Final = "before\u2028between\u2029after" event: Final = { @@ -2944,6 +2949,202 @@ async def test_failed_upstream_stream_relays_its_body_only_when_that_is_a_json_o } +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: dict[str, object], + response_obj: object, + start_time: datetime | None, + end_time: datetime | None, + ) -> None: + payload: Final = kwargs["standard_logging_object"] + assert isinstance(payload, Mapping) + self.failure_payloads.append(payload) + + async def async_post_call_failure_hook( + self, + request_data: dict[str, object], + original_exception: Exception, + user_api_key_dict: UserAPIKeyAuth, + traceback_str: str | None = 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]: + from litellm.proxy import proxy_server + from litellm.proxy.agent_endpoints import agent_registry + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + path_agent_id: Final = "denied-agent" if registered else "no-such-agent" + registry: Final = agent_registry.AgentRegistry() + if registered: + registry.register_agent( + AgentResponse( + agent_id="registered-denied-id", + agent_name=path_agent_id, + agent_card_params={"name": "Test Agent", "url": "https://agent.example.com"}, + ) + ) + monkeypatch.setattr(agent_registry, "global_agent_registry", registry) + monkeypatch.setattr(proxy_server, "prisma_client", None) + 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", + object_permission=LiteLLM_ObjectPermissionTable(object_permission_id="caller-grants", agents=["other-agent"]), + request_route=f"/a2a/{path_agent_id}", + ) + with respx.mock as 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: + assert not upstream.calls + return e.status_code, e.detail + assert not upstream.calls + return response.status_code, json.loads(response.body.decode()) + + +_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 = await _invoke_rejected( + monkeypatch, registered=registered, proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()) + ) + + assert (got_status, got_body) == (status, body) + 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, + caplog: pytest.LogCaptureFixture, + 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 = await _invoke_rejected( + monkeypatch, registered=registered, proxy_logging_obj=proxy_logging_obj + ) + + assert (got_status, got_body) == (status, body) + proxy_logging_obj.post_call_failure_hook.assert_awaited_once() + assert any(record.getMessage() == "Failed to log rejected A2A request" for record in caplog.records) + + +@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) + + def _a2a_app_refusing_with(refusal: Exception) -> TestClient: from fastapi import FastAPI diff --git a/tests/unit/proxy/agent_endpoints/test_agent_search.py b/tests/unit/proxy/agent_endpoints/test_agent_search.py index 750fadadab0..ab72a44b7ce 100644 --- a/tests/unit/proxy/agent_endpoints/test_agent_search.py +++ b/tests/unit/proxy/agent_endpoints/test_agent_search.py @@ -261,6 +261,42 @@ class TestSearchAgents: assert isinstance(outcome, AgentSearchNotConfigured) assert "agent_search_embedding_model" in outcome.reason + @pytest.mark.parametrize("embedding_model", ("", " "), ids=("empty", "whitespace")) + @pytest.mark.asyncio + async def test_empty_embedding_model_is_not_configured(self, embedding_model: str) -> None: + router: Final = MagicMock() + router.aembedding = AsyncMock( + side_effect=lambda model, input, metadata: litellm.EmbeddingResponse( + model=model, + data=[{"object": "embedding", "index": i, "embedding": list(VECTORS[t])} for i, t in enumerate(input)], + ) + ) + not_configured: Final = await search_agents( + "language translation", + AGENTS, + 5, + router=router, + embedding_model=None, + index=AgentSearchIndex(), + user_api_key_dict=CALLER, + proxy_logging_obj=_pass_through_key_limits(), + ) + outcome: Final = await search_agents( + "language translation", + AGENTS, + 5, + router=router, + embedding_model=embedding_model, + index=AgentSearchIndex(), + user_api_key_dict=CALLER, + proxy_logging_obj=_pass_through_key_limits(), + ) + + assert isinstance(not_configured, AgentSearchNotConfigured) + assert isinstance(outcome, AgentSearchNotConfigured) + assert outcome.reason == not_configured.reason + router.aembedding.assert_not_awaited() + @pytest.mark.asyncio async def test_no_router_is_not_configured(self) -> None: outcome = await search_agents( diff --git a/tests/unit/proxy/auth/test_route_checks.py b/tests/unit/proxy/auth/test_route_checks.py index 529f639999d..5df3fbab9a6 100644 --- a/tests/unit/proxy/auth/test_route_checks.py +++ b/tests/unit/proxy/auth/test_route_checks.py @@ -3925,6 +3925,7 @@ AGENT_INFERENCE_ROUTES = [ "/a2a/abc-123", "/a2a/abc-123/message/send", "/a2a/abc-123/message/stream", + "/v1/a2a/abc-123/message/stream", "/a2a/abc-123/.well-known/agent-card.json", ] diff --git a/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py b/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py index 8a3e942e148..84f4d7a4173 100644 --- a/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py +++ b/tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py @@ -2025,6 +2025,28 @@ class TestStreamingHttpErrorFrames: def _use_real_mappings(self, monkeypatch): _patch_translation_mappings(monkeypatch, load_guardrail_translation_mappings()) + @pytest.mark.asyncio + async def test_v1_a2a_stream_route_scans_response_chunks(self): + guardrail: Final = RecordingGuardrail() + guardrail.streaming_sampling_rate = 1 + chunk: Final = { + "jsonrpc": "2.0", + "id": "stream-1", + "result": { + "kind": "task", + "artifacts": [{"parts": [{"kind": "text", "text": "streamed text"}]}], + }, + } + + await _drive_stream( + UnifiedLLMGuardrails(), + guardrail, + [chunk], + request_route="/v1/a2a/my-agent/message/stream", + ) + + assert guardrail.apply_calls[0]["inputs"]["texts"] == ["streamed text"] + @pytest.mark.asyncio async def test_chat_eos_block_emits_data_error_frame(self): import json as _json diff --git a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py index b21a5e6c337..5e31906a4b9 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py +++ b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py @@ -286,6 +286,20 @@ async def test_deepgram_listen_refuses_sessions_it_cannot_price(query, missing_k assert "dg-provider-key" not in websocket.closed[1] +@pytest.mark.asyncio +async def test_deepgram_listen_refusal_for_a_long_unpriced_model_fits_a_close_frame( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + monkeypatch.setenv("DEEPGRAM_API_KEY", "dg-provider-key") + websocket: Final = _FakeWebSocket("/deepgram/v1/listen", "model=" + "%C3%A9" * 100) + + relay: Final = await _serve(websocket) + + assert relay.calls == [] + assert websocket.closed == (1008, "No streaming price for 'deepgram/streaming/" + "é" * 40) + + @pytest.mark.asyncio async def test_deepgram_listen_relays_once_the_operator_prices_the_model(monkeypatch): monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) diff --git a/tests/unit/proxy/test__lazy_features.py b/tests/unit/proxy/test__lazy_features.py index 0749fc12ea9..a4bd47ccb74 100644 --- a/tests/unit/proxy/test__lazy_features.py +++ b/tests/unit/proxy/test__lazy_features.py @@ -10,6 +10,7 @@ from fastapi.testclient import TestClient from pydantic import BaseModel from litellm.proxy._lazy_features import ( + LAZY_FEATURES, LazyFeature, LazyFeatureMiddleware, attach_lazy_features, @@ -66,6 +67,11 @@ def _has_lazy_middleware(app: FastAPI) -> bool: return any(middleware.cls is LazyFeatureMiddleware for middleware in app.user_middleware) +def test_a2a_feature_matches_v1_message_stream() -> None: + feature: Final = next(feature for feature in LAZY_FEATURES if feature.name == "a2a") + assert feature.matches("/v1/a2a/x/message/stream") + + @pytest.mark.parametrize("disable_lazy_routes", ("false", "true")) def test_retired_roi_calculator_cannot_be_loaded_or_called( monkeypatch: pytest.MonkeyPatch, disable_lazy_routes: str diff --git a/tests/unit/router_strategy/complexity_router/test_jev_classifier.py b/tests/unit/router_strategy/complexity_router/test_jev_classifier.py index abbb1ed80d0..bcbf69e14a6 100644 --- a/tests/unit/router_strategy/complexity_router/test_jev_classifier.py +++ b/tests/unit/router_strategy/complexity_router/test_jev_classifier.py @@ -1316,3 +1316,24 @@ async def test_jev_open_breaker_preserves_input_budget_in_fallback_decision(fall transport.handle_async_request.assert_awaited_once() finally: await handler.client.aclose() + + +def test_jev_input_targets_still_takes_a_router_with_a_deprecation_warning() -> None: + from litellm.router_strategy.complexity_router.jev_classifier import jev_input_targets + + router = litellm.Router( + model_list=[ + { + "model_name": "decision-deployment", + "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-test"}, + "model_info": {"id": "d1", "max_input_tokens": 1234}, + } + ] + ) + config = JevClassifierConfig.model_validate({"deployment_name": "decision-deployment"}) + + with pytest.warns(DeprecationWarning, match="pass context"): + old_way = jev_input_targets(router, config) + + assert old_way == jev_input_targets(router._core_ctx, config) + assert [target.max_tokens for target in old_way] == [1234] diff --git a/tests/unit/router_strategy/test_litellm_encoder.py b/tests/unit/router_strategy/test_litellm_encoder.py index d950025afdb..f12aa7cdb6a 100644 --- a/tests/unit/router_strategy/test_litellm_encoder.py +++ b/tests/unit/router_strategy/test_litellm_encoder.py @@ -134,3 +134,19 @@ class TestEmbeddingInputCap: _encoder(router, max_input_chars=20).encode_queries(["route me please, then a huge pasted file"]) assert router.embedded_inputs == [["route me please, the"]] + + +@pytest.mark.parametrize( + "embeds", + [ + {"data": [{"embedding": [0.1]}]}, + [{"embedding": [0.1]}], + type("HasData", (), {"data": [{"embedding": [0.1]}]})(), + ], + ids=["dict", "list", "object-with-data"], +) +def test_litellm_to_list_rejects_anything_but_an_embedding_response(embeds: object) -> None: + from litellm.router_strategy.auto_router.litellm_encoder import litellm_to_list + + with pytest.raises(ValueError, match="No embeddings found"): + litellm_to_list(embeds) diff --git a/tests/unit/router_strategy/test_router_tag_routing.py b/tests/unit/router_strategy/test_router_tag_routing.py index b4e29df8116..0c19cfc72f6 100644 --- a/tests/unit/router_strategy/test_router_tag_routing.py +++ b/tests/unit/router_strategy/test_router_tag_routing.py @@ -1225,6 +1225,32 @@ def test_chain_tag_filtering_override_falls_back_to_healthy_deployments_on_looku assert _chain_tag_filtering_override(_BrokenCtx(), "gpt-4", healthy_deployments) is False +@pytest.mark.asyncio() +async def test_get_deployments_for_tag_without_a_model_keeps_the_router_wide_setting(): + # model=None names no group, so no group-level enable_tag_filtering applies: the router-wide + # setting (off here) decides, even when a healthy deployment opts in on its own + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": {"model": "gpt-4o", "api_key": "sk-test", "tags": ["teamA"]}, + "model_info": {"id": "a", "enable_tag_filtering": True}, + } + ] + ) + healthy_deployments = router.registry().deployments("gpt-4", source="all") + + picked = await get_deployments_for_tag( + router._core_ctx, + None, # pyright: ignore[reportArgumentType] # a direct caller passing no model + healthy_deployments, + request_kwargs={"metadata": {"tags": ["zzz"]}}, + request_deployments=healthy_deployments, + ) + + assert [d["model_info"]["id"] for d in picked] == ["a"] + + @pytest.mark.asyncio() async def test_chain_enable_tag_filtering_true_overrides_router_level_false(): # Router-wide tag filtering is off; this model group opts in on its own via diff --git a/tests/unit/router_utils/test_handle_error.py b/tests/unit/router_utils/test_handle_error.py index 151fb0bfc68..22bd6e74190 100644 --- a/tests/unit/router_utils/test_handle_error.py +++ b/tests/unit/router_utils/test_handle_error.py @@ -273,3 +273,10 @@ def setup_and_teardown(): yield loop.close() asyncio.set_event_loop(None) + + +@pytest.mark.asyncio +async def test_send_llm_exception_alert_without_a_router_does_nothing(): + from litellm.router_utils.handle_error import send_llm_exception_alert + + assert await send_llm_exception_alert(None, {"model": "gpt-5-mini"}, "traceback", Exception("boom")) is None diff --git a/tests/unit/test_router/core/test_endpoints.py b/tests/unit/test_router/core/test_endpoints.py index b2ad602c4c6..c0c4c8cf6dd 100644 --- a/tests/unit/test_router/core/test_endpoints.py +++ b/tests/unit/test_router/core/test_endpoints.py @@ -414,6 +414,8 @@ def _scripted_stream(*texts: str) -> Callable[..., CustomStreamWrapper]: async def chunks() -> AsyncIterator[ModelResponseStream]: for text in texts: yield ModelResponseStream(choices=[{"index": 0, "delta": {"role": "assistant", "content": text}}]) + # A stream that ends without a finish reason fails since #47665. + yield ModelResponseStream(choices=[{"index": 0, "delta": {}, "finish_reason": "stop"}]) yield ModelResponseStream(choices=[{"index": 0, "delta": {}, "finish_reason": "stop"}]) logging_obj: Final = LiteLLMLogging( diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 0dce821f862..a736f5a2519 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -452,6 +452,30 @@ export interface paths { patch?: never; trace?: never; }; + "/a2a/{agent_id}/message/stream": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Invoke Agent A2A + * @description Invoke an agent using the A2A protocol (JSON-RPC 2.0). + * + * Supported methods: + * - message/send: Send a message and get a response + * - message/stream: Send a message and stream the response + */ + post: operations["invoke_agent_a2a_a2a__agent_id__message_stream_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/access_group/list": { parameters: { query?: never; @@ -19307,6 +19331,30 @@ export interface paths { patch?: never; trace?: never; }; + "/v1/a2a/{agent_id}/message/stream": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Invoke Agent A2A + * @description Invoke an agent using the A2A protocol (JSON-RPC 2.0). + * + * Supported methods: + * - message/send: Send a message and get a response + * - message/stream: Send a message and stream the response + */ + post: operations["invoke_agent_a2a_v1_a2a__agent_id__message_stream_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/v1/access_group": { parameters: { query?: never; @@ -50790,6 +50838,24 @@ export interface components { /** Status */ status?: ("pending" | "running" | "paused" | "completed" | "failed") | null; }; + /** _A2AJsonRpcResponse */ + _A2AJsonRpcResponse: { + /** Error */ + error?: { + [key: string]: unknown; + } | null; + /** Id */ + id: string | number | null; + /** + * Jsonrpc + * @constant + */ + jsonrpc: "2.0"; + /** Result */ + result?: unknown | null; + } & { + [key: string]: unknown; + }; /** ModelInfo */ litellm__proxy___types__ModelInfo: { /** Base Model */ @@ -52149,6 +52215,77 @@ export interface operations { }; }; }; + invoke_agent_a2a_a2a__agent_id__message_stream_post: { + parameters: { + query?: never; + header?: never; + path: { + agent_id: string; + }; + cookie?: never; + }; + requestBody: { + content: { + /** + * @example { + * "id": "stream-1", + * "jsonrpc": "2.0", + * "method": "message/stream", + * "params": { + * "message": { + * "kind": "message", + * "messageId": "message-1", + * "parts": [ + * { + * "kind": "text", + * "text": "Hello" + * } + * ], + * "role": "user" + * } + * } + * } + */ + "application/json": { + /** Id */ + id?: string | number | null; + /** + * Jsonrpc + * @constant + */ + jsonrpc: "2.0"; + /** Method */ + method: string; + /** Params */ + params: { + [key: string]: unknown; + }; + } & { + [key: string]: unknown; + }; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["_A2AJsonRpcResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; list_access_groups_access_group_list_get: { parameters: { query?: never; @@ -77782,6 +77919,77 @@ export interface operations { }; }; }; + invoke_agent_a2a_v1_a2a__agent_id__message_stream_post: { + parameters: { + query?: never; + header?: never; + path: { + agent_id: string; + }; + cookie?: never; + }; + requestBody: { + content: { + /** + * @example { + * "id": "stream-1", + * "jsonrpc": "2.0", + * "method": "message/stream", + * "params": { + * "message": { + * "kind": "message", + * "messageId": "message-1", + * "parts": [ + * { + * "kind": "text", + * "text": "Hello" + * } + * ], + * "role": "user" + * } + * } + * } + */ + "application/json": { + /** Id */ + id?: string | number | null; + /** + * Jsonrpc + * @constant + */ + jsonrpc: "2.0"; + /** Method */ + method: string; + /** Params */ + params: { + [key: string]: unknown; + }; + } & { + [key: string]: unknown; + }; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["_A2AJsonRpcResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; list_access_groups_v1_access_group_get: { parameters: { query?: never;