From 8e6dfd18f6a168678a73319abed08f3bc7d3184e Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Tue, 3 Feb 2026 13:56:25 -0800 Subject: [PATCH] route_a2a_agent_request --- litellm/proxy/agent_endpoints/a2a_routing.py | 51 ++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 litellm/proxy/agent_endpoints/a2a_routing.py diff --git a/litellm/proxy/agent_endpoints/a2a_routing.py b/litellm/proxy/agent_endpoints/a2a_routing.py new file mode 100644 index 00000000000..f48cbbb18a8 --- /dev/null +++ b/litellm/proxy/agent_endpoints/a2a_routing.py @@ -0,0 +1,51 @@ +""" +A2A Agent Routing + +Handles routing for A2A agents (models with "a2a/" prefix). +Looks up agents in the registry and injects their API base URL. +""" + +from typing import Any, Optional + +import litellm +from litellm._logging import verbose_proxy_logger +from litellm.proxy.agent_endpoints.agent_registry import global_agent_registry +from litellm.proxy.route_llm_request import ( + ROUTE_ENDPOINT_MAPPING, + ProxyModelNotFoundError, +) + + +async def route_a2a_agent_request(data: dict, route_type: str) -> Optional[Any]: + """ + Route A2A agent requests directly to litellm with injected API base. + + Returns None if not an A2A request (allows normal routing to continue). + """ + model_name = data.get("model", "") + + # Check if this is an A2A agent request + if not isinstance(model_name, str) or not model_name.startswith("a2a/"): + return None + + # Extract agent name (e.g., "a2a/my-agent" -> "my-agent") + agent_name = model_name[4:] + + # Look up agent in registry + agent = global_agent_registry.get_agent_by_name(agent_name) + if agent is None: + verbose_proxy_logger.error(f"[A2A] Agent '{agent_name}' not found in registry") + route_name = ROUTE_ENDPOINT_MAPPING.get(route_type, route_type) + raise ProxyModelNotFoundError(route=route_name, model_name=model_name) + + # Get API base URL from agent config + if not agent.agent_card_params or "url" not in agent.agent_card_params: + verbose_proxy_logger.error(f"[A2A] Agent '{agent_name}' has no URL configured") + route_name = ROUTE_ENDPOINT_MAPPING.get(route_type, route_type) + raise ProxyModelNotFoundError(route=route_name, model_name=model_name) + + # Inject API base and route to litellm + data["api_base"] = agent.agent_card_params["url"] + verbose_proxy_logger.debug(f"[A2A] Routing {model_name} to {data['api_base']}") + + return getattr(litellm, f"{route_type}")(**data)