diff --git a/litellm/a2a_protocol/main.py b/litellm/a2a_protocol/main.py index 3ad5485dea..6979e1ac65 100644 --- a/litellm/a2a_protocol/main.py +++ b/litellm/a2a_protocol/main.py @@ -159,7 +159,9 @@ async def _send_message_via_completion_bridge( api_base=api_base, ) - return LiteLLMSendMessageResponse.from_dict(response_dict) + return LiteLLMSendMessageResponse.from_dict( + response_dict, request_id=str(request.id) + ) async def _execute_a2a_send_with_retry( @@ -317,15 +319,6 @@ async def asend_message( ) card_url = getattr(agent_card, "url", None) if agent_card else None - context_id = trace_id or str(uuid.uuid4()) - message = request.params.message - if isinstance(message, dict): - if message.get("context_id") is None: - message["context_id"] = context_id - else: - if getattr(message, "context_id", None) is None: - message.context_id = context_id - a2a_response = await _execute_a2a_send_with_retry( a2a_client=a2a_client, request=request, @@ -338,7 +331,9 @@ async def asend_message( verbose_logger.info(f"A2A send_message completed, request_id={request.id}") # Wrap in LiteLLM response type for _hidden_params support - response = LiteLLMSendMessageResponse.from_a2a_response(a2a_response) + response = LiteLLMSendMessageResponse.from_a2a_response( + a2a_response, request_id=str(request.id) + ) # Calculate token usage from request and response response_dict = a2a_response.model_dump(mode="json", exclude_none=True) diff --git a/litellm/proxy/agent_endpoints/a2a_endpoints.py b/litellm/proxy/agent_endpoints/a2a_endpoints.py index 7b56155982..9f1403d432 100644 --- a/litellm/proxy/agent_endpoints/a2a_endpoints.py +++ b/litellm/proxy/agent_endpoints/a2a_endpoints.py @@ -6,12 +6,14 @@ The A2A SDK can point to LiteLLM's URL and invoke agents registered with LiteLLM """ import json -from typing import Any, Dict, List, Optional +from typing import Any, AsyncGenerator, Dict, List, Optional +from urllib.parse import urlparse from fastapi import APIRouter, Depends, HTTPException, Request, Response from fastapi.responses import JSONResponse, StreamingResponse from litellm._logging import verbose_proxy_logger +from litellm.litellm_core_utils.url_utils import SSRFError, validate_url from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.agent_endpoints.utils import merge_agent_headers from litellm.proxy.auth.user_api_key_auth import user_api_key_auth @@ -19,9 +21,66 @@ from litellm.types.utils import all_litellm_params router = APIRouter() +_PASCAL_TO_WIRE: Dict[str, str] = { + "GetTask": "tasks/get", + "ListTasks": "tasks/list", + "CancelTask": "tasks/cancel", + "SubscribeToTask": "tasks/resubscribe", + "CreateTaskPushNotificationConfig": "tasks/pushNotificationConfig/set", + "GetTaskPushNotificationConfig": "tasks/pushNotificationConfig/get", + "ListTaskPushNotificationConfigs": "tasks/pushNotificationConfig/list", + "DeleteTaskPushNotificationConfig": "tasks/pushNotificationConfig/delete", + "GetExtendedAgentCard": "agent/getAuthenticatedExtendedCard", +} + + +def _validate_push_notification_url(url: str) -> None: + parsed = urlparse(url) + if parsed.scheme != "https": + raise HTTPException( + status_code=400, + detail="Push notification URL must use HTTPS", + ) + try: + validate_url(url) + except (SSRFError, ValueError) as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + +def _caller_identity_headers(user_api_key_dict: UserAPIKeyAuth) -> Dict[str, str]: + headers: Dict[str, str] = {} + if user_api_key_dict.user_id: + headers["X-LiteLLM-User-Id"] = user_api_key_dict.user_id + if user_api_key_dict.team_id: + headers["X-LiteLLM-Team-Id"] = user_api_key_dict.team_id + return headers + + +def _forwarding_headers( + user_api_key_dict: UserAPIKeyAuth, + request_data: dict, + agent_extra_headers: Optional[Dict[str, str]], +) -> Optional[Dict[str, str]]: + sanitized = ( + { + k: v + for k, v in agent_extra_headers.items() + if not k.lower().startswith("x-litellm-") + } + if agent_extra_headers + else None + ) + merged = merge_agent_headers(dynamic_headers=sanitized, static_headers=None) or {} + identity = _caller_identity_headers(user_api_key_dict) + trace_id = request_data.get("litellm_trace_id") + if trace_id: + identity["X-LiteLLM-Trace-Id"] = str(trace_id) + merged.update(identity) + return merged or None + def _jsonrpc_error( - request_id: Optional[str], + request_id: Optional[Any], code: int, message: str, status_code: int = 400, @@ -67,9 +126,158 @@ def _enforce_inbound_trace_id(agent: Any, request: Request) -> None: ) +async def _forward_jsonrpc( + agent_url: str, + body: dict, + extra_headers: Optional[Dict[str, str]] = None, +) -> dict: + from litellm.llms.custom_httpx.http_handler import get_async_httpx_client + from litellm.types.llms.custom_http import httpxSpecialProvider + + headers = {"Content-Type": "application/json", **(extra_headers or {})} + handler = get_async_httpx_client( + llm_provider=httpxSpecialProvider.A2A, + params={"timeout": 60.0}, + ) + resp = await handler.post(agent_url, json=body, headers=headers) + try: + result = resp.json() + except Exception: + resp.raise_for_status() + raise + if not resp.is_success and "error" not in result: + resp.raise_for_status() + return result + + +async def _a2a_sse_event_source( + agent_url: str, + body: dict, + request_id: Optional[Any] = None, + extra_headers: Optional[Dict[str, str]] = None, +) -> AsyncGenerator[dict, None]: + """Stream an upstream A2A SSE response as parsed JSON-RPC event dicts. + + Upstream HTTP/JSON-RPC errors are surfaced as a single JSON-RPC error event + so the caller can relay them instead of breaking the stream. + """ + from litellm.llms.custom_httpx.http_handler import get_async_httpx_client + from litellm.types.agents import _normalize_a2a_jsonrpc_response + from litellm.types.llms.custom_http import httpxSpecialProvider + + headers = { + "Content-Type": "application/json", + "Accept": "text/event-stream", + **(extra_headers or {}), + } + handler = get_async_httpx_client( + llm_provider=httpxSpecialProvider.A2A, + params={"timeout": None}, + ) + async_client = handler.client + req = async_client.build_request("POST", agent_url, json=body, headers=headers) + resp = await async_client.send(req, stream=True) + try: + if not resp.is_success: + error_body = await resp.aread() + error_event: Optional[dict] = None + try: + parsed = json.loads(error_body) + if isinstance(parsed, dict) and "error" in parsed: + error_event = _normalize_a2a_jsonrpc_response( + parsed, request_id=request_id + ) + except Exception: + error_event = None + yield error_event or { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": -32603, "message": resp.reason_phrase}, + } + return + async for line in resp.aiter_lines(): + stripped = line.strip() + if not stripped.startswith("data:"): + continue + payload = stripped[len("data:") :].strip() + if not payload: + continue + try: + yield json.loads(payload) + except Exception: + continue + finally: + await resp.aclose() + + +async def _forward_jsonrpc_sse( + agent_url: str, + body: dict, + request_id: Optional[Any] = None, + extra_headers: Optional[Dict[str, str]] = None, + proxy_logging_obj: Optional[Any] = None, + user_api_key_dict: Optional[Any] = None, + request_data: Optional[dict] = None, +) -> StreamingResponse: + event_source = _a2a_sse_event_source( + agent_url, body, request_id=request_id, extra_headers=extra_headers + ) + + def _serialize_chunk(chunk: Any) -> str: + return f"data: {json.dumps(chunk)}\n\n" + + def _serialize_error(proxy_exc: Any) -> str: + return ( + "data: " + + json.dumps( + { + "jsonrpc": "2.0", + "id": request_id, + "error": { + "code": -32603, + "message": getattr(proxy_exc, "message", str(proxy_exc)), + }, + } + ) + + "\n\n" + ) + + if ( + proxy_logging_obj is not None + and user_api_key_dict is not None + and request_data is not None + ): + # Route streamed events through the shared streaming generator so the + # post-call streaming hook (and therefore agent guardrails) inspects + # tasks/resubscribe output the same way message/stream does. + from litellm.proxy.common_request_processing import ( + ProxyBaseLLMRequestProcessing, + ) + + generator: AsyncGenerator[str, None] = ( + ProxyBaseLLMRequestProcessing.async_streaming_data_generator( + response=event_source, + user_api_key_dict=user_api_key_dict, + request_data=request_data, + proxy_logging_obj=proxy_logging_obj, + serialize_chunk=_serialize_chunk, + serialize_error=_serialize_error, + ) + ) + else: + + async def _passthrough() -> AsyncGenerator[str, None]: + async for chunk in event_source: + yield _serialize_chunk(chunk) + + generator = _passthrough() + + return StreamingResponse(generator, media_type="text/event-stream") + + async def _handle_stream_message( api_base: Optional[str], - request_id: str, + request_id: Any, params: dict, litellm_params: Optional[dict] = None, agent_id: Optional[str] = None, @@ -310,8 +518,6 @@ async def invoke_agent_a2a( # noqa: PLR0915 - message/send: Send a message and get a response - message/stream: Send a message and stream the response """ - from litellm.a2a_protocol import asend_message - from litellm.a2a_protocol.main import A2A_SDK_AVAILABLE from litellm.proxy.agent_endpoints.auth.agent_permission_handler import ( AgentRequestHandler, ) @@ -322,9 +528,11 @@ async def invoke_agent_a2a( # noqa: PLR0915 version, ) - body = {} + body: Dict[str, Any] = {} + request_data: Dict[str, Any] = body try: body = await request.json() + request_data = body verbose_proxy_logger.debug(f"A2A request for agent '{agent_id}': {body}") @@ -334,11 +542,14 @@ async def invoke_agent_a2a( # noqa: PLR0915 body.get("id"), -32600, "Invalid Request: jsonrpc must be '2.0'" ) - request_id = body.get("id") - method = body.get("method") + request_id: Optional[Any] = body.get("id") + method: Optional[str] = body.get("method") params = body.get("params", {}) - if params: + if method: + method = _PASCAL_TO_WIRE.get(method, method) + + if isinstance(params, dict): # extract any litellm params from the params - eg. 'guardrails' # ``metadata`` is intentionally excluded: it's a first-class A2A # ``MessageSendParams`` field that the completion bridge forwards @@ -347,20 +558,12 @@ async def invoke_agent_a2a( # noqa: PLR0915 # silently drop the caller's A2A request-level metadata. params_to_remove = [] for key, value in params.items(): - if key in all_litellm_params and key != "metadata": + if key in all_litellm_params and key not in {"id", "metadata"}: params_to_remove.append(key) body[key] = value for key in params_to_remove: params.pop(key) - if not A2A_SDK_AVAILABLE: - return _jsonrpc_error( - request_id, - -32603, - "Server error: 'a2a' package not installed. Please install 'a2a-sdk'.", - 500, - ) - # Find the agent agent = _get_agent(agent_id) if agent is None: @@ -441,6 +644,7 @@ async def invoke_agent_a2a( # noqa: PLR0915 route_type="asend_message", version=version, ) + request_data = data # Build merged headers for the backend agent static_headers: Dict[str, str] = dict(agent.static_headers or {}) @@ -453,9 +657,10 @@ async def invoke_agent_a2a( # noqa: PLR0915 # 1. Admin-configured extra_headers: forward named headers from client request if agent.extra_headers: for header_name in agent.extra_headers: - val = normalized.get(header_name.lower()) + header_name_str = str(header_name) + val = normalized.get(header_name_str.lower()) if val is not None: - dynamic_headers[header_name] = val + dynamic_headers[header_name_str] = val # 2. Convention-based forwarding: x-a2a-{agent_id_or_name}-{header_name} # Matches both agent_id (UUID) and agent_name (alias), case-insensitive. @@ -489,10 +694,20 @@ async def invoke_agent_a2a( # noqa: PLR0915 # Route through SDK functions if method == "message/send": + from litellm.a2a_protocol import asend_message + from litellm.a2a_protocol.main import A2A_SDK_AVAILABLE + + if not A2A_SDK_AVAILABLE: + return _jsonrpc_error( + request_id, + -32603, + "Server error: 'a2a' package not installed. Please install 'a2a-sdk'.", + 500, + ) from a2a.types import MessageSendParams, SendMessageRequest a2a_request = SendMessageRequest( - id=request_id, + id=request_id if request_id is not None else "", params=MessageSendParams(**params), ) # Defer spend-log until after post_call_success_hook so guardrail @@ -532,7 +747,7 @@ async def invoke_agent_a2a( # noqa: PLR0915 elif method == "message/stream": return await _handle_stream_message( api_base=agent_url, - request_id=request_id, + request_id=request_id if request_id is not None else "", params=params, litellm_params=litellm_params, agent_id=agent.agent_id, @@ -543,6 +758,106 @@ async def invoke_agent_a2a( # noqa: PLR0915 request_data=data, proxy_logging_obj=proxy_logging_obj, ) + elif method in { + "tasks/get", + "tasks/list", + "tasks/cancel", + "tasks/pushNotificationConfig/set", + "tasks/pushNotificationConfig/get", + "tasks/pushNotificationConfig/list", + "tasks/pushNotificationConfig/delete", + "agent/getAuthenticatedExtendedCard", + }: + if not agent_url: + return _jsonrpc_error( + request_id, -32000, f"Agent '{agent_id}' has no URL configured", 500 + ) + if method == "tasks/pushNotificationConfig/set": + if not isinstance(params, dict): + raise HTTPException( + status_code=400, + detail="params must be an object", + ) + push_config = params.get("pushNotificationConfig", {}) + if "pushNotificationConfig" in params and not isinstance( + push_config, dict + ): + raise HTTPException( + status_code=400, + detail="pushNotificationConfig must be an object", + ) + for callback_url in (params.get("url"), push_config.get("url")): + if not callback_url: + continue + if not isinstance(callback_url, str): + raise HTTPException( + status_code=400, + detail="Push notification URL must be a string", + ) + _validate_push_notification_url(callback_url) + forward_body = { + "jsonrpc": "2.0", + "id": request_id, + "method": method, + "params": params, + } + caller_headers = _forwarding_headers( + user_api_key_dict=user_api_key_dict, + request_data=data, + agent_extra_headers=agent_extra_headers, + ) + result = await _forward_jsonrpc( + agent_url, forward_body, extra_headers=caller_headers + ) + if method == "agent/getAuthenticatedExtendedCard": + if isinstance(result.get("result"), dict) and "url" in result["result"]: + result["result"][ + "url" + ] = f"{str(request.base_url).rstrip('/')}/a2a/{agent_id}" + from litellm.types.agents import LiteLLMSendMessageResponse + + response = LiteLLMSendMessageResponse.from_dict( + result, request_id=request_id + ) + response = await proxy_logging_obj.post_call_success_hook( + user_api_key_dict=user_api_key_dict, + data=data, + response=response, + ) + return JSONResponse( + content=( + response.model_dump(mode="json", exclude_none=True) + if hasattr(response, "model_dump") + else response + ) + ) + + elif method == "tasks/resubscribe": + if not agent_url: + return _jsonrpc_error( + request_id, -32000, f"Agent '{agent_id}' has no URL configured", 500 + ) + forward_body = { + "jsonrpc": "2.0", + "id": request_id, + "method": method, + "params": params, + } + sse_caller_headers = _forwarding_headers( + user_api_key_dict=user_api_key_dict, + request_data=data, + agent_extra_headers=agent_extra_headers, + ) + return await _forward_jsonrpc_sse( + agent_url, + forward_body, + request_id=request_id, + extra_headers=sse_caller_headers, + proxy_logging_obj=proxy_logging_obj, + user_api_key_dict=user_api_key_dict, + request_data=data, + ) + else: return _jsonrpc_error(request_id, -32601, f"Method '{method}' not found") @@ -550,4 +865,12 @@ async def invoke_agent_a2a( # noqa: PLR0915 raise except Exception as e: verbose_proxy_logger.exception(f"Error invoking agent: {e}") + try: + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, + original_exception=e, + request_data=request_data, + ) + except Exception: + pass return _jsonrpc_error(body.get("id"), -32603, f"Internal error: {str(e)}", 500) diff --git a/litellm/types/agents.py b/litellm/types/agents.py index 8556b6bac9..f34631b560 100644 --- a/litellm/types/agents.py +++ b/litellm/types/agents.py @@ -298,6 +298,23 @@ class MakeAgentsPublicRequest(BaseModel): agent_ids: List[str] +def _normalize_a2a_jsonrpc_response( + response_dict: Dict[str, Any], + request_id: Optional[Any] = None, +) -> Dict[str, Any]: + """ + Ensure JSON-RPC responses include ``id`` when the caller supplied one. + + The a2a SDK may omit ``id`` on error payloads even when the upstream agent + returned it. Backfill from the outbound request id so LiteLLM can surface the + agent error instead of failing Pydantic validation. + """ + normalized = dict(response_dict) + if normalized.get("id") is None and request_id is not None: + normalized["id"] = str(request_id) + return normalized + + class LiteLLMSendMessageResponse(LiteLLMPydanticObjectBase): """ LiteLLM wrapper for A2A SendMessageResponse. @@ -322,31 +339,42 @@ class LiteLLMSendMessageResponse(LiteLLMPydanticObjectBase): @classmethod def from_a2a_response( - cls, response: "SendMessageResponse" + cls, + response: "SendMessageResponse", + request_id: Optional[Any] = None, ) -> "LiteLLMSendMessageResponse": """ Create a LiteLLMSendMessageResponse from an a2a SDK SendMessageResponse. Args: response: The a2a SDK SendMessageResponse + request_id: JSON-RPC request id to backfill when the SDK omits it on errors Returns: LiteLLMSendMessageResponse with _hidden_params support """ - # Convert the a2a response to a dict response_dict = response.model_dump(mode="json", exclude_none=True) - + response_dict = _normalize_a2a_jsonrpc_response( + response_dict, request_id=request_id + ) return cls(**response_dict) @classmethod - def from_dict(cls, response_dict: Dict[str, Any]) -> "LiteLLMSendMessageResponse": + def from_dict( + cls, + response_dict: Dict[str, Any], + request_id: Optional[Any] = None, + ) -> "LiteLLMSendMessageResponse": """ Create a LiteLLMSendMessageResponse from a dict. Args: response_dict: Dict with A2A response structure + request_id: JSON-RPC request id to backfill when missing on error payloads Returns: LiteLLMSendMessageResponse with _hidden_params support """ - return cls(**response_dict) + return cls( + **_normalize_a2a_jsonrpc_response(response_dict, request_id=request_id) + ) diff --git a/tests/test_litellm/a2a_protocol/test_send_message_response.py b/tests/test_litellm/a2a_protocol/test_send_message_response.py new file mode 100644 index 0000000000..832aa288c7 --- /dev/null +++ b/tests/test_litellm/a2a_protocol/test_send_message_response.py @@ -0,0 +1,43 @@ +"""Tests for LiteLLMSendMessageResponse JSON-RPC normalization.""" + +from litellm.types.agents import LiteLLMSendMessageResponse + + +def test_from_dict_backfills_id_on_agent_error_response(): + agent_error = { + "jsonrpc": "2.0", + "error": {"code": -32054, "message": "Session not found"}, + } + + response = LiteLLMSendMessageResponse.from_dict( + agent_error, request_id="r1" + ) + + assert response.id == "r1" + assert response.error == {"code": -32054, "message": "Session not found"} + assert response.result is None + + +def test_from_dict_preserves_existing_id(): + payload = { + "id": "upstream-id", + "jsonrpc": "2.0", + "error": {"code": -32001, "message": "Task not found"}, + } + + response = LiteLLMSendMessageResponse.from_dict( + payload, request_id="r1" + ) + + assert response.id == "upstream-id" + + +def test_from_dict_without_request_id_still_requires_id(): + try: + LiteLLMSendMessageResponse.from_dict( + {"jsonrpc": "2.0", "error": {"code": -32054, "message": "x"}} + ) + except Exception as exc: + assert "id" in str(exc).lower() + else: + raise AssertionError("expected validation error when id and request_id missing") diff --git a/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py b/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py index a32f2eadb9..07e878401e 100644 --- a/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py +++ b/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py @@ -5,7 +5,9 @@ Tests that invoke_agent_a2a properly integrates with add_litellm_data_to_request """ import json +import socket import sys +from contextlib import ExitStack from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -348,3 +350,1213 @@ async def test_invoke_agent_a2a_injects_authenticated_key_hash_for_bridge(): captured.get("litellm_params", {}).get(A2A_USER_API_KEY_HASH_PARAM) == mock_user_api_key_dict.api_key ), "authenticated key hash was not forwarded to the completion bridge" + + +def _make_agent_mock(url: str = "http://backend-agent:10001") -> MagicMock: + agent = MagicMock() + agent.agent_id = "test-agent" + agent.agent_name = "test-agent" + agent.agent_card_params = {"url": url, "name": "Test Agent"} + agent.litellm_params = {} + agent.static_headers = None + agent.extra_headers = None + return agent + + +def _make_request_mock( + method: str, params: dict, request_id: object = "req-1" +) -> MagicMock: + req = MagicMock() + req.headers = {} + req.json = AsyncMock( + return_value={ + "jsonrpc": "2.0", + "id": request_id, + "method": method, + "params": params, + } + ) + return req + + +def _base_patches(agent: MagicMock): + return [ + patch( + "litellm.proxy.agent_endpoints.a2a_endpoints._get_agent", + return_value=agent, + ), + patch( + "litellm.proxy.agent_endpoints.auth.agent_permission_handler.AgentRequestHandler.is_agent_allowed", + new=AsyncMock(return_value=True), + ), + patch( + "litellm.proxy.common_request_processing.add_litellm_data_to_request", + new=AsyncMock(side_effect=_add_proxy_data), + ), + patch("litellm.proxy.proxy_server.general_settings", {}), + patch("litellm.proxy.proxy_server.proxy_config", MagicMock()), + patch("litellm.proxy.proxy_server.version", "1.0.0"), + ] + + +async def _add_proxy_data(data, **kwargs): + data["proxy_server_request"] = { + "url": "http://localhost:4000", + "method": "POST", + "headers": {}, + "body": {}, + } + data.setdefault("metadata", {}) + return data + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["message/send", "message/stream"]) +async def test_message_methods_preserve_numeric_zero_request_id(method: str): + from fastapi.responses import JSONResponse + from litellm.proxy._types import UserAPIKeyAuth + + class MessageSendParams: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + class SendMessageRequest: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + agent = _make_agent_mock() + params = { + "message": { + "role": "user", + "parts": [{"kind": "text", "text": "Hello"}], + "messageId": "msg-123", + } + } + mock_request = _make_request_mock(method, params, request_id=0) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + captured = {} + + async def capture_asend_message(request, **kwargs): + captured["request_id"] = request.id + response = MagicMock() + response.model_dump.return_value = { + "jsonrpc": "2.0", + "id": request.id, + "result": {"status": "success"}, + } + return response + + async def capture_stream_message(**kwargs): + captured["request_id"] = kwargs["request_id"] + return JSONResponse({"jsonrpc": "2.0", "id": kwargs["request_id"]}) + + mock_a2a_types = MagicMock() + mock_a2a_types.MessageSendParams = MessageSendParams + mock_a2a_types.SendMessageRequest = SendMessageRequest + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context(patch("litellm.a2a_protocol.main.A2A_SDK_AVAILABLE", True)) + if method == "message/send": + stack.enter_context( + patch.dict( + sys.modules, + {"a2a": MagicMock(), "a2a.types": mock_a2a_types}, + ) + ) + stack.enter_context( + patch( + "litellm.a2a_protocol.asend_message", + new=AsyncMock(side_effect=capture_asend_message), + ) + ) + else: + stack.enter_context( + patch( + "litellm.proxy.agent_endpoints.a2a_endpoints._handle_stream_message", + new=AsyncMock(side_effect=capture_stream_message), + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert captured["request_id"] == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "method,params", + [ + ("tasks/get", {"id": "task-1"}), + ("tasks/list", {"contextId": "ctx-1"}), + ("tasks/cancel", {"id": "task-1"}), + ( + "tasks/pushNotificationConfig/set", + {"taskId": "task-1", "url": "https://webhook.example.com"}, + ), + ("tasks/pushNotificationConfig/get", {"taskId": "task-1", "id": "cfg-1"}), + ("tasks/pushNotificationConfig/list", {"taskId": "task-1"}), + ("tasks/pushNotificationConfig/delete", {"taskId": "task-1", "id": "cfg-1"}), + ], +) +async def test_task_methods_forward_jsonrpc(method: str, params: dict): + from litellm.proxy._types import UserAPIKeyAuth + + upstream_response = { + "jsonrpc": "2.0", + "id": "req-1", + "result": {"id": "task-1", "status": {"state": "completed"}}, + } + agent = _make_agent_mock() + mock_request = _make_request_mock(method, params) + + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_response + mock_http_response.is_success = True + mock_http_response.raise_for_status = MagicMock() + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + mock_handler.client = MagicMock() + + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + stack.enter_context( + patch( + "litellm.proxy.agent_endpoints.a2a_endpoints.validate_url", + return_value=("https://webhook.example.com", "webhook.example.com"), + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + body = json.loads(response.body.decode()) + assert body["jsonrpc"] == "2.0" + assert body["result"]["id"] == "task-1" + + posted = mock_handler.post.call_args + assert posted is not None + forwarded_body = posted.kwargs.get("json") or posted.args[1] + assert forwarded_body["method"] == method + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["tasks/get", "tasks/resubscribe"]) +async def test_task_methods_extract_litellm_params_before_forwarding(method: str): + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + params = { + "id": "task-1", + "guardrails": ["guardrail-1"], + "tags": ["tag-1"], + } + mock_request = _make_request_mock(method, params) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + captured_data = {} + + async def capture_proxy_data(data, **kwargs): + captured_data.update(data) + return await _add_proxy_data(data, **kwargs) + + upstream_response = { + "jsonrpc": "2.0", + "id": "req-1", + "result": {"id": "task-1", "status": {"state": "completed"}}, + } + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_response + mock_http_response.is_success = True + + async def fake_aiter_lines(): + yield 'data: {"jsonrpc":"2.0","id":"req-1","result":{"taskId":"task-1"}}' + + mock_resp = AsyncMock() + mock_resp.is_success = True + mock_resp.aiter_lines = fake_aiter_lines + mock_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_resp) + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + mock_handler.client = mock_async_client + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.proxy.common_request_processing.add_litellm_data_to_request", + new=AsyncMock(side_effect=capture_proxy_data), + ) + ) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + if method == "tasks/resubscribe": + async for _ in response.body_iterator: + pass + + if method == "tasks/resubscribe": + forwarded_body = mock_async_client.build_request.call_args.kwargs["json"] + else: + forwarded_body = mock_handler.post.call_args.kwargs["json"] + assert forwarded_body["params"] == {"id": "task-1"} + assert captured_data["guardrails"] == ["guardrail-1"] + assert captured_data["tags"] == ["tag-1"] + + +@pytest.mark.asyncio +async def test_subscribe_to_task_returns_sse_stream(): + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock("SubscribeToTask", {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + sse_lines = [ + 'data: {"jsonrpc":"2.0","id":"req-1","result":{"taskId":"task-1","status":{"state":"working"}}}', + 'data: {"jsonrpc":"2.0","id":"req-1","result":{"taskId":"task-1","status":{"state":"completed"}}}', + ] + + async def fake_aiter_lines(): + for line in sse_lines: + yield line + + mock_resp = AsyncMock() + mock_resp.is_success = True + mock_resp.aiter_lines = fake_aiter_lines + mock_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_resp) + + mock_handler = MagicMock() + mock_handler.client = mock_async_client + mock_handler.post = AsyncMock() + + chunks = [] + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert response.media_type == "text/event-stream" + async for chunk in response.body_iterator: + chunks.append(chunk) + + full = "".join(chunks) + assert "working" in full + assert "completed" in full + + +@pytest.mark.asyncio +async def test_subscribe_to_task_calls_pre_call_hook(): + """tasks/resubscribe must run pre_call_hook so guardrails configured on + the agent are enforced before streaming begins.""" + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock("tasks/resubscribe", {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + async def fake_aiter_lines(): + yield 'data: {"jsonrpc":"2.0","id":"req-1","result":{"taskId":"task-1","status":{"state":"completed"}}}' + + mock_resp = AsyncMock() + mock_resp.is_success = True + mock_resp.aiter_lines = fake_aiter_lines + mock_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_resp) + + mock_handler = MagicMock() + mock_handler.client = mock_async_client + mock_handler.post = AsyncMock() + + async def _passthrough_iterator(response, **kwargs): + async for chunk in response: + yield chunk + + mock_proxy_logging = MagicMock() + mock_proxy_logging.pre_call_hook = AsyncMock( + side_effect=lambda user_api_key_dict, data, call_type: data + ) + mock_proxy_logging.async_post_call_streaming_iterator_hook = _passthrough_iterator + mock_proxy_logging.post_call_failure_hook = AsyncMock(return_value=None) + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + stack.enter_context( + patch( + "litellm.proxy.proxy_server.proxy_logging_obj", + mock_proxy_logging, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert response.media_type == "text/event-stream" + async for _ in response.body_iterator: + pass + + mock_proxy_logging.pre_call_hook.assert_awaited_once() + call_kwargs = mock_proxy_logging.pre_call_hook.await_args.kwargs + assert call_kwargs.get("call_type") == "asend_message" + assert call_kwargs.get("user_api_key_dict") == user_api_key_dict + + +@pytest.mark.asyncio +async def test_subscribe_to_task_runs_post_call_streaming_guardrail(): + """tasks/resubscribe must route streamed events through the post-call + streaming hook so output guardrails configured on the agent inspect the + streamed task content. Regression: the SSE path previously returned the raw + upstream stream and bypassed guardrails entirely.""" + import litellm + from litellm.integrations.custom_guardrail import CustomGuardrail + from litellm.proxy._types import UserAPIKeyAuth + + inspected: list = [] + + class _RecordingGuardrail(CustomGuardrail): + async def async_post_call_streaming_hook(self, user_api_key_dict, response): + inspected.append(response) + return response + + guardrail = _RecordingGuardrail( + guardrail_name="record-a2a", default_on=True, event_hook="post_call" + ) + + agent = _make_agent_mock() + mock_request = _make_request_mock("tasks/resubscribe", {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + async def fake_aiter_lines(): + yield ( + 'data: {"jsonrpc":"2.0","id":"req-1","result":' + '{"kind":"message","parts":[{"kind":"text","text":"resubscribe-secret"}]}}' + ) + + mock_resp = AsyncMock() + mock_resp.is_success = True + mock_resp.aiter_lines = fake_aiter_lines + mock_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_resp) + + mock_handler = MagicMock() + mock_handler.client = mock_async_client + mock_handler.post = AsyncMock() + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + stack.enter_context(patch.object(litellm, "callbacks", [guardrail])) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert response.media_type == "text/event-stream" + async for _ in response.body_iterator: + pass + + assert any("resubscribe-secret" in str(r) for r in inspected), ( + "tasks/resubscribe streamed content was not passed to the post-call " + "streaming guardrail hook" + ) + + +@pytest.mark.asyncio +async def test_task_method_failure_hook_uses_enriched_request_data(): + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock("tasks/get", {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + async def add_proxy_data_copy(data, **kwargs): + enriched = dict(data) + enriched["proxy_server_request"] = { + "url": "http://localhost:4000", + "method": "POST", + "headers": {}, + "body": {}, + } + enriched.setdefault("metadata", {}) + return enriched + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(side_effect=RuntimeError("upstream failed")) + + mock_proxy_logging = MagicMock() + mock_proxy_logging.pre_call_hook = AsyncMock( + side_effect=lambda user_api_key_dict, data, call_type: data + ) + mock_proxy_logging.post_call_failure_hook = AsyncMock(return_value=None) + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.proxy.common_request_processing.add_litellm_data_to_request", + new=AsyncMock(side_effect=add_proxy_data_copy), + ) + ) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + stack.enter_context( + patch( + "litellm.proxy.proxy_server.proxy_logging_obj", + mock_proxy_logging, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + body = json.loads(response.body.decode()) + assert body["error"]["code"] == -32603 + failure_data = mock_proxy_logging.post_call_failure_hook.await_args.kwargs[ + "request_data" + ] + assert failure_data.get("litellm_call_id") + assert failure_data.get("agent_id") == "test-agent" + + +@pytest.mark.asyncio +async def test_get_extended_agent_card_rewrites_url(): + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock("GetExtendedAgentCard", {}) + mock_request.base_url = "http://localhost:4000/" + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + upstream_card = { + "name": "Test Agent", + "url": "http://backend-agent:10001", + "description": "A test agent", + } + upstream_response = {"jsonrpc": "2.0", "id": "req-1", "result": upstream_card} + + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_response + mock_http_response.is_success = True + mock_http_response.raise_for_status = MagicMock() + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + mock_handler.client = MagicMock() + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + body = json.loads(response.body.decode()) + assert body["result"]["url"] == "http://localhost:4000/a2a/test-agent" + assert body["result"]["name"] == "Test Agent" + + +@pytest.mark.asyncio +async def test_unknown_method_returns_jsonrpc_error(): + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock("SomeUnknownMethod", {}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + body = json.loads(response.body.decode()) + assert body["error"]["code"] == -32601 + assert "SomeUnknownMethod" in body["error"]["message"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "pascal_method,expected_wire_method", + [ + ("GetTask", "tasks/get"), + ("ListTasks", "tasks/list"), + ("CancelTask", "tasks/cancel"), + ("SubscribeToTask", "tasks/resubscribe"), + ("CreateTaskPushNotificationConfig", "tasks/pushNotificationConfig/set"), + ("GetTaskPushNotificationConfig", "tasks/pushNotificationConfig/get"), + ("ListTaskPushNotificationConfigs", "tasks/pushNotificationConfig/list"), + ("DeleteTaskPushNotificationConfig", "tasks/pushNotificationConfig/delete"), + ("GetExtendedAgentCard", "agent/getAuthenticatedExtendedCard"), + ], +) +async def test_pascal_method_names_normalize_to_wire_format( + pascal_method: str, expected_wire_method: str +): + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock(pascal_method, {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + upstream_response = {"jsonrpc": "2.0", "id": "req-1", "result": {"id": "task-1"}} + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_response + mock_http_response.is_success = True + mock_http_response.raise_for_status = MagicMock() + + async def _empty_aiter_lines(): + return + yield # make it an async generator + + mock_sse_resp = AsyncMock() + mock_sse_resp.is_success = True + mock_sse_resp.aiter_lines = _empty_aiter_lines + mock_sse_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_sse_resp) + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + mock_handler.client = mock_async_client + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + if expected_wire_method == "tasks/resubscribe": + assert response.media_type == "text/event-stream" + async for _ in response.body_iterator: + pass + else: + body = json.loads(response.body.decode()) + assert "error" not in body, f"Got error: {body}" + + if expected_wire_method != "tasks/resubscribe": + posted = mock_handler.post.call_args + forwarded_body = posted.kwargs.get("json") or posted.args[1] + assert forwarded_body["method"] == expected_wire_method, ( + f"Expected '{expected_wire_method}' forwarded for PascalCase '{pascal_method}', " + f"but got '{forwarded_body['method']}'" + ) + + +@pytest.mark.asyncio +async def test_task_method_upstream_jsonrpc_error_on_http_4xx_is_relayed(): + """When upstream returns HTTP 4xx with a JSON-RPC error body, the error body + must be relayed to the client unchanged, not replaced with a generic string.""" + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock("tasks/get", {"id": "nonexistent"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + upstream_error = { + "jsonrpc": "2.0", + "id": "req-1", + "error": {"code": -32001, "message": "Task not found"}, + } + + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_error + mock_http_response.is_success = False + mock_http_response.raise_for_status = MagicMock( + side_effect=Exception("404 Not Found") + ) + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + mock_handler.client = MagicMock() + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + body = json.loads(response.body.decode()) + assert body["error"]["code"] == -32001 + assert body["error"]["message"] == "Task not found" + + +@pytest.mark.asyncio +async def test_subscribe_to_task_upstream_error_yields_jsonrpc_error_event(): + """When upstream returns a non-2xx response for tasks/resubscribe, the SSE + stream must yield a JSON-RPC error event instead of silently breaking.""" + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock("tasks/resubscribe", {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + mock_resp = AsyncMock() + mock_resp.is_success = False + mock_resp.status_code = 404 + mock_resp.reason_phrase = "Not Found" + mock_resp.aread = AsyncMock( + return_value=b'{"jsonrpc":"2.0","error":{"code":-32001,"message":"Task not found"}}' + ) + mock_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_resp) + + mock_handler = MagicMock() + mock_handler.client = mock_async_client + mock_handler.post = AsyncMock() + + chunks = [] + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert response.media_type == "text/event-stream" + async for chunk in response.body_iterator: + chunks.append(chunk) + + full = "".join(chunks) + body = json.loads(full.removeprefix("data: ").strip()) + assert body["id"] == "req-1" + assert body["error"]["code"] == -32001 + assert body["error"]["message"] == "Task not found" + + +@pytest.mark.asyncio +async def test_forward_jsonrpc_sse_fallback_error_uses_jsonrpc_error_code(): + mock_resp = AsyncMock() + mock_resp.is_success = False + mock_resp.status_code = 503 + mock_resp.reason_phrase = "Service Unavailable" + mock_resp.aread = AsyncMock(return_value=b"upstream unavailable") + mock_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_resp) + + mock_handler = MagicMock() + mock_handler.client = mock_async_client + + with patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ): + from litellm.proxy.agent_endpoints.a2a_endpoints import _forward_jsonrpc_sse + + response = await _forward_jsonrpc_sse( + agent_url="http://backend-agent:10001", + body={"jsonrpc": "2.0", "id": "req-1", "method": "tasks/resubscribe"}, + request_id="req-1", + ) + + chunks = [] + async for chunk in response.body_iterator: + chunks.append(chunk) + + body = json.loads("".join(chunks).removeprefix("data: ").strip()) + assert body["error"]["code"] == -32603 + assert body["error"]["message"] == "Service Unavailable" + + +@pytest.mark.asyncio +async def test_task_methods_forward_caller_identity_headers(): + """Task operations must forward X-LiteLLM-User-Id and X-LiteLLM-Team-Id so the + upstream agent can scope resources to the authenticated caller.""" + from litellm.proxy._types import UserAPIKeyAuth + + upstream_response = { + "jsonrpc": "2.0", + "id": "req-1", + "result": {"id": "task-1", "status": {"state": "completed"}}, + } + agent = _make_agent_mock() + mock_request = _make_request_mock("tasks/get", {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth( + api_key="sk-test", user_id="user-abc", team_id="team-xyz" + ) + + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_response + mock_http_response.is_success = True + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + posted_headers = mock_handler.post.call_args.kwargs.get("headers") or {} + assert posted_headers.get("X-LiteLLM-User-Id") == "user-abc" + assert posted_headers.get("X-LiteLLM-Team-Id") == "team-xyz" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["tasks/get", "tasks/resubscribe"]) +async def test_task_methods_forward_trace_header(method: str): + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock(method, {"id": "task-1"}) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + async def add_proxy_data_with_trace(data, **kwargs): + data = await _add_proxy_data(data, **kwargs) + data["litellm_trace_id"] = "trace-123" + return data + + upstream_response = { + "jsonrpc": "2.0", + "id": "req-1", + "result": {"id": "task-1", "status": {"state": "completed"}}, + } + + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_response + mock_http_response.is_success = True + + async def fake_aiter_lines(): + yield 'data: {"jsonrpc":"2.0","id":"req-1","result":{"taskId":"task-1"}}' + + mock_resp = AsyncMock() + mock_resp.is_success = True + mock_resp.aiter_lines = fake_aiter_lines + mock_resp.aclose = AsyncMock() + + mock_async_client = MagicMock() + mock_async_client.build_request = MagicMock(return_value=MagicMock()) + mock_async_client.send = AsyncMock(return_value=mock_resp) + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + mock_handler.client = mock_async_client + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.proxy.common_request_processing.add_litellm_data_to_request", + new=AsyncMock(side_effect=add_proxy_data_with_trace), + ) + ) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + response = await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + if method == "tasks/resubscribe": + async for _ in response.body_iterator: + pass + + if method == "tasks/resubscribe": + forwarded_headers = mock_async_client.build_request.call_args.kwargs["headers"] + else: + forwarded_headers = mock_handler.post.call_args.kwargs["headers"] + assert forwarded_headers.get("X-LiteLLM-Trace-Id") == "trace-123" + + +@pytest.mark.asyncio +async def test_push_notification_config_set_rejects_http_url(): + """tasks/pushNotificationConfig/set must reject non-HTTPS callback URLs to prevent SSRF.""" + from fastapi import HTTPException + + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock( + "tasks/pushNotificationConfig/set", + {"taskId": "task-1", "url": "http://internal-webhook.example.com/hook"}, + ) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + with pytest.raises(HTTPException) as exc_info: + await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert exc_info.value.status_code == 400 + assert "HTTPS" in exc_info.value.detail + + +@pytest.mark.asyncio +async def test_push_notification_config_set_rejects_private_ip(): + """tasks/pushNotificationConfig/set must reject callback URLs pointing to private IP ranges.""" + from fastapi import HTTPException + + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock( + "tasks/pushNotificationConfig/set", + {"taskId": "task-1", "url": "https://192.168.1.100/hook"}, + ) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + with pytest.raises(HTTPException) as exc_info: + await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert exc_info.value.status_code == 400 + assert "blocked address" in exc_info.value.detail.lower() + + +@pytest.mark.asyncio +async def test_push_notification_config_set_validates_nested_url_when_top_level_present(): + """A safe top-level params.url must not let a private pushNotificationConfig.url bypass SSRF checks. + + Both URL-bearing fields are forwarded to the agent, so both must be validated independently. + """ + from fastapi import HTTPException + + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock( + "tasks/pushNotificationConfig/set", + { + "taskId": "task-1", + "url": "https://1.1.1.1/hook", + "pushNotificationConfig": {"url": "https://192.168.1.100/hook"}, + }, + ) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + with pytest.raises(HTTPException) as exc_info: + await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert exc_info.value.status_code == 400 + assert "blocked address" in exc_info.value.detail.lower() + + +def test_push_notification_config_set_rejects_private_dns_resolution(): + from fastapi import HTTPException + + from litellm.proxy.agent_endpoints.a2a_endpoints import ( + _validate_push_notification_url, + ) + + with patch( + "litellm.litellm_core_utils.url_utils.socket.getaddrinfo", + return_value=[ + ( + socket.AF_INET, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("10.0.0.5", 443), + ) + ], + ): + with pytest.raises(HTTPException) as exc_info: + _validate_push_notification_url("https://webhook.example.com/hook") + + assert exc_info.value.status_code == 400 + assert "blocked address" in exc_info.value.detail.lower() + + +@pytest.mark.asyncio +async def test_push_notification_config_set_rejects_null_push_config(): + from fastapi import HTTPException + + from litellm.proxy._types import UserAPIKeyAuth + + agent = _make_agent_mock() + mock_request = _make_request_mock( + "tasks/pushNotificationConfig/set", + {"taskId": "task-1", "pushNotificationConfig": None}, + ) + user_api_key_dict = UserAPIKeyAuth(api_key="sk-test", user_id="u1", team_id="t1") + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + with pytest.raises(HTTPException) as exc_info: + await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + assert exc_info.value.status_code == 400 + assert "pushNotificationConfig must be an object" in exc_info.value.detail + + +@pytest.mark.asyncio +async def test_caller_identity_headers_cannot_be_spoofed_via_forwarded_headers(): + """A client must not be able to override X-LiteLLM-User-Id / X-LiteLLM-Team-Id + by including x-a2a--x-litellm-user-id in their request headers. + The authenticated identity must always win.""" + from litellm.proxy._types import UserAPIKeyAuth + + upstream_response = { + "jsonrpc": "2.0", + "id": "req-1", + "result": {"id": "task-1", "status": {"state": "completed"}}, + } + agent = _make_agent_mock() + mock_request = _make_request_mock("tasks/get", {"id": "task-1"}) + mock_request.headers = { + "x-a2a-test-agent-x-litellm-user-id": "attacker-user", + "x-a2a-test-agent-x-litellm-team-id": "attacker-team", + } + user_api_key_dict = UserAPIKeyAuth( + api_key="sk-test", user_id="real-user", team_id="real-team" + ) + + mock_http_response = MagicMock() + mock_http_response.json.return_value = upstream_response + mock_http_response.is_success = True + + mock_handler = MagicMock() + mock_handler.post = AsyncMock(return_value=mock_http_response) + + with ExitStack() as stack: + for p in _base_patches(agent): + stack.enter_context(p) + stack.enter_context( + patch( + "litellm.llms.custom_httpx.http_handler.get_async_httpx_client", + return_value=mock_handler, + ) + ) + + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + await invoke_agent_a2a( + agent_id="test-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=user_api_key_dict, + ) + + posted_headers = mock_handler.post.call_args.kwargs.get("headers") or {} + assert ( + posted_headers.get("X-LiteLLM-User-Id") == "real-user" + ), "authenticated user id must not be overridden by forwarded client headers" + assert ( + posted_headers.get("X-LiteLLM-Team-Id") == "real-team" + ), "authenticated team id must not be overridden by forwarded client headers"