feat(a2a): well-known agent-card discovery + LangGraph Platform mode#28860
Merged
mateo-berri merged 27 commits intoMay 30, 2026
Conversation
Adds a registration-time discovery flow so admins can paste an upstream agent URL, see its skills/capabilities, pick what to expose, and have the proxy front it with a LiteLLM-shaped agent card. Backend (new litellm/proxy/a2a/ module): - fetch_well_known_card walks /.well-known/agent-card.json, /.well-known/agent.json, /agent.json by default. langgraph_platform mode hits the canonical path with ?assistant_id=<id> (LangGraph serves one shared endpoint per deployment). - merge_agent_card overlays LiteLLM overrides on the upstream card: drops upstream url, forces protocolVersion=1.0, replaces securitySchemes with LiteLLMKey bearer, emits supportedInterfaces pointing at the proxy, filters capabilities to a small allowlist, strips non-v1.0 fields. - POST /v1/a2a/discover returns the raw upstream card (admin-only) so the UI can render skills/capabilities for selection. - create/update/patch agent endpoints pre-generate the agent_id and run merge_agent_card before storing, so DB.agent_card_params already embeds the proxy-fronted URL. UI (ui/litellm-dashboard): - New AgentCardDiscovery component with a parent-driven plan: discovery_mode + params + display URL. For LangGraph the parent composes (api_base, assistant_id); for pure A2A it uses the url field. Component hides the manual URL input when the parent drives. - add_agent_form wires discovery for every non-custom agent type and overlays the user's selections onto agent_card_params at submit, fixing the bug where dynamic agent forms ignored discovery picks. Completion-bridge fixes (paired): - Add kind: "message" to A2A response messages and unwrap result so it's a Message directly per spec (matches a2a SDK SendMessageResponse validation). - Forward A2A metadata to LangGraph runs via extra_body.metadata.
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
Contributor
Greptile SummaryThis PR adds a registration-time A2A agent card discovery flow (
Confidence Score: 5/5Safe to merge — the previously flagged crashes for cardless agents are fixed, the discover endpoint is admin-only and SSRF-guarded, and the A2A response shape changes follow the spec. All three previously flagged crashes (agent_card_params.get() on None in invoke_agent_a2a, dict(None) in get_agent_card, unconditional card synthesis for plain LLM agents) are addressed in this diff. The new discovery flow is isolated behind an admin role check and the existing async_safe_get SSRF guard. The response shape change is spec-mandated and paired with updated tests. No credential-forwarding surface is exposed through the discover endpoint (no headers field in DiscoverAgentRequest). Deletions (duplicate providers/litellm_completion) leave no dangling imports. No files require special attention — the most sensitive paths (agent registration and A2A invocation) both have new targeted tests covering the edge cases introduced by this PR.
|
| Filename | Overview |
|---|---|
| litellm/proxy/a2a/discovery.py | New module: async card fetcher with SSRF guard, fallback path logic, and LangGraph Platform query-param mode — clean and well-tested |
| litellm/proxy/a2a/agent_card.py | New module: pure merge logic that replaces upstream security schemes, filters capabilities, and builds the proxy-fronted card — allowlist and deep-copy guard look correct |
| litellm/proxy/a2a/endpoints.py | New admin-only POST /v1/a2a/discover endpoint; no headers field exposed in DiscoverAgentRequest so credential-forwarding concern from prior review is not applicable |
| litellm/proxy/agent_endpoints/endpoints.py | create/update/patch agent now conditionally run merge_agent_card only when agent_card_params is provided; pre-generates agent_id for create so the stored card can reference its own URL |
| litellm/proxy/agent_endpoints/a2a_endpoints.py | Fixes agent_card_params None crash in invoke_agent_a2a (agent_card_params or {}); adds 404 guard in get_agent_card; excludes 'metadata' key from litellm param stripping to preserve A2A request metadata |
| litellm/a2a_protocol/litellm_completion_bridge/transformation.py | Adds metadata forwarding helpers and applies kind='message' to A2A non-streaming responses; handles A2A parts without explicit kind field; removes deprecated openai_chunk_to_a2a_chunk method |
| litellm/proxy/_lazy_features.py | Splits a2a lazy feature into two: /a2a prefix + /message/send suffix for a2a_endpoints, and /v1/a2a/discover prefix for new a2a_registration module; extracts matches() helper to share logic |
| litellm/a2a_protocol/providers/litellm_completion/handler.py | Deleted — duplicate bridge module removed in favour of litellm_completion_bridge; no remaining imports found in the codebase |
Reviews (23): Last reviewed commit: "fix(a2a): include suffix-matched routes ..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
…ct forwarded metadata - Streaming chunk: move final out of the message object into the result envelope per the A2A spec. - Agent card merge: keep upstream url on the stored card so the runtime invocation path can locate the upstream backend; the public well-known endpoint already rewrites this field to the proxy URL before exposing it to clients. - Completion bridge: apply A2A forward metadata after merging litellm_params so an agent-configured extra_body cannot overwrite the forwarded metadata. Co-authored-by: Yassin Kortam <[email protected]>
|
|
…erge - providers/litellm_completion: move 'final' out of the message object into the result envelope per the A2A spec (matches the bridge fix). - agent endpoints test: the runtime invocation path now preserves the top-level 'url' on the stored card, so update the assertion to match. - completion bridge metadata: when forwarding A2A metadata via extra_body.metadata, merge into any existing extra_body.metadata instead of replacing it, so an agent-configured metadata block is preserved (forward metadata still wins on key conflicts). Co-authored-by: Yassin Kortam <[email protected]>
…eaders field from /v1/a2a/discover Co-authored-by: Yassin Kortam <[email protected]>
The commit afc8b10 bundled real A2A fixes alongside an unintended re-introduction of the */index.html layout that 8513d7f had already reverted. Restore all 35 static-export pages back to the flat *.html structure that matches the upstream main branch. Co-authored-by: Cursor <[email protected]>
UI: - Auto-trigger discovery when connection details are filled; remove the "Use these selections" button (selection syncs live to parent, user just clicks Next). - Edit Settings: auto-discover upstream card on open; cross-check with DB-stored card so only already-saved skills/capabilities are pre-ticked. - Extract shared buildDiscoveryRequest + selectionsFromSavedAgentCard helpers into agent_discovery_utils.ts so both add and edit flows share the same logic. Backend: - agent_card.py: rename the proxy security requirements field from the non-standard ``securityRequirements`` to the spec-correct ``security`` key (matches AgentCard TypedDict and A2A/OpenAPI convention). - agent_card.py: remove ``securityRequirements`` from _ALLOWED_TOP_LEVEL_KEYS. - endpoints.py: _build_merged_agent_card now forwards agent_name and description from the request so the stored card reflects the admin- supplied name, not just whatever the upstream card advertised. - utils.py: remove overly-broad ``or "parts" in result`` fallback; use ``kind == "message"`` check only to avoid false matches on future result types that happen to include a ``parts`` field. - test_agent_card.py: update assertions to expect ``security`` key. Co-authored-by: Cursor <[email protected]>
The previous revert removed __next.* metadata subdirectories from git tracking entirely, but these directories exist on origin/main alongside the flat .html files. Restore them via checkout from origin/main so the PR diff only reflects actual code changes. Co-authored-by: Cursor <[email protected]>
Collaborator
Author
|
@greptileai re review |
The backend /v1/a2a/discover endpoint no longer accepts a headers field (removed in 78591b2 for SSRF safety), so any headers passed through DiscoverAgentCardOptions were silently discarded by the API request body. Remove the field and the conditional that copies it onto the request body.
Collaborator
![veria-ai[bot]](https://avatars.githubusercontent.com/in/1723776?s=60&v=4){kind=small}
Contributor
PR overviewAll previously flagged issues have been addressed. No open security concerns remain on this pull request. Security reviewNo open security issues remain on this pull request. Fixed/addressed: 3 · PR risk: 0/10 |
…shape The agent create/update/patch handlers ran the LiteLLM-fronting merge unconditionally, so registrations that did not provide agent_card_params still ended up with a synthesised card carrying supportedInterfaces, securitySchemes, and default skills. Gate the merge on a non-empty agent_card_params so plain chat/LLM agents stay non-A2A in the registry. Also move kind: 'message' inside the a2a_message dict in the Pydantic AI non-streaming response so its construction matches the completion bridge rather than spreading kind on top of a separate dict.
Collaborator
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.
There are 6 total unresolved issues (including 3 from previous reviews).
Autofix Details
Bugbot Autofix prepared fixes for all 3 issues found in the latest run.
- ✅ Resolved by another fix: Excessive discovery API calls on unrelated form changes
- Already fixed by upstream commit a65357f, which stabilizes handleDiscover via a discoveryRequestRef plus primitive-only deps (discoveryMode + serialized discoveryParamsKey) so unrelated form keystrokes no longer recreate the callback.
- ✅ Fixed: Stale closure captures outdated
savedAgentCardinhandleDiscover- Added savedAgentCard to handleDiscover's useCallback dependency array so 'Re-discover' refreshes the resetSelections closure whenever the parent passes in a new saved card.
- ✅ Resolved by another fix: Server-side request to arbitrary URLs without host validation
- Already fixed by upstream commit a65357f, which routes the well-known fetch through async_safe_get (validating each redirect against the SSRF blocklist for private/loopback/cloud-metadata addresses) and maps SSRFError to AgentCardDiscoveryError.
Preview (a3f958d8d5)
diff --git a/litellm/a2a_protocol/litellm_completion_bridge/handler.py b/litellm/a2a_protocol/litellm_completion_bridge/handler.py
--- a/litellm/a2a_protocol/litellm_completion_bridge/handler.py
+++ b/litellm/a2a_protocol/litellm_completion_bridge/handler.py
@@ -107,6 +107,14 @@
if k not in ("model", "custom_llm_provider") and k not in _AGENT_ONLY_PARAMS
}
completion_params.update(litellm_params_to_add)
+ # Apply forward metadata AFTER the litellm_params merge so an
+ # agent-configured ``extra_body`` does not overwrite the forwarded
+ # A2A metadata; the helper merges into any existing ``extra_body``.
+ A2ACompletionBridgeTransformation.apply_forward_metadata_to_completion_params(
+ completion_params=completion_params,
+ a2a_message=message,
+ params=params,
+ )
# Call litellm.acompletion
response = await litellm.acompletion(**completion_params)
@@ -214,6 +222,14 @@
if k not in ("model", "custom_llm_provider") and k not in _AGENT_ONLY_PARAMS
}
completion_params.update(litellm_params_to_add)
+ # Apply forward metadata AFTER the litellm_params merge so an
+ # agent-configured ``extra_body`` does not overwrite the forwarded
+ # A2A metadata; the helper merges into any existing ``extra_body``.
+ A2ACompletionBridgeTransformation.apply_forward_metadata_to_completion_params(
+ completion_params=completion_params,
+ a2a_message=message,
+ params=params,
+ )
# 1. Emit initial task event (kind: "task", status: "submitted")
task_event = A2ACompletionBridgeTransformation.create_task_event(ctx)
diff --git a/litellm/a2a_protocol/litellm_completion_bridge/transformation.py b/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
--- a/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
+++ b/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
@@ -46,9 +46,79 @@
"""
@staticmethod
+ def _extract_text_from_a2a_parts(parts: List[Dict[str, Any]]) -> str:
+ """Extract text from A2A parts (with or without explicit ``kind``)."""
+ content_parts: List[str] = []
+ for part in parts:
+ if not isinstance(part, dict):
+ continue
+ kind = part.get("kind")
+ text = part.get("text")
+ if text is None:
+ continue
+ if kind in (None, "", "text"):
+ content_parts.append(str(text))
+ return "\n".join(content_parts)
+
+ @staticmethod
+ def get_forward_metadata(
+ a2a_message: Dict[str, Any],
+ params: Optional[Dict[str, Any]] = None,
+ ) -> Optional[Dict[str, Any]]:
+ """
+ Merge A2A metadata from MessageSendParams and the message for downstream providers.
+
+ Forwarded once on the LangGraph run payload (``metadata``), not duplicated on
+ each input message — see ``apply_forward_metadata_to_completion_params``.
+ """
+ merged: Dict[str, Any] = {}
+ if params and isinstance(params.get("metadata"), dict):
+ merged.update(params["metadata"])
+ message_metadata = a2a_message.get("metadata")
+ if isinstance(message_metadata, dict):
+ merged.update(message_metadata)
+ return merged or None
+
+ @staticmethod
+ def apply_forward_metadata_to_completion_params(
+ completion_params: Dict[str, Any],
+ a2a_message: Dict[str, Any],
+ params: Optional[Dict[str, Any]] = None,
+ ) -> None:
+ """
+ Attach A2A metadata to completion kwargs for provider bridges (e.g. LangGraph).
+
+ Uses ``extra_body`` so we do not collide with LiteLLM's spend-log ``metadata`` kwarg.
+ """
+ forward_metadata = A2ACompletionBridgeTransformation.get_forward_metadata(
+ a2a_message=a2a_message,
+ params=params,
+ )
+ if not forward_metadata:
+ return
+
+ extra_body = completion_params.get("extra_body")
+ if not isinstance(extra_body, dict):
+ extra_body = {}
+ # Merge into any existing ``extra_body.metadata`` so an
+ # agent-configured ``extra_body: {metadata: {...}}`` is preserved;
+ # forwarded A2A metadata takes precedence on key conflicts.
+ existing_metadata = extra_body.get("metadata")
+ merged_metadata: Dict[str, Any] = (
+ {**existing_metadata} if isinstance(existing_metadata, dict) else {}
+ )
+ merged_metadata.update(forward_metadata)
+ extra_body = {**extra_body, "metadata": merged_metadata}
+ completion_params["extra_body"] = extra_body
+
+ verbose_logger.debug(
+ f"A2A -> completion forward metadata keys={list(forward_metadata.keys())}"
+ )
+
+ @staticmethod
def a2a_message_to_openai_messages(
a2a_message: Dict[str, Any],
- ) -> List[Dict[str, str]]:
+ ) -> List[Dict[str, Any]]:
"""
Transform an A2A message to OpenAI message format.
@@ -70,21 +140,20 @@
elif role == "system":
openai_role = "system"
- # Extract text content from parts
- content_parts = []
- for part in parts:
- kind = part.get("kind", "")
- if kind == "text":
- text = part.get("text", "")
- content_parts.append(text)
+ if not isinstance(parts, list):
+ parts = []
- content = "\n".join(content_parts) if content_parts else ""
+ content = A2ACompletionBridgeTransformation._extract_text_from_a2a_parts(parts)
+ # Do not attach A2A message.metadata here — the completion bridge forwards it
+ # once at run level via extra_body.metadata (LangGraph POST /runs/wait shape).
+ openai_message: Dict[str, Any] = {"role": openai_role, "content": content}
+
verbose_logger.debug(
f"A2A -> OpenAI transform: role={role} -> {openai_role}, content_length={len(content)}"
)
- return [{"role": openai_role, "content": content}]
+ return [openai_message]
@staticmethod
def openai_response_to_a2a_response(
@@ -110,6 +179,7 @@
# Build A2A message
a2a_message = {
+ "kind": "message",
"role": "agent",
"parts": [{"kind": "text", "text": content}],
"messageId": uuid4().hex,
@@ -119,9 +189,7 @@
a2a_response = {
"jsonrpc": "2.0",
"id": request_id,
- "result": {
- "message": a2a_message,
- },
+ "result": a2a_message,
}
verbose_logger.debug(f"OpenAI -> A2A transform: content_length={len(content)}")
@@ -267,16 +335,22 @@
if not content and not is_final:
return None
- # Build A2A streaming chunk (legacy format)
+ # Build A2A streaming chunk. Mirrors the non-streaming response
+ # shape (``result`` is the message itself, with ``kind: "message"``
+ # as the result-level event discriminator — matching how
+ # ``create_artifact_update_event`` uses ``kind: "artifact-update"``
+ # at the result level). ``final`` is an envelope-level streaming
+ # property per the A2A spec and is appended alongside the message
+ # fields so consumers can read a uniform ``result`` shape across
+ # streaming and non-streaming.
a2a_chunk = {
"jsonrpc": "2.0",
"id": request_id,
"result": {
- "message": {
- "role": "agent",
- "parts": [{"kind": "text", "text": content}],
- "messageId": uuid4().hex,
- },
+ "kind": "message",
+ "role": "agent",
+ "parts": [{"kind": "text", "text": content}],
+ "messageId": uuid4().hex,
"final": is_final,
},
}
diff --git a/litellm/a2a_protocol/providers/litellm_completion/README.md b/litellm/a2a_protocol/providers/litellm_completion/README.md
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/README.md
+++ /dev/null
@@ -1,74 +1,0 @@
-# A2A to LiteLLM Completion Bridge
-
-Routes A2A protocol requests through `litellm.acompletion`, enabling any LiteLLM-supported provider to be invoked via A2A.
-
-## Flow
-
-```
-A2A Request → Transform → litellm.acompletion → Transform → A2A Response
-```
-
-## SDK Usage
-
-Use the existing `asend_message` and `asend_message_streaming` functions with `litellm_params`:
-
-```python
-from litellm.a2a_protocol import asend_message, asend_message_streaming
-from a2a.types import SendMessageRequest, SendStreamingMessageRequest, MessageSendParams
-from uuid import uuid4
-
-# Non-streaming
-request = SendMessageRequest(
- id=str(uuid4()),
- params=MessageSendParams(
- message={"role": "user", "parts": [{"kind": "text", "text": "Hello!"}], "messageId": uuid4().hex}
- )
-)
-response = await asend_message(
- request=request,
- api_base="http://localhost:2024",
- litellm_params={"custom_llm_provider": "langgraph", "model": "agent"},
-)
-
-# Streaming
-stream_request = SendStreamingMessageRequest(
- id=str(uuid4()),
- params=MessageSendParams(
- message={"role": "user", "parts": [{"kind": "text", "text": "Hello!"}], "messageId": uuid4().hex}
- )
-)
-async for chunk in asend_message_streaming(
- request=stream_request,
- api_base="http://localhost:2024",
- litellm_params={"custom_llm_provider": "langgraph", "model": "agent"},
-):
- print(chunk)
-```
-
-## Proxy Usage
-
-Configure an agent with `custom_llm_provider` in `litellm_params`:
-
-```yaml
-agents:
- - agent_name: my-langgraph-agent
- agent_card_params:
- name: "LangGraph Agent"
- url: "http://localhost:2024" # Used as api_base
- litellm_params:
- custom_llm_provider: langgraph
- model: agent
-```
-
-When an A2A request hits `/a2a/{agent_id}/message/send`, the bridge:
-
-1. Detects `custom_llm_provider` in agent's `litellm_params`
-2. Transforms A2A message → OpenAI messages
-3. Calls `litellm.acompletion(model="langgraph/agent", api_base="http://localhost:2024")`
-4. Transforms response → A2A format
-
-## Classes
-
-- `A2ACompletionBridgeTransformation` - Static methods for message format conversion
-- `A2ACompletionBridgeHandler` - Static methods for handling requests (streaming/non-streaming)
-
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/__init__.py b/litellm/a2a_protocol/providers/litellm_completion/__init__.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/__init__.py
+++ /dev/null
@@ -1,5 +1,0 @@
-"""
-LiteLLM Completion bridge provider for A2A protocol.
-
-Routes A2A requests through litellm.acompletion based on custom_llm_provider.
-"""
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/handler.py b/litellm/a2a_protocol/providers/litellm_completion/handler.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/handler.py
+++ /dev/null
@@ -1,301 +1,0 @@
-"""
-Handler for A2A to LiteLLM completion bridge.
-
-Routes A2A requests through litellm.acompletion based on custom_llm_provider.
-
-A2A Streaming Events (in order):
-1. Task event (kind: "task") - Initial task creation with status "submitted"
-2. Status update (kind: "status-update") - Status change to "working"
-3. Artifact update (kind: "artifact-update") - Content/artifact delivery
-4. Status update (kind: "status-update") - Final status "completed" with final=true
-"""
-
-from typing import Any, AsyncIterator, Dict, Optional
-
-import litellm
-from litellm._logging import verbose_logger
-from litellm.a2a_protocol.litellm_completion_bridge.pydantic_ai_transformation import (
- PydanticAITransformation,
-)
-from litellm.a2a_protocol.litellm_completion_bridge.transformation import (
- A2ACompletionBridgeTransformation,
- A2AStreamingContext,
-)
-
-
-class A2ACompletionBridgeHandler:
- """
- Static methods for handling A2A requests via LiteLLM completion.
- """
-
- @staticmethod
- async def handle_non_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Handle non-streaming A2A request via litellm.acompletion.
-
- Args:
- request_id: A2A JSON-RPC request ID
- params: A2A MessageSendParams containing the message
- litellm_params: Agent's litellm_params (custom_llm_provider, model, etc.)
- api_base: API base URL from agent_card_params
-
- Returns:
- A2A SendMessageResponse dict
- """
- # Check if this is a Pydantic AI agent request
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- if custom_llm_provider == "pydantic_ai_agents":
- if api_base is None:
- raise ValueError("api_base is required for Pydantic AI agents")
-
- verbose_logger.info(
- f"Pydantic AI: Routing to Pydantic AI agent at {api_base}"
- )
-
- # Send request directly to Pydantic AI agent
- response_data = await PydanticAITransformation.send_non_streaming_request(
- api_base=api_base,
- request_id=request_id,
- params=params,
- )
-
- return response_data
-
- # Extract message from params
- message = params.get("message", {})
-
- # Transform A2A message to OpenAI format
- openai_messages = (
- A2ACompletionBridgeTransformation.a2a_message_to_openai_messages(message)
- )
-
- # Get completion params
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- model = litellm_params.get("model", "agent")
-
- # Build full model string if provider specified
- # Skip prepending if model already starts with the provider prefix
- if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/"):
- full_model = f"{custom_llm_provider}/{model}"
- else:
- full_model = model
-
- verbose_logger.info(
- f"A2A completion bridge: model={full_model}, api_base={api_base}"
- )
-
- # Build completion params dict
- completion_params = {
- "model": full_model,
- "messages": openai_messages,
- "api_base": api_base,
- "stream": False,
- }
- # Add litellm_params (contains api_key, client_id, client_secret, tenant_id, etc.)
- litellm_params_to_add = {
- k: v
- for k, v in litellm_params.items()
- if k not in ("model", "custom_llm_provider")
- }
- completion_params.update(litellm_params_to_add)
-
- # Call litellm.acompletion
- response = await litellm.acompletion(**completion_params)
-
- # Transform response to A2A format
- a2a_response = (
- A2ACompletionBridgeTransformation.openai_response_to_a2a_response(
- response=response,
- request_id=request_id,
- )
- )
-
- verbose_logger.info(f"A2A completion bridge completed: request_id={request_id}")
-
- return a2a_response
-
- @staticmethod
- async def handle_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
- ) -> AsyncIterator[Dict[str, Any]]:
- """
- Handle streaming A2A request via litellm.acompletion with stream=True.
-
- Emits proper A2A streaming events:
- 1. Task event (kind: "task") - Initial task with status "submitted"
- 2. Status update (kind: "status-update") - Status "working"
- 3. Artifact update (kind: "artifact-update") - Content delivery
- 4. Status update (kind: "status-update") - Final "completed" status
-
- Args:
- request_id: A2A JSON-RPC request ID
- params: A2A MessageSendParams containing the message
- litellm_params: Agent's litellm_params (custom_llm_provider, model, etc.)
- api_base: API base URL from agent_card_params
-
- Yields:
- A2A streaming response events
- """
- # Check if this is a Pydantic AI agent request
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- if custom_llm_provider == "pydantic_ai_agents":
- if api_base is None:
- raise ValueError("api_base is required for Pydantic AI agents")
-
- verbose_logger.info(
- f"Pydantic AI: Faking streaming for Pydantic AI agent at {api_base}"
- )
-
- # Get non-streaming response first
- response_data = await PydanticAITransformation.send_non_streaming_request(
- api_base=api_base,
- request_id=request_id,
- params=params,
- )
-
- # Convert to fake streaming
- async for chunk in PydanticAITransformation.fake_streaming_from_response(
- response_data=response_data,
- request_id=request_id,
- ):
- yield chunk
-
- return
-
- # Extract message from params
- message = params.get("message", {})
-
- # Create streaming context
- ctx = A2AStreamingContext(
- request_id=request_id,
- input_message=message,
- )
-
- # Transform A2A message to OpenAI format
- openai_messages = (
- A2ACompletionBridgeTransformation.a2a_message_to_openai_messages(message)
- )
-
- # Get completion params
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- model = litellm_params.get("model", "agent")
-
- # Build full model string if provider specified
- # Skip prepending if model already starts with the provider prefix
- if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/"):
- full_model = f"{custom_llm_provider}/{model}"
- else:
- full_model = model
-
- verbose_logger.info(
- f"A2A completion bridge streaming: model={full_model}, api_base={api_base}"
- )
-
- # Build completion params dict
- completion_params = {
- "model": full_model,
- "messages": openai_messages,
- "api_base": api_base,
- "stream": True,
- }
- # Add litellm_params (contains api_key, client_id, client_secret, tenant_id, etc.)
- litellm_params_to_add = {
- k: v
- for k, v in litellm_params.items()
- if k not in ("model", "custom_llm_provider")
- }
- completion_params.update(litellm_params_to_add)
-
- # 1. Emit initial task event (kind: "task", status: "submitted")
- task_event = A2ACompletionBridgeTransformation.create_task_event(ctx)
- yield task_event
-
- # 2. Emit status update (kind: "status-update", status: "working")
- working_event = A2ACompletionBridgeTransformation.create_status_update_event(
- ctx=ctx,
- state="working",
- final=False,
- message_text="Processing request...",
- )
- yield working_event
-
- # Call litellm.acompletion with streaming
- response = await litellm.acompletion(**completion_params)
-
- # 3. Accumulate content and emit artifact update
- accumulated_text = ""
- chunk_count = 0
- async for chunk in response: # type: ignore[union-attr]
- chunk_count += 1
-
- # Extract delta content
- content = ""
- if chunk is not None and hasattr(chunk, "choices") and chunk.choices:
- choice = chunk.choices[0]
- if hasattr(choice, "delta") and choice.delta:
- content = choice.delta.content or ""
-
- if content:
- accumulated_text += content
-
- # Emit artifact update with accumulated content
- if accumulated_text:
- artifact_event = (
- A2ACompletionBridgeTransformation.create_artifact_update_event(
- ctx=ctx,
- text=accumulated_text,
- )
- )
- yield artifact_event
-
- # 4. Emit final status update (kind: "status-update", status: "completed", final: true)
- completed_event = A2ACompletionBridgeTransformation.create_status_update_event(
- ctx=ctx,
- state="completed",
- final=True,
- )
- yield completed_event
-
- verbose_logger.info(
- f"A2A completion bridge streaming completed: request_id={request_id}, chunks={chunk_count}"
- )
-
-
-# Convenience functions that delegate to the class methods
-async def handle_a2a_completion(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
-) -> Dict[str, Any]:
- """Convenience function for non-streaming A2A completion."""
- return await A2ACompletionBridgeHandler.handle_non_streaming(
- request_id=request_id,
- params=params,
- litellm_params=litellm_params,
- api_base=api_base,
- )
-
-
-async def handle_a2a_completion_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
-) -> AsyncIterator[Dict[str, Any]]:
- """Convenience function for streaming A2A completion."""
- async for chunk in A2ACompletionBridgeHandler.handle_streaming(
- request_id=request_id,
- params=params,
- litellm_params=litellm_params,
- api_base=api_base,
- ):
- yield chunk
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/transformation.py b/litellm/a2a_protocol/providers/litellm_completion/transformation.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/transformation.py
+++ /dev/null
@@ -1,284 +1,0 @@
-"""
-Transformation utilities for A2A <-> OpenAI message format conversion.
-
-A2A Message Format:
-{
- "role": "user",
- "parts": [{"kind": "text", "text": "Hello!"}],
- "messageId": "abc123"
-}
-
-OpenAI Message Format:
-{"role": "user", "content": "Hello!"}
-
-A2A Streaming Events:
-- Task event (kind: "task") - Initial task creation with status "submitted"
-- Status update (kind: "status-update") - Status changes (working, completed)
-- Artifact update (kind: "artifact-update") - Content/artifact delivery
-"""
-
-from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional
-from uuid import uuid4
-
-from litellm._logging import verbose_logger
-
-
-class A2AStreamingContext:
- """
- Context holder for A2A streaming state.
- Tracks task_id, context_id, and message accumulation.
- """
-
- def __init__(self, request_id: str, input_message: Dict[str, Any]):
- self.request_id = request_id
- self.task_id = str(uuid4())
- self.context_id = str(uuid4())
- self.input_message = input_message
- self.accumulated_text = ""
- self.has_emitted_task = False
- self.has_emitted_working = False
-
-
-class A2ACompletionBridgeTransformation:
- """
- Static methods for transforming between A2A and OpenAI message formats.
- """
-
- @staticmethod
- def a2a_message_to_openai_messages(
- a2a_message: Dict[str, Any],
- ) -> List[Dict[str, str]]:
- """
- Transform an A2A message to OpenAI message format.
-
- Args:
- a2a_message: A2A message with role, parts, and messageId
-
- Returns:
- List of OpenAI-format messages
- """
- role = a2a_message.get("role", "user")
- parts = a2a_message.get("parts", [])
-
- # Map A2A roles to OpenAI roles
- openai_role = role
- if role == "user":
- openai_role = "user"
- elif role == "assistant":
- openai_role = "assistant"
- elif role == "system":
- openai_role = "system"
-
- # Extract text content from parts
- content_parts = []
- for part in parts:
- kind = part.get("kind", "")
- if kind == "text":
- text = part.get("text", "")
- content_parts.append(text)
-
- content = "\n".join(content_parts) if content_parts else ""
-
- verbose_logger.debug(
- f"A2A -> OpenAI transform: role={role} -> {openai_role}, content_length={len(content)}"
- )
-
- return [{"role": openai_role, "content": content}]
-
- @staticmethod
- def openai_response_to_a2a_response(
- response: Any,
- request_id: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Transform a LiteLLM ModelResponse to A2A SendMessageResponse format.
-
- Args:
- response: LiteLLM ModelResponse object
- request_id: Original A2A request ID
-
- Returns:
- A2A SendMessageResponse dict
- """
- # Extract content from response
- content = ""
- if hasattr(response, "choices") and response.choices:
- choice = response.choices[0]
- if hasattr(choice, "message") and choice.message:
- content = choice.message.content or ""
-
- # Build A2A message
- a2a_message = {
- "role": "agent",
- "parts": [{"kind": "text", "text": content}],
- "messageId": uuid4().hex,
- }
-
- # Build A2A response
- a2a_response = {
- "jsonrpc": "2.0",
- "id": request_id,
- "result": {
- "message": a2a_message,
- },
- }
-
- verbose_logger.debug(f"OpenAI -> A2A transform: content_length={len(content)}")
-
- return a2a_response
-
- @staticmethod
- def _get_timestamp() -> str:
- """Get current timestamp in ISO format with timezone."""
- return datetime.now(timezone.utc).isoformat()
-
- @staticmethod
- def create_task_event(
- ctx: A2AStreamingContext,
- ) -> Dict[str, Any]:
- """
- Create the initial task event with status 'submitted'.
-
- This is the first event emitted in an A2A streaming response.
- """
- return {
- "id": ctx.request_id,
- "jsonrpc": "2.0",
- "result": {
- "contextId": ctx.context_id,
- "history": [
- {
- "contextId": ctx.context_id,
- "kind": "message",
- "messageId": ctx.input_message.get("messageId", uuid4().hex),
- "parts": ctx.input_message.get("parts", []),
- "role": ctx.input_message.get("role", "user"),
- "taskId": ctx.task_id,
- }
- ],
- "id": ctx.task_id,
- "kind": "task",
- "status": {
- "state": "submitted",
- },
- },
- }
-
- @staticmethod
- def create_status_update_event(
- ctx: A2AStreamingContext,
- state: str,
- final: bool = False,
- message_text: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Create a status update event.
-
- Args:
- ctx: Streaming context
- state: Status state ('working', 'completed')
- final: Whether this is the final event
- message_text: Optional message text for 'working' status
- """
- status: Dict[str, Any] = {
- "state": state,
- "timestamp": A2ACompletionBridgeTransformation._get_timestamp(),
- }
-
- # Add message for 'working' status
- if state == "working" and message_text:
- status["message"] = {
- "contextId": ctx.context_id,
- "kind": "message",
- "messageId": str(uuid4()),
- "parts": [{"kind": "text", "text": message_text}],
- "role": "agent",
- "taskId": ctx.task_id,
... diff truncated: showing 800 of 3745 linesYou can send follow-ups to the cloud agent here.
1. UI: Stabilize discoveryRequest deps to avoid redundant /v1/a2a/discover API calls. The parent rebuilds the discoveryRequest object on every form keystroke, so depend on primitive proxies (discovery_mode + serialized params) rather than the object identity. Read the actual object via a ref inside handleDiscover. 2. Backend: Route the well-known card fetch through async_safe_get so the admin /v1/a2a/discover endpoint can't be used to probe private/loopback addresses or cloud metadata endpoints. SSRFError is a separate handled case so it surfaces a clear AgentCardDiscoveryError. 3. Streaming: Make openai_chunk_to_a2a_chunk emit the same flat result shape as the non-streaming response (kind/role/parts/messageId at the result level), with envelope-level 'final' added. Matches the existing create_artifact_update_event pattern and lets consumers read a uniform result shape across streaming and non-streaming. Co-authored-by: Yassin Kortam <[email protected]>
Collaborator
|
Bugbot Autofix prepared fixes for all 3 issues found in the latest run.
Preview (a65357f9bb)diff --git a/litellm/a2a_protocol/litellm_completion_bridge/handler.py b/litellm/a2a_protocol/litellm_completion_bridge/handler.py
--- a/litellm/a2a_protocol/litellm_completion_bridge/handler.py
+++ b/litellm/a2a_protocol/litellm_completion_bridge/handler.py
@@ -107,6 +107,14 @@
if k not in ("model", "custom_llm_provider") and k not in _AGENT_ONLY_PARAMS
}
completion_params.update(litellm_params_to_add)
+ # Apply forward metadata AFTER the litellm_params merge so an
+ # agent-configured ``extra_body`` does not overwrite the forwarded
+ # A2A metadata; the helper merges into any existing ``extra_body``.
+ A2ACompletionBridgeTransformation.apply_forward_metadata_to_completion_params(
+ completion_params=completion_params,
+ a2a_message=message,
+ params=params,
+ )
# Call litellm.acompletion
response = await litellm.acompletion(**completion_params)
@@ -214,6 +222,14 @@
if k not in ("model", "custom_llm_provider") and k not in _AGENT_ONLY_PARAMS
}
completion_params.update(litellm_params_to_add)
+ # Apply forward metadata AFTER the litellm_params merge so an
+ # agent-configured ``extra_body`` does not overwrite the forwarded
+ # A2A metadata; the helper merges into any existing ``extra_body``.
+ A2ACompletionBridgeTransformation.apply_forward_metadata_to_completion_params(
+ completion_params=completion_params,
+ a2a_message=message,
+ params=params,
+ )
# 1. Emit initial task event (kind: "task", status: "submitted")
task_event = A2ACompletionBridgeTransformation.create_task_event(ctx)
diff --git a/litellm/a2a_protocol/litellm_completion_bridge/transformation.py b/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
--- a/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
+++ b/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
@@ -46,9 +46,79 @@
"""
@staticmethod
+ def _extract_text_from_a2a_parts(parts: List[Dict[str, Any]]) -> str:
+ """Extract text from A2A parts (with or without explicit ``kind``)."""
+ content_parts: List[str] = []
+ for part in parts:
+ if not isinstance(part, dict):
+ continue
+ kind = part.get("kind")
+ text = part.get("text")
+ if text is None:
+ continue
+ if kind in (None, "", "text"):
+ content_parts.append(str(text))
+ return "\n".join(content_parts)
+
+ @staticmethod
+ def get_forward_metadata(
+ a2a_message: Dict[str, Any],
+ params: Optional[Dict[str, Any]] = None,
+ ) -> Optional[Dict[str, Any]]:
+ """
+ Merge A2A metadata from MessageSendParams and the message for downstream providers.
+
+ Forwarded once on the LangGraph run payload (``metadata``), not duplicated on
+ each input message — see ``apply_forward_metadata_to_completion_params``.
+ """
+ merged: Dict[str, Any] = {}
+ if params and isinstance(params.get("metadata"), dict):
+ merged.update(params["metadata"])
+ message_metadata = a2a_message.get("metadata")
+ if isinstance(message_metadata, dict):
+ merged.update(message_metadata)
+ return merged or None
+
+ @staticmethod
+ def apply_forward_metadata_to_completion_params(
+ completion_params: Dict[str, Any],
+ a2a_message: Dict[str, Any],
+ params: Optional[Dict[str, Any]] = None,
+ ) -> None:
+ """
+ Attach A2A metadata to completion kwargs for provider bridges (e.g. LangGraph).
+
+ Uses ``extra_body`` so we do not collide with LiteLLM's spend-log ``metadata`` kwarg.
+ """
+ forward_metadata = A2ACompletionBridgeTransformation.get_forward_metadata(
+ a2a_message=a2a_message,
+ params=params,
+ )
+ if not forward_metadata:
+ return
+
+ extra_body = completion_params.get("extra_body")
+ if not isinstance(extra_body, dict):
+ extra_body = {}
+ # Merge into any existing ``extra_body.metadata`` so an
+ # agent-configured ``extra_body: {metadata: {...}}`` is preserved;
+ # forwarded A2A metadata takes precedence on key conflicts.
+ existing_metadata = extra_body.get("metadata")
+ merged_metadata: Dict[str, Any] = (
+ {**existing_metadata} if isinstance(existing_metadata, dict) else {}
+ )
+ merged_metadata.update(forward_metadata)
+ extra_body = {**extra_body, "metadata": merged_metadata}
+ completion_params["extra_body"] = extra_body
+
+ verbose_logger.debug(
+ f"A2A -> completion forward metadata keys={list(forward_metadata.keys())}"
+ )
+
+ @staticmethod
def a2a_message_to_openai_messages(
a2a_message: Dict[str, Any],
- ) -> List[Dict[str, str]]:
+ ) -> List[Dict[str, Any]]:
"""
Transform an A2A message to OpenAI message format.
@@ -70,21 +140,20 @@
elif role == "system":
openai_role = "system"
- # Extract text content from parts
- content_parts = []
- for part in parts:
- kind = part.get("kind", "")
- if kind == "text":
- text = part.get("text", "")
- content_parts.append(text)
+ if not isinstance(parts, list):
+ parts = []
- content = "\n".join(content_parts) if content_parts else ""
+ content = A2ACompletionBridgeTransformation._extract_text_from_a2a_parts(parts)
+ # Do not attach A2A message.metadata here — the completion bridge forwards it
+ # once at run level via extra_body.metadata (LangGraph POST /runs/wait shape).
+ openai_message: Dict[str, Any] = {"role": openai_role, "content": content}
+
verbose_logger.debug(
f"A2A -> OpenAI transform: role={role} -> {openai_role}, content_length={len(content)}"
)
- return [{"role": openai_role, "content": content}]
+ return [openai_message]
@staticmethod
def openai_response_to_a2a_response(
@@ -110,6 +179,7 @@
# Build A2A message
a2a_message = {
+ "kind": "message",
"role": "agent",
"parts": [{"kind": "text", "text": content}],
"messageId": uuid4().hex,
@@ -119,9 +189,7 @@
a2a_response = {
"jsonrpc": "2.0",
"id": request_id,
- "result": {
- "message": a2a_message,
- },
+ "result": a2a_message,
}
verbose_logger.debug(f"OpenAI -> A2A transform: content_length={len(content)}")
@@ -267,16 +335,22 @@
if not content and not is_final:
return None
- # Build A2A streaming chunk (legacy format)
+ # Build A2A streaming chunk. Mirrors the non-streaming response
+ # shape (``result`` is the message itself, with ``kind: "message"``
+ # as the result-level event discriminator — matching how
+ # ``create_artifact_update_event`` uses ``kind: "artifact-update"``
+ # at the result level). ``final`` is an envelope-level streaming
+ # property per the A2A spec and is appended alongside the message
+ # fields so consumers can read a uniform ``result`` shape across
+ # streaming and non-streaming.
a2a_chunk = {
"jsonrpc": "2.0",
"id": request_id,
"result": {
- "message": {
- "role": "agent",
- "parts": [{"kind": "text", "text": content}],
- "messageId": uuid4().hex,
- },
+ "kind": "message",
+ "role": "agent",
+ "parts": [{"kind": "text", "text": content}],
+ "messageId": uuid4().hex,
"final": is_final,
},
}
diff --git a/litellm/a2a_protocol/providers/litellm_completion/README.md b/litellm/a2a_protocol/providers/litellm_completion/README.md
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/README.md
+++ /dev/null
@@ -1,74 +1,0 @@
-# A2A to LiteLLM Completion Bridge
-
-Routes A2A protocol requests through `litellm.acompletion`, enabling any LiteLLM-supported provider to be invoked via A2A.
-
-## Flow
-
-```
-A2A Request → Transform → litellm.acompletion → Transform → A2A Response
-```
-
-## SDK Usage
-
-Use the existing `asend_message` and `asend_message_streaming` functions with `litellm_params`:
-
-```python
-from litellm.a2a_protocol import asend_message, asend_message_streaming
-from a2a.types import SendMessageRequest, SendStreamingMessageRequest, MessageSendParams
-from uuid import uuid4
-
-# Non-streaming
-request = SendMessageRequest(
- id=str(uuid4()),
- params=MessageSendParams(
- message={"role": "user", "parts": [{"kind": "text", "text": "Hello!"}], "messageId": uuid4().hex}
- )
-)
-response = await asend_message(
- request=request,
- api_base="http://localhost:2024",
- litellm_params={"custom_llm_provider": "langgraph", "model": "agent"},
-)
-
-# Streaming
-stream_request = SendStreamingMessageRequest(
- id=str(uuid4()),
- params=MessageSendParams(
- message={"role": "user", "parts": [{"kind": "text", "text": "Hello!"}], "messageId": uuid4().hex}
- )
-)
-async for chunk in asend_message_streaming(
- request=stream_request,
- api_base="http://localhost:2024",
- litellm_params={"custom_llm_provider": "langgraph", "model": "agent"},
-):
- print(chunk)
-```
-
-## Proxy Usage
-
-Configure an agent with `custom_llm_provider` in `litellm_params`:
-
-```yaml
-agents:
- - agent_name: my-langgraph-agent
- agent_card_params:
- name: "LangGraph Agent"
- url: "http://localhost:2024" # Used as api_base
- litellm_params:
- custom_llm_provider: langgraph
- model: agent
-```
-
-When an A2A request hits `/a2a/{agent_id}/message/send`, the bridge:
-
-1. Detects `custom_llm_provider` in agent's `litellm_params`
-2. Transforms A2A message → OpenAI messages
-3. Calls `litellm.acompletion(model="langgraph/agent", api_base="http://localhost:2024")`
-4. Transforms response → A2A format
-
-## Classes
-
-- `A2ACompletionBridgeTransformation` - Static methods for message format conversion
-- `A2ACompletionBridgeHandler` - Static methods for handling requests (streaming/non-streaming)
-
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/__init__.py b/litellm/a2a_protocol/providers/litellm_completion/__init__.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/__init__.py
+++ /dev/null
@@ -1,5 +1,0 @@
-"""
-LiteLLM Completion bridge provider for A2A protocol.
-
-Routes A2A requests through litellm.acompletion based on custom_llm_provider.
-"""
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/handler.py b/litellm/a2a_protocol/providers/litellm_completion/handler.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/handler.py
+++ /dev/null
@@ -1,301 +1,0 @@
-"""
-Handler for A2A to LiteLLM completion bridge.
-
-Routes A2A requests through litellm.acompletion based on custom_llm_provider.
-
-A2A Streaming Events (in order):
-1. Task event (kind: "task") - Initial task creation with status "submitted"
-2. Status update (kind: "status-update") - Status change to "working"
-3. Artifact update (kind: "artifact-update") - Content/artifact delivery
-4. Status update (kind: "status-update") - Final status "completed" with final=true
-"""
-
-from typing import Any, AsyncIterator, Dict, Optional
-
-import litellm
-from litellm._logging import verbose_logger
-from litellm.a2a_protocol.litellm_completion_bridge.pydantic_ai_transformation import (
- PydanticAITransformation,
-)
-from litellm.a2a_protocol.litellm_completion_bridge.transformation import (
- A2ACompletionBridgeTransformation,
- A2AStreamingContext,
-)
-
-
-class A2ACompletionBridgeHandler:
- """
- Static methods for handling A2A requests via LiteLLM completion.
- """
-
- @staticmethod
- async def handle_non_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Handle non-streaming A2A request via litellm.acompletion.
-
- Args:
- request_id: A2A JSON-RPC request ID
- params: A2A MessageSendParams containing the message
- litellm_params: Agent's litellm_params (custom_llm_provider, model, etc.)
- api_base: API base URL from agent_card_params
-
- Returns:
- A2A SendMessageResponse dict
- """
- # Check if this is a Pydantic AI agent request
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- if custom_llm_provider == "pydantic_ai_agents":
- if api_base is None:
- raise ValueError("api_base is required for Pydantic AI agents")
-
- verbose_logger.info(
- f"Pydantic AI: Routing to Pydantic AI agent at {api_base}"
- )
-
- # Send request directly to Pydantic AI agent
- response_data = await PydanticAITransformation.send_non_streaming_request(
- api_base=api_base,
- request_id=request_id,
- params=params,
- )
-
- return response_data
-
- # Extract message from params
- message = params.get("message", {})
-
- # Transform A2A message to OpenAI format
- openai_messages = (
- A2ACompletionBridgeTransformation.a2a_message_to_openai_messages(message)
- )
-
- # Get completion params
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- model = litellm_params.get("model", "agent")
-
- # Build full model string if provider specified
- # Skip prepending if model already starts with the provider prefix
- if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/"):
- full_model = f"{custom_llm_provider}/{model}"
- else:
- full_model = model
-
- verbose_logger.info(
- f"A2A completion bridge: model={full_model}, api_base={api_base}"
- )
-
- # Build completion params dict
- completion_params = {
- "model": full_model,
- "messages": openai_messages,
- "api_base": api_base,
- "stream": False,
- }
- # Add litellm_params (contains api_key, client_id, client_secret, tenant_id, etc.)
- litellm_params_to_add = {
- k: v
- for k, v in litellm_params.items()
- if k not in ("model", "custom_llm_provider")
- }
- completion_params.update(litellm_params_to_add)
-
- # Call litellm.acompletion
- response = await litellm.acompletion(**completion_params)
-
- # Transform response to A2A format
- a2a_response = (
- A2ACompletionBridgeTransformation.openai_response_to_a2a_response(
- response=response,
- request_id=request_id,
- )
- )
-
- verbose_logger.info(f"A2A completion bridge completed: request_id={request_id}")
-
- return a2a_response
-
- @staticmethod
- async def handle_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
- ) -> AsyncIterator[Dict[str, Any]]:
- """
- Handle streaming A2A request via litellm.acompletion with stream=True.
-
- Emits proper A2A streaming events:
- 1. Task event (kind: "task") - Initial task with status "submitted"
- 2. Status update (kind: "status-update") - Status "working"
- 3. Artifact update (kind: "artifact-update") - Content delivery
- 4. Status update (kind: "status-update") - Final "completed" status
-
- Args:
- request_id: A2A JSON-RPC request ID
- params: A2A MessageSendParams containing the message
- litellm_params: Agent's litellm_params (custom_llm_provider, model, etc.)
- api_base: API base URL from agent_card_params
-
- Yields:
- A2A streaming response events
- """
- # Check if this is a Pydantic AI agent request
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- if custom_llm_provider == "pydantic_ai_agents":
- if api_base is None:
- raise ValueError("api_base is required for Pydantic AI agents")
-
- verbose_logger.info(
- f"Pydantic AI: Faking streaming for Pydantic AI agent at {api_base}"
- )
-
- # Get non-streaming response first
- response_data = await PydanticAITransformation.send_non_streaming_request(
- api_base=api_base,
- request_id=request_id,
- params=params,
- )
-
- # Convert to fake streaming
- async for chunk in PydanticAITransformation.fake_streaming_from_response(
- response_data=response_data,
- request_id=request_id,
- ):
- yield chunk
-
- return
-
- # Extract message from params
- message = params.get("message", {})
-
- # Create streaming context
- ctx = A2AStreamingContext(
- request_id=request_id,
- input_message=message,
- )
-
- # Transform A2A message to OpenAI format
- openai_messages = (
- A2ACompletionBridgeTransformation.a2a_message_to_openai_messages(message)
- )
-
- # Get completion params
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- model = litellm_params.get("model", "agent")
-
- # Build full model string if provider specified
- # Skip prepending if model already starts with the provider prefix
- if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/"):
- full_model = f"{custom_llm_provider}/{model}"
- else:
- full_model = model
-
- verbose_logger.info(
- f"A2A completion bridge streaming: model={full_model}, api_base={api_base}"
- )
-
- # Build completion params dict
- completion_params = {
- "model": full_model,
- "messages": openai_messages,
- "api_base": api_base,
- "stream": True,
- }
- # Add litellm_params (contains api_key, client_id, client_secret, tenant_id, etc.)
- litellm_params_to_add = {
- k: v
- for k, v in litellm_params.items()
- if k not in ("model", "custom_llm_provider")
- }
- completion_params.update(litellm_params_to_add)
-
- # 1. Emit initial task event (kind: "task", status: "submitted")
- task_event = A2ACompletionBridgeTransformation.create_task_event(ctx)
- yield task_event
-
- # 2. Emit status update (kind: "status-update", status: "working")
- working_event = A2ACompletionBridgeTransformation.create_status_update_event(
- ctx=ctx,
- state="working",
- final=False,
- message_text="Processing request...",
- )
- yield working_event
-
- # Call litellm.acompletion with streaming
- response = await litellm.acompletion(**completion_params)
-
- # 3. Accumulate content and emit artifact update
- accumulated_text = ""
- chunk_count = 0
- async for chunk in response: # type: ignore[union-attr]
- chunk_count += 1
-
- # Extract delta content
- content = ""
- if chunk is not None and hasattr(chunk, "choices") and chunk.choices:
- choice = chunk.choices[0]
- if hasattr(choice, "delta") and choice.delta:
- content = choice.delta.content or ""
-
- if content:
- accumulated_text += content
-
- # Emit artifact update with accumulated content
- if accumulated_text:
- artifact_event = (
- A2ACompletionBridgeTransformation.create_artifact_update_event(
- ctx=ctx,
- text=accumulated_text,
- )
- )
- yield artifact_event
-
- # 4. Emit final status update (kind: "status-update", status: "completed", final: true)
- completed_event = A2ACompletionBridgeTransformation.create_status_update_event(
- ctx=ctx,
- state="completed",
- final=True,
- )
- yield completed_event
-
- verbose_logger.info(
- f"A2A completion bridge streaming completed: request_id={request_id}, chunks={chunk_count}"
- )
-
-
-# Convenience functions that delegate to the class methods
-async def handle_a2a_completion(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
-) -> Dict[str, Any]:
- """Convenience function for non-streaming A2A completion."""
- return await A2ACompletionBridgeHandler.handle_non_streaming(
- request_id=request_id,
- params=params,
- litellm_params=litellm_params,
- api_base=api_base,
- )
-
-
-async def handle_a2a_completion_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
-) -> AsyncIterator[Dict[str, Any]]:
- """Convenience function for streaming A2A completion."""
- async for chunk in A2ACompletionBridgeHandler.handle_streaming(
- request_id=request_id,
- params=params,
- litellm_params=litellm_params,
- api_base=api_base,
- ):
- yield chunk
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/transformation.py b/litellm/a2a_protocol/providers/litellm_completion/transformation.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/transformation.py
+++ /dev/null
@@ -1,284 +1,0 @@
-"""
-Transformation utilities for A2A <-> OpenAI message format conversion.
-
-A2A Message Format:
-{
- "role": "user",
- "parts": [{"kind": "text", "text": "Hello!"}],
- "messageId": "abc123"
-}
-
-OpenAI Message Format:
-{"role": "user", "content": "Hello!"}
-
-A2A Streaming Events:
-- Task event (kind: "task") - Initial task creation with status "submitted"
-- Status update (kind: "status-update") - Status changes (working, completed)
-- Artifact update (kind: "artifact-update") - Content/artifact delivery
-"""
-
-from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional
-from uuid import uuid4
-
-from litellm._logging import verbose_logger
-
-
-class A2AStreamingContext:
- """
- Context holder for A2A streaming state.
- Tracks task_id, context_id, and message accumulation.
- """
-
- def __init__(self, request_id: str, input_message: Dict[str, Any]):
- self.request_id = request_id
- self.task_id = str(uuid4())
- self.context_id = str(uuid4())
- self.input_message = input_message
- self.accumulated_text = ""
- self.has_emitted_task = False
- self.has_emitted_working = False
-
-
-class A2ACompletionBridgeTransformation:
- """
- Static methods for transforming between A2A and OpenAI message formats.
- """
-
- @staticmethod
- def a2a_message_to_openai_messages(
- a2a_message: Dict[str, Any],
- ) -> List[Dict[str, str]]:
- """
- Transform an A2A message to OpenAI message format.
-
- Args:
- a2a_message: A2A message with role, parts, and messageId
-
- Returns:
- List of OpenAI-format messages
- """
- role = a2a_message.get("role", "user")
- parts = a2a_message.get("parts", [])
-
- # Map A2A roles to OpenAI roles
- openai_role = role
- if role == "user":
- openai_role = "user"
- elif role == "assistant":
- openai_role = "assistant"
- elif role == "system":
- openai_role = "system"
-
- # Extract text content from parts
- content_parts = []
- for part in parts:
- kind = part.get("kind", "")
- if kind == "text":
- text = part.get("text", "")
- content_parts.append(text)
-
- content = "\n".join(content_parts) if content_parts else ""
-
- verbose_logger.debug(
- f"A2A -> OpenAI transform: role={role} -> {openai_role}, content_length={len(content)}"
- )
-
- return [{"role": openai_role, "content": content}]
-
- @staticmethod
- def openai_response_to_a2a_response(
- response: Any,
- request_id: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Transform a LiteLLM ModelResponse to A2A SendMessageResponse format.
-
- Args:
- response: LiteLLM ModelResponse object
- request_id: Original A2A request ID
-
- Returns:
- A2A SendMessageResponse dict
- """
- # Extract content from response
- content = ""
- if hasattr(response, "choices") and response.choices:
- choice = response.choices[0]
- if hasattr(choice, "message") and choice.message:
- content = choice.message.content or ""
-
- # Build A2A message
- a2a_message = {
- "role": "agent",
- "parts": [{"kind": "text", "text": content}],
- "messageId": uuid4().hex,
- }
-
- # Build A2A response
- a2a_response = {
- "jsonrpc": "2.0",
- "id": request_id,
- "result": {
- "message": a2a_message,
- },
- }
-
- verbose_logger.debug(f"OpenAI -> A2A transform: content_length={len(content)}")
-
- return a2a_response
-
- @staticmethod
- def _get_timestamp() -> str:
- """Get current timestamp in ISO format with timezone."""
- return datetime.now(timezone.utc).isoformat()
-
- @staticmethod
- def create_task_event(
- ctx: A2AStreamingContext,
- ) -> Dict[str, Any]:
- """
- Create the initial task event with status 'submitted'.
-
- This is the first event emitted in an A2A streaming response.
- """
- return {
- "id": ctx.request_id,
- "jsonrpc": "2.0",
- "result": {
- "contextId": ctx.context_id,
- "history": [
- {
- "contextId": ctx.context_id,
- "kind": "message",
- "messageId": ctx.input_message.get("messageId", uuid4().hex),
- "parts": ctx.input_message.get("parts", []),
- "role": ctx.input_message.get("role", "user"),
- "taskId": ctx.task_id,
- }
- ],
- "id": ctx.task_id,
- "kind": "task",
- "status": {
- "state": "submitted",
- },
- },
- }
-
- @staticmethod
- def create_status_update_event(
- ctx: A2AStreamingContext,
- state: str,
- final: bool = False,
- message_text: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Create a status update event.
-
- Args:
- ctx: Streaming context
- state: Status state ('working', 'completed')
- final: Whether this is the final event
- message_text: Optional message text for 'working' status
- """
- status: Dict[str, Any] = {
- "state": state,
- "timestamp": A2ACompletionBridgeTransformation._get_timestamp(),
- }
-
- # Add message for 'working' status
- if state == "working" and message_text:
- status["message"] = {
- "contextId": ctx.context_id,
- "kind": "message",
- "messageId": str(uuid4()),
- "parts": [{"kind": "text", "text": message_text}],
- "role": "agent",
- "taskId": ctx.task_id,
... diff truncated: showing 800 of 3735 linesYou can send follow-ups to the cloud agent here. |
…ic agents Mirror the edit-form overlay in agent_info.tsx so dynamic agent types (e.g. LangGraph) whose forms don't register name/description as Form.Items don't silently lose those discovery-panel edits on save. Co-authored-by: Yassin Kortam <[email protected]>
Collaborator
Collaborator
|
bugbot run |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 4 potential issues.
Autofix Details
Bugbot Autofix prepared fixes for all 4 issues found in the latest run.
- ✅ Fixed: Agent card name edit silently overwritten by agent_name
- Changed
_build_merged_agent_cardto prefer the card-suppliedname(set by the discovery UI's editable input viaoverlayDiscoveredCardParams) and only fall back toagent_namewhen the card has no name of its own.
- Changed
- ✅ Resolved by another fix: Duplicated overlay logic across create and edit forms
- Already resolved on the base branch by commit 1a9bce3 which extracted
overlayDiscoveredCardParamsintoagent_discovery_utils.tsand replaced the inline duplicates inadd_agent_form.tsxandagent_info.tsx.
- Already resolved on the base branch by commit 1a9bce3 which extracted
- ✅ Fixed: Discovery crashes in production with URL validation enabled
- Changed the
async_safe_getcall site infetch_well_known_cardto passheaders=headers or {}so the validation path's{**caller_headers, "Host": ...}spread doesn't crash when no headers are supplied.
- Changed the
- ✅ Fixed: Edit-flow discovery omits
urlfield from form update- Added
url: selection.upstream_urltofieldsToSetinagent_info.tsx'shandleApplyDiscoveredCardso re-discovering against a different upstream refreshes the A2A form's URL input, matching the create flow.
- Added
Preview (65484a831c)
diff --git a/.github/workflows/test-unit-proxy-endpoints.yml b/.github/workflows/test-unit-proxy-endpoints.yml
--- a/.github/workflows/test-unit-proxy-endpoints.yml
+++ b/.github/workflows/test-unit-proxy-endpoints.yml
@@ -33,6 +33,7 @@
tests/test_litellm/proxy/image_endpoints
tests/test_litellm/proxy/vector_store_endpoints
tests/test_litellm/proxy/agent_endpoints
+ tests/test_litellm/proxy/a2a
tests/test_litellm/proxy/discovery_endpoints
tests/test_litellm/proxy/health_endpoints
tests/test_litellm/proxy/public_endpoints
diff --git a/litellm/a2a_protocol/litellm_completion_bridge/handler.py b/litellm/a2a_protocol/litellm_completion_bridge/handler.py
--- a/litellm/a2a_protocol/litellm_completion_bridge/handler.py
+++ b/litellm/a2a_protocol/litellm_completion_bridge/handler.py
@@ -107,6 +107,14 @@
if k not in ("model", "custom_llm_provider") and k not in _AGENT_ONLY_PARAMS
}
completion_params.update(litellm_params_to_add)
+ # Apply forward metadata AFTER the litellm_params merge so the helper
+ # sees any agent-owner-configured ``extra_body.metadata`` and can keep
+ # those keys authoritative over the client-supplied A2A metadata.
+ A2ACompletionBridgeTransformation.apply_forward_metadata_to_completion_params(
+ completion_params=completion_params,
+ a2a_message=message,
+ params=params,
+ )
# Call litellm.acompletion
response = await litellm.acompletion(**completion_params)
@@ -214,6 +222,14 @@
if k not in ("model", "custom_llm_provider") and k not in _AGENT_ONLY_PARAMS
}
completion_params.update(litellm_params_to_add)
+ # Apply forward metadata AFTER the litellm_params merge so the helper
+ # sees any agent-owner-configured ``extra_body.metadata`` and can keep
+ # those keys authoritative over the client-supplied A2A metadata.
+ A2ACompletionBridgeTransformation.apply_forward_metadata_to_completion_params(
+ completion_params=completion_params,
+ a2a_message=message,
+ params=params,
+ )
# 1. Emit initial task event (kind: "task", status: "submitted")
task_event = A2ACompletionBridgeTransformation.create_task_event(ctx)
diff --git a/litellm/a2a_protocol/litellm_completion_bridge/transformation.py b/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
--- a/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
+++ b/litellm/a2a_protocol/litellm_completion_bridge/transformation.py
@@ -46,9 +46,79 @@
"""
@staticmethod
+ def _extract_text_from_a2a_parts(parts: List[Dict[str, Any]]) -> str:
+ """Extract text from A2A parts (with or without explicit ``kind``)."""
+ content_parts: List[str] = []
+ for part in parts:
+ if not isinstance(part, dict):
+ continue
+ kind = part.get("kind")
+ text = part.get("text")
+ if text is None:
+ continue
+ if kind in (None, "", "text"):
+ content_parts.append(str(text))
+ return "\n".join(content_parts)
+
+ @staticmethod
+ def get_forward_metadata(
+ a2a_message: Dict[str, Any],
+ params: Optional[Dict[str, Any]] = None,
+ ) -> Optional[Dict[str, Any]]:
+ """
+ Merge A2A metadata from MessageSendParams and the message for downstream providers.
+
+ Forwarded once on the LangGraph run payload (``metadata``), not duplicated on
+ each input message — see ``apply_forward_metadata_to_completion_params``.
+ """
+ merged: Dict[str, Any] = {}
+ if params and isinstance(params.get("metadata"), dict):
+ merged.update(params["metadata"])
+ message_metadata = a2a_message.get("metadata")
+ if isinstance(message_metadata, dict):
+ merged.update(message_metadata)
+ return merged or None
+
+ @staticmethod
+ def apply_forward_metadata_to_completion_params(
+ completion_params: Dict[str, Any],
+ a2a_message: Dict[str, Any],
+ params: Optional[Dict[str, Any]] = None,
+ ) -> None:
+ """
+ Attach A2A metadata to completion kwargs for provider bridges (e.g. LangGraph).
+
+ Uses ``extra_body`` so we do not collide with LiteLLM's spend-log ``metadata`` kwarg.
+ """
+ forward_metadata = A2ACompletionBridgeTransformation.get_forward_metadata(
+ a2a_message=a2a_message,
+ params=params,
+ )
+ if not forward_metadata:
+ return
+
+ extra_body = completion_params.get("extra_body")
+ if not isinstance(extra_body, dict):
+ extra_body = {}
+ # Layer client-supplied A2A metadata under any agent-owner-configured
+ # ``extra_body.metadata`` so the configured keys remain authoritative
+ # and an A2A caller cannot overwrite server-set run metadata.
+ existing_metadata = extra_body.get("metadata")
+ existing_dict: Dict[str, Any] = (
+ existing_metadata if isinstance(existing_metadata, dict) else {}
+ )
+ merged_metadata: Dict[str, Any] = {**forward_metadata, **existing_dict}
+ extra_body = {**extra_body, "metadata": merged_metadata}
+ completion_params["extra_body"] = extra_body
+
+ verbose_logger.debug(
+ f"A2A -> completion forward metadata keys={list(forward_metadata.keys())}"
+ )
+
+ @staticmethod
def a2a_message_to_openai_messages(
a2a_message: Dict[str, Any],
- ) -> List[Dict[str, str]]:
+ ) -> List[Dict[str, Any]]:
"""
Transform an A2A message to OpenAI message format.
@@ -70,21 +140,20 @@
elif role == "system":
openai_role = "system"
- # Extract text content from parts
- content_parts = []
- for part in parts:
- kind = part.get("kind", "")
- if kind == "text":
- text = part.get("text", "")
- content_parts.append(text)
+ if not isinstance(parts, list):
+ parts = []
- content = "\n".join(content_parts) if content_parts else ""
+ content = A2ACompletionBridgeTransformation._extract_text_from_a2a_parts(parts)
+ # Do not attach A2A message.metadata here — the completion bridge forwards it
+ # once at run level via extra_body.metadata (LangGraph POST /runs/wait shape).
+ openai_message: Dict[str, Any] = {"role": openai_role, "content": content}
+
verbose_logger.debug(
f"A2A -> OpenAI transform: role={role} -> {openai_role}, content_length={len(content)}"
)
- return [{"role": openai_role, "content": content}]
+ return [openai_message]
@staticmethod
def openai_response_to_a2a_response(
@@ -110,6 +179,7 @@
# Build A2A message
a2a_message = {
+ "kind": "message",
"role": "agent",
"parts": [{"kind": "text", "text": content}],
"messageId": uuid4().hex,
@@ -119,9 +189,7 @@
a2a_response = {
"jsonrpc": "2.0",
"id": request_id,
- "result": {
- "message": a2a_message,
- },
+ "result": a2a_message,
}
verbose_logger.debug(f"OpenAI -> A2A transform: content_length={len(content)}")
@@ -235,50 +303,3 @@
"taskId": ctx.task_id,
},
}
-
- @staticmethod
- def openai_chunk_to_a2a_chunk(
- chunk: Any,
- request_id: Optional[str] = None,
- is_final: bool = False,
- ) -> Optional[Dict[str, Any]]:
- """
- Transform a LiteLLM streaming chunk to A2A streaming format.
-
- NOTE: This method is deprecated for streaming. Use the event-based
- methods (create_task_event, create_status_update_event,
- create_artifact_update_event) instead for proper A2A streaming.
-
- Args:
- chunk: LiteLLM ModelResponse chunk
- request_id: Original A2A request ID
- is_final: Whether this is the final chunk
-
- Returns:
- A2A streaming chunk dict or None if no content
- """
- # Extract delta content
- content = ""
- if chunk is not None and hasattr(chunk, "choices") and chunk.choices:
- choice = chunk.choices[0]
- if hasattr(choice, "delta") and choice.delta:
- content = choice.delta.content or ""
-
- if not content and not is_final:
- return None
-
- # Build A2A streaming chunk (legacy format)
- a2a_chunk = {
- "jsonrpc": "2.0",
- "id": request_id,
- "result": {
- "message": {
- "role": "agent",
- "parts": [{"kind": "text", "text": content}],
- "messageId": uuid4().hex,
- },
- "final": is_final,
- },
- }
-
- return a2a_chunk
diff --git a/litellm/a2a_protocol/providers/litellm_completion/README.md b/litellm/a2a_protocol/providers/litellm_completion/README.md
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/README.md
+++ /dev/null
@@ -1,74 +1,0 @@
-# A2A to LiteLLM Completion Bridge
-
-Routes A2A protocol requests through `litellm.acompletion`, enabling any LiteLLM-supported provider to be invoked via A2A.
-
-## Flow
-
-```
-A2A Request → Transform → litellm.acompletion → Transform → A2A Response
-```
-
-## SDK Usage
-
-Use the existing `asend_message` and `asend_message_streaming` functions with `litellm_params`:
-
-```python
-from litellm.a2a_protocol import asend_message, asend_message_streaming
-from a2a.types import SendMessageRequest, SendStreamingMessageRequest, MessageSendParams
-from uuid import uuid4
-
-# Non-streaming
-request = SendMessageRequest(
- id=str(uuid4()),
- params=MessageSendParams(
- message={"role": "user", "parts": [{"kind": "text", "text": "Hello!"}], "messageId": uuid4().hex}
- )
-)
-response = await asend_message(
- request=request,
- api_base="http://localhost:2024",
- litellm_params={"custom_llm_provider": "langgraph", "model": "agent"},
-)
-
-# Streaming
-stream_request = SendStreamingMessageRequest(
- id=str(uuid4()),
- params=MessageSendParams(
- message={"role": "user", "parts": [{"kind": "text", "text": "Hello!"}], "messageId": uuid4().hex}
- )
-)
-async for chunk in asend_message_streaming(
- request=stream_request,
- api_base="http://localhost:2024",
- litellm_params={"custom_llm_provider": "langgraph", "model": "agent"},
-):
- print(chunk)
-```
-
-## Proxy Usage
-
-Configure an agent with `custom_llm_provider` in `litellm_params`:
-
-```yaml
-agents:
- - agent_name: my-langgraph-agent
- agent_card_params:
- name: "LangGraph Agent"
- url: "http://localhost:2024" # Used as api_base
- litellm_params:
- custom_llm_provider: langgraph
- model: agent
-```
-
-When an A2A request hits `/a2a/{agent_id}/message/send`, the bridge:
-
-1. Detects `custom_llm_provider` in agent's `litellm_params`
-2. Transforms A2A message → OpenAI messages
-3. Calls `litellm.acompletion(model="langgraph/agent", api_base="http://localhost:2024")`
-4. Transforms response → A2A format
-
-## Classes
-
-- `A2ACompletionBridgeTransformation` - Static methods for message format conversion
-- `A2ACompletionBridgeHandler` - Static methods for handling requests (streaming/non-streaming)
-
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/__init__.py b/litellm/a2a_protocol/providers/litellm_completion/__init__.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/__init__.py
+++ /dev/null
@@ -1,5 +1,0 @@
-"""
-LiteLLM Completion bridge provider for A2A protocol.
-
-Routes A2A requests through litellm.acompletion based on custom_llm_provider.
-"""
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/handler.py b/litellm/a2a_protocol/providers/litellm_completion/handler.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/handler.py
+++ /dev/null
@@ -1,301 +1,0 @@
-"""
-Handler for A2A to LiteLLM completion bridge.
-
-Routes A2A requests through litellm.acompletion based on custom_llm_provider.
-
-A2A Streaming Events (in order):
-1. Task event (kind: "task") - Initial task creation with status "submitted"
-2. Status update (kind: "status-update") - Status change to "working"
-3. Artifact update (kind: "artifact-update") - Content/artifact delivery
-4. Status update (kind: "status-update") - Final status "completed" with final=true
-"""
-
-from typing import Any, AsyncIterator, Dict, Optional
-
-import litellm
-from litellm._logging import verbose_logger
-from litellm.a2a_protocol.litellm_completion_bridge.pydantic_ai_transformation import (
- PydanticAITransformation,
-)
-from litellm.a2a_protocol.litellm_completion_bridge.transformation import (
- A2ACompletionBridgeTransformation,
- A2AStreamingContext,
-)
-
-
-class A2ACompletionBridgeHandler:
- """
- Static methods for handling A2A requests via LiteLLM completion.
- """
-
- @staticmethod
- async def handle_non_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Handle non-streaming A2A request via litellm.acompletion.
-
- Args:
- request_id: A2A JSON-RPC request ID
- params: A2A MessageSendParams containing the message
- litellm_params: Agent's litellm_params (custom_llm_provider, model, etc.)
- api_base: API base URL from agent_card_params
-
- Returns:
- A2A SendMessageResponse dict
- """
- # Check if this is a Pydantic AI agent request
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- if custom_llm_provider == "pydantic_ai_agents":
- if api_base is None:
- raise ValueError("api_base is required for Pydantic AI agents")
-
- verbose_logger.info(
- f"Pydantic AI: Routing to Pydantic AI agent at {api_base}"
- )
-
- # Send request directly to Pydantic AI agent
- response_data = await PydanticAITransformation.send_non_streaming_request(
- api_base=api_base,
- request_id=request_id,
- params=params,
- )
-
- return response_data
-
- # Extract message from params
- message = params.get("message", {})
-
- # Transform A2A message to OpenAI format
- openai_messages = (
- A2ACompletionBridgeTransformation.a2a_message_to_openai_messages(message)
- )
-
- # Get completion params
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- model = litellm_params.get("model", "agent")
-
- # Build full model string if provider specified
- # Skip prepending if model already starts with the provider prefix
- if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/"):
- full_model = f"{custom_llm_provider}/{model}"
- else:
- full_model = model
-
- verbose_logger.info(
- f"A2A completion bridge: model={full_model}, api_base={api_base}"
- )
-
- # Build completion params dict
- completion_params = {
- "model": full_model,
- "messages": openai_messages,
- "api_base": api_base,
- "stream": False,
- }
- # Add litellm_params (contains api_key, client_id, client_secret, tenant_id, etc.)
- litellm_params_to_add = {
- k: v
- for k, v in litellm_params.items()
- if k not in ("model", "custom_llm_provider")
- }
- completion_params.update(litellm_params_to_add)
-
- # Call litellm.acompletion
- response = await litellm.acompletion(**completion_params)
-
- # Transform response to A2A format
- a2a_response = (
- A2ACompletionBridgeTransformation.openai_response_to_a2a_response(
- response=response,
- request_id=request_id,
- )
- )
-
- verbose_logger.info(f"A2A completion bridge completed: request_id={request_id}")
-
- return a2a_response
-
- @staticmethod
- async def handle_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
- ) -> AsyncIterator[Dict[str, Any]]:
- """
- Handle streaming A2A request via litellm.acompletion with stream=True.
-
- Emits proper A2A streaming events:
- 1. Task event (kind: "task") - Initial task with status "submitted"
- 2. Status update (kind: "status-update") - Status "working"
- 3. Artifact update (kind: "artifact-update") - Content delivery
- 4. Status update (kind: "status-update") - Final "completed" status
-
- Args:
- request_id: A2A JSON-RPC request ID
- params: A2A MessageSendParams containing the message
- litellm_params: Agent's litellm_params (custom_llm_provider, model, etc.)
- api_base: API base URL from agent_card_params
-
- Yields:
- A2A streaming response events
- """
- # Check if this is a Pydantic AI agent request
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- if custom_llm_provider == "pydantic_ai_agents":
- if api_base is None:
- raise ValueError("api_base is required for Pydantic AI agents")
-
- verbose_logger.info(
- f"Pydantic AI: Faking streaming for Pydantic AI agent at {api_base}"
- )
-
- # Get non-streaming response first
- response_data = await PydanticAITransformation.send_non_streaming_request(
- api_base=api_base,
- request_id=request_id,
- params=params,
- )
-
- # Convert to fake streaming
- async for chunk in PydanticAITransformation.fake_streaming_from_response(
- response_data=response_data,
- request_id=request_id,
- ):
- yield chunk
-
- return
-
- # Extract message from params
- message = params.get("message", {})
-
- # Create streaming context
- ctx = A2AStreamingContext(
- request_id=request_id,
- input_message=message,
- )
-
- # Transform A2A message to OpenAI format
- openai_messages = (
- A2ACompletionBridgeTransformation.a2a_message_to_openai_messages(message)
- )
-
- # Get completion params
- custom_llm_provider = litellm_params.get("custom_llm_provider")
- model = litellm_params.get("model", "agent")
-
- # Build full model string if provider specified
- # Skip prepending if model already starts with the provider prefix
- if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/"):
- full_model = f"{custom_llm_provider}/{model}"
- else:
- full_model = model
-
- verbose_logger.info(
- f"A2A completion bridge streaming: model={full_model}, api_base={api_base}"
- )
-
- # Build completion params dict
- completion_params = {
- "model": full_model,
- "messages": openai_messages,
- "api_base": api_base,
- "stream": True,
- }
- # Add litellm_params (contains api_key, client_id, client_secret, tenant_id, etc.)
- litellm_params_to_add = {
- k: v
- for k, v in litellm_params.items()
- if k not in ("model", "custom_llm_provider")
- }
- completion_params.update(litellm_params_to_add)
-
- # 1. Emit initial task event (kind: "task", status: "submitted")
- task_event = A2ACompletionBridgeTransformation.create_task_event(ctx)
- yield task_event
-
- # 2. Emit status update (kind: "status-update", status: "working")
- working_event = A2ACompletionBridgeTransformation.create_status_update_event(
- ctx=ctx,
- state="working",
- final=False,
- message_text="Processing request...",
- )
- yield working_event
-
- # Call litellm.acompletion with streaming
- response = await litellm.acompletion(**completion_params)
-
- # 3. Accumulate content and emit artifact update
- accumulated_text = ""
- chunk_count = 0
- async for chunk in response: # type: ignore[union-attr]
- chunk_count += 1
-
- # Extract delta content
- content = ""
- if chunk is not None and hasattr(chunk, "choices") and chunk.choices:
- choice = chunk.choices[0]
- if hasattr(choice, "delta") and choice.delta:
- content = choice.delta.content or ""
-
- if content:
- accumulated_text += content
-
- # Emit artifact update with accumulated content
- if accumulated_text:
- artifact_event = (
- A2ACompletionBridgeTransformation.create_artifact_update_event(
- ctx=ctx,
- text=accumulated_text,
- )
- )
- yield artifact_event
-
- # 4. Emit final status update (kind: "status-update", status: "completed", final: true)
- completed_event = A2ACompletionBridgeTransformation.create_status_update_event(
- ctx=ctx,
- state="completed",
- final=True,
- )
- yield completed_event
-
- verbose_logger.info(
- f"A2A completion bridge streaming completed: request_id={request_id}, chunks={chunk_count}"
- )
-
-
-# Convenience functions that delegate to the class methods
-async def handle_a2a_completion(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
-) -> Dict[str, Any]:
- """Convenience function for non-streaming A2A completion."""
- return await A2ACompletionBridgeHandler.handle_non_streaming(
- request_id=request_id,
- params=params,
- litellm_params=litellm_params,
- api_base=api_base,
- )
-
-
-async def handle_a2a_completion_streaming(
- request_id: str,
- params: Dict[str, Any],
- litellm_params: Dict[str, Any],
- api_base: Optional[str] = None,
-) -> AsyncIterator[Dict[str, Any]]:
- """Convenience function for streaming A2A completion."""
- async for chunk in A2ACompletionBridgeHandler.handle_streaming(
- request_id=request_id,
- params=params,
- litellm_params=litellm_params,
- api_base=api_base,
- ):
- yield chunk
\ No newline at end of file
diff --git a/litellm/a2a_protocol/providers/litellm_completion/transformation.py b/litellm/a2a_protocol/providers/litellm_completion/transformation.py
deleted file mode 100644
--- a/litellm/a2a_protocol/providers/litellm_completion/transformation.py
+++ /dev/null
@@ -1,284 +1,0 @@
-"""
-Transformation utilities for A2A <-> OpenAI message format conversion.
-
-A2A Message Format:
-{
- "role": "user",
- "parts": [{"kind": "text", "text": "Hello!"}],
- "messageId": "abc123"
-}
-
-OpenAI Message Format:
-{"role": "user", "content": "Hello!"}
-
-A2A Streaming Events:
-- Task event (kind: "task") - Initial task creation with status "submitted"
-- Status update (kind: "status-update") - Status changes (working, completed)
-- Artifact update (kind: "artifact-update") - Content/artifact delivery
-"""
-
-from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional
-from uuid import uuid4
-
-from litellm._logging import verbose_logger
-
-
-class A2AStreamingContext:
- """
- Context holder for A2A streaming state.
- Tracks task_id, context_id, and message accumulation.
- """
-
- def __init__(self, request_id: str, input_message: Dict[str, Any]):
- self.request_id = request_id
- self.task_id = str(uuid4())
- self.context_id = str(uuid4())
- self.input_message = input_message
- self.accumulated_text = ""
- self.has_emitted_task = False
- self.has_emitted_working = False
-
-
-class A2ACompletionBridgeTransformation:
- """
- Static methods for transforming between A2A and OpenAI message formats.
- """
-
- @staticmethod
- def a2a_message_to_openai_messages(
- a2a_message: Dict[str, Any],
- ) -> List[Dict[str, str]]:
- """
- Transform an A2A message to OpenAI message format.
-
- Args:
- a2a_message: A2A message with role, parts, and messageId
-
- Returns:
- List of OpenAI-format messages
- """
- role = a2a_message.get("role", "user")
- parts = a2a_message.get("parts", [])
-
- # Map A2A roles to OpenAI roles
- openai_role = role
- if role == "user":
- openai_role = "user"
- elif role == "assistant":
- openai_role = "assistant"
- elif role == "system":
- openai_role = "system"
-
- # Extract text content from parts
- content_parts = []
- for part in parts:
- kind = part.get("kind", "")
- if kind == "text":
- text = part.get("text", "")
- content_parts.append(text)
-
- content = "\n".join(content_parts) if content_parts else ""
-
- verbose_logger.debug(
- f"A2A -> OpenAI transform: role={role} -> {openai_role}, content_length={len(content)}"
- )
-
- return [{"role": openai_role, "content": content}]
-
- @staticmethod
- def openai_response_to_a2a_response(
- response: Any,
- request_id: Optional[str] = None,
- ) -> Dict[str, Any]:
- """
- Transform a LiteLLM ModelResponse to A2A SendMessageResponse format.
-
- Args:
- response: LiteLLM ModelResponse object
- request_id: Original A2A request ID
-
- Returns:
- A2A SendMessageResponse dict
- """
- # Extract content from response
- content = ""
- if hasattr(response, "choices") and response.choices:
- choice = response.choices[0]
- if hasattr(choice, "message") and choice.message:
- content = choice.message.content or ""
-
- # Build A2A message
- a2a_message = {
- "role": "agent",
- "parts": [{"kind": "text", "text": content}],
- "messageId": uuid4().hex,
- }
-
- # Build A2A response
- a2a_response = {
- "jsonrpc": "2.0",
- "id": request_id,
- "result": {
- "message": a2a_message,
- },
- }
-
- verbose_logger.debug(f"OpenAI -> A2A transform: content_length={len(content)}")
-
- return a2a_response
-
- @staticmethod
- def _get_timestamp() -> str:
- """Get current timestamp in ISO format with timezone."""
- return datetime.now(timezone.utc).isoformat()
-
- @staticmethod
- def create_task_event(
- ctx: A2AStreamingContext,
- ) -> Dict[str, Any]:
- """
- Create the initial task event with status 'submitted'.
-
- This is the first event emitted in an A2A streaming response.
- """
- return {
- "id": ctx.request_id,
- "jsonrpc": "2.0",
- "result": {
- "contextId": ctx.context_id,
- "history": [
- {
- "contextId": ctx.context_id,
- "kind": "message",
- "messageId": ctx.input_message.get("messageId", uuid4().hex),
- "parts": ctx.input_message.get("parts", []),
- "role": ctx.input_message.get("role", "user"),
- "taskId": ctx.task_id,
- }
- ],
- "id": ctx.task_id,
- "kind": "task",
- "status": {
- "state": "submitted",
... diff truncated: showing 800 of 4058 linesYou can send follow-ups to the cloud agent here.
…ookup, scope discovery auto-fire to A2A types - merge_agent_card now defaults version to 1.0.0 when upstream omits it (A2A v1.0 schema requires the field). - invoke_agent_a2a guards against agent_card_params being None so plain chat agents routed via the A2A path return a JSON-RPC error instead of AttributeError. - buildDiscoveryRequest no longer falls back to any URL-shaped credential field for non-A2A agent types (Azure AI Foundry, Bedrock AgentCore, Vertex). Discovery only auto-fires for pure A2A and use_a2a_form_fields runtimes; the manual URL input remains available as an escape hatch.
Collaborator
… discovery Two findings from greptile review: 1. `overlayDiscoveredCardParams` was copy-pasted between `add_agent_form.tsx` and `agent_info.tsx`. Move it to `agent_discovery_utils.ts` so the create and edit flows share the same overlay logic and there's only one place to update when discovered fields change. 2. `agent_card_discovery.tsx` used a zero-debounce path for parent-driven mode, which fires one discovery HTTP request per keystroke when an admin types into the parent form's URL / api_base / assistant_id fields (the parent rebuilds the plan from watched form values every render). Apply the same 400ms debounce uniformly.
Collaborator
…nc url on re-discover
- _build_merged_agent_card: prefer card-supplied name over agent_name so
the discovery panel's editable 'Name (shown to API clients)' value is
not silently overwritten by the internal identifier.
- async_safe_get call in fetch_well_known_card: pass headers or {} to
avoid TypeError({**None, 'Host': ...}) when URL validation is enabled
in production (default).
- agent_info handleApplyDiscoveredCard: set url: selection.upstream_url
in fieldsToSet so re-discovery during edit refreshes the form's URL
field for pure A2A agents (matches add_agent_form).
Co-authored-by: Yassin Kortam <[email protected]>
Collaborator
Collaborator
|
bugbot run |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 2 potential issues.
Bugbot Autofix is ON, but it could not run because the spend limit has been reached. To enable Bugbot Autofix, have a team admin raise the spend limit in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 65484a8. Configure here.
Collaborator
|
bugbot run |
Bugbot is paused — on-demand spend limit reachedBugbot uses usage-based billing for this team and has hit its on-demand spend limit. A team admin can raise the spend limit in the Cursor dashboard, or wait for the next billing cycle to continue. |
Collaborator
|
bugbot run |
Bugbot is paused — on-demand spend limit reachedBugbot uses usage-based billing for this team and has hit its on-demand spend limit. A team admin can raise the spend limit in the Cursor dashboard, or wait for the next billing cycle to continue. |
Public agent_hub returned agent_card_params verbatim, exposing the
retained upstream backend url to unauthenticated callers. Rewrite the
url to the proxy /a2a/{agent_id} entrypoint on response, matching the
behavior of the authenticated well-known agent-card endpoint, so the
backend cannot be reached outside LiteLLM's auth, budget, and logging
path.
Collaborator
Collaborator
ryan-crabbe-berri
approved these changes
May 30, 2026
mateo-berri
merged commit af17400
into
litellm_internal_staging
116 of 118 checks passed
1 task
fzowl
pushed a commit
to fzowl/litellm
that referenced
this pull request
Jun 24, 2026
…erriAI#28860) * feat(a2a): well-known agent-card discovery + LangGraph Platform mode Adds a registration-time discovery flow so admins can paste an upstream agent URL, see its skills/capabilities, pick what to expose, and have the proxy front it with a LiteLLM-shaped agent card. Backend (new litellm/proxy/a2a/ module): - fetch_well_known_card walks /.well-known/agent-card.json, /.well-known/agent.json, /agent.json by default. langgraph_platform mode hits the canonical path with ?assistant_id=<id> (LangGraph serves one shared endpoint per deployment). - merge_agent_card overlays LiteLLM overrides on the upstream card: drops upstream url, forces protocolVersion=1.0, replaces securitySchemes with LiteLLMKey bearer, emits supportedInterfaces pointing at the proxy, filters capabilities to a small allowlist, strips non-v1.0 fields. - POST /v1/a2a/discover returns the raw upstream card (admin-only) so the UI can render skills/capabilities for selection. - create/update/patch agent endpoints pre-generate the agent_id and run merge_agent_card before storing, so DB.agent_card_params already embeds the proxy-fronted URL. UI (ui/litellm-dashboard): - New AgentCardDiscovery component with a parent-driven plan: discovery_mode + params + display URL. For LangGraph the parent composes (api_base, assistant_id); for pure A2A it uses the url field. Component hides the manual URL input when the parent drives. - add_agent_form wires discovery for every non-custom agent type and overlays the user's selections onto agent_card_params at submit, fixing the bug where dynamic agent forms ignored discovery picks. Completion-bridge fixes (paired): - Add kind: "message" to A2A response messages and unwrap result so it's a Message directly per spec (matches a2a SDK SendMessageResponse validation). - Forward A2A metadata to LangGraph runs via extra_body.metadata. * fix(a2a): preserve agent url, fix streaming chunk envelope, and protect forwarded metadata - Streaming chunk: move final out of the message object into the result envelope per the A2A spec. - Agent card merge: keep upstream url on the stored card so the runtime invocation path can locate the upstream backend; the public well-known endpoint already rewrites this field to the proxy URL before exposing it to clients. - Completion bridge: apply A2A forward metadata after merging litellm_params so an agent-configured extra_body cannot overwrite the forwarded metadata. Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a): fix legacy streaming chunk, agent card test, and metadata merge - providers/litellm_completion: move 'final' out of the message object into the result envelope per the A2A spec (matches the bridge fix). - agent endpoints test: the runtime invocation path now preserves the top-level 'url' on the stored card, so update the assertion to match. - completion bridge metadata: when forwarding A2A metadata via extra_body.metadata, merge into any existing extra_body.metadata instead of replacing it, so an agent-configured metadata block is preserved (forward metadata still wins on key conflicts). Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a): remove dead duplicate transformation dir; drop SSRF-prone headers field from /v1/a2a/discover Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a): revert accidental html→index.html rename from afc8b10 The commit afc8b10 bundled real A2A fixes alongside an unintended re-introduction of the */index.html layout that 8513d7f had already reverted. Restore all 35 static-export pages back to the flat *.html structure that matches the upstream main branch. Co-authored-by: Cursor <[email protected]> * fix(a2a): address PR review comments UI: - Auto-trigger discovery when connection details are filled; remove the "Use these selections" button (selection syncs live to parent, user just clicks Next). - Edit Settings: auto-discover upstream card on open; cross-check with DB-stored card so only already-saved skills/capabilities are pre-ticked. - Extract shared buildDiscoveryRequest + selectionsFromSavedAgentCard helpers into agent_discovery_utils.ts so both add and edit flows share the same logic. Backend: - agent_card.py: rename the proxy security requirements field from the non-standard ``securityRequirements`` to the spec-correct ``security`` key (matches AgentCard TypedDict and A2A/OpenAPI convention). - agent_card.py: remove ``securityRequirements`` from _ALLOWED_TOP_LEVEL_KEYS. - endpoints.py: _build_merged_agent_card now forwards agent_name and description from the request so the stored card reflects the admin- supplied name, not just whatever the upstream card advertised. - utils.py: remove overly-broad ``or "parts" in result`` fallback; use ``kind == "message"`` check only to avoid false matches on future result types that happen to include a ``parts`` field. - test_agent_card.py: update assertions to expect ``security`` key. Co-authored-by: Cursor <[email protected]> * fix: restore Next.js metadata directories to match upstream main The previous revert removed __next.* metadata subdirectories from git tracking entirely, but these directories exist on origin/main alongside the flat .html files. Restore them via checkout from origin/main so the PR diff only reflects actual code changes. Co-authored-by: Cursor <[email protected]> * fix(a2a): drop dead headers option from discoverAgentCardCall The backend /v1/a2a/discover endpoint no longer accepts a headers field (removed in 78591b2 for SSRF safety), so any headers passed through DiscoverAgentCardOptions were silently discarded by the API request body. Remove the field and the conditional that copies it onto the request body. * fix(a2a): skip merge for non-A2A agents and align pydantic-ai result shape The agent create/update/patch handlers ran the LiteLLM-fronting merge unconditionally, so registrations that did not provide agent_card_params still ended up with a synthesised card carrying supportedInterfaces, securitySchemes, and default skills. Gate the merge on a non-empty agent_card_params so plain chat/LLM agents stay non-A2A in the registry. Also move kind: 'message' inside the a2a_message dict in the Pydantic AI non-streaming response so its construction matches the completion bridge rather than spreading kind on top of a separate dict. * Fix three bugs in A2A discovery flow 1. UI: Stabilize discoveryRequest deps to avoid redundant /v1/a2a/discover API calls. The parent rebuilds the discoveryRequest object on every form keystroke, so depend on primitive proxies (discovery_mode + serialized params) rather than the object identity. Read the actual object via a ref inside handleDiscover. 2. Backend: Route the well-known card fetch through async_safe_get so the admin /v1/a2a/discover endpoint can't be used to probe private/loopback addresses or cloud metadata endpoints. SSRFError is a separate handled case so it surfaces a clear AgentCardDiscoveryError. 3. Streaming: Make openai_chunk_to_a2a_chunk emit the same flat result shape as the non-streaming response (kind/role/parts/messageId at the result level), with envelope-level 'final' added. Matches the existing create_artifact_update_event pattern and lets consumers read a uniform result shape across streaming and non-streaming. Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a/ui): include savedAgentCard in handleDiscover deps The previous deps list omitted savedAgentCard, so handleDiscover (and the resetSelections it calls) kept the closure's saved-card value even after the parent refetched the agent. Clicking 'Re-discover' would then pre-select skills against stale data. Adding savedAgentCard to the deps array forces the callback to refresh whenever the saved card changes. Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a): align pydantic-ai test + docstring with direct-Message result shape The non-streaming A2A response was changed so that `result` is the Message itself (kind="message"), per spec / SendMessageResponse. Update the PydanticAITransformation._transform_to_a2a_response test and docstring that still described the old `result.message` envelope so internal consumers match the producer. * fix(a2a): strip additionalInterfaces and let configured metadata win over A2A request - merge_agent_card no longer carries upstream additionalInterfaces through; storing those alternate URLs would let authenticated agent callers reach the backend directly and bypass proxy auth/budget/logging. - apply_forward_metadata_to_completion_params now layers client-supplied A2A metadata UNDER any agent-owner-configured extra_body.metadata, so server-set run metadata stays authoritative on key conflicts. * fix(agents): merge agent card even when agent_card_params is an empty dict Treat an explicitly provided empty agent_card_params ({}) as 'card provided but empty' instead of 'no card', so the LiteLLM-fronting merge still injects securitySchemes, supportedInterfaces, and protocolVersion. Without this, the well-known endpoint could serve a bare card with only a rewritten url, advertising no authentication to A2A clients. Co-authored-by: Yassin Kortam <[email protected]> * refactor(a2a): drop dead openai_chunk_to_a2a_chunk helper The deprecated single-chunk helper has no callers anywhere in the codebase — the streaming path emits proper A2A events via create_task_event / create_status_update_event / create_artifact_update_event in handler.py. Removing the dead method also eliminates the inconsistency where the unused chunk inlined the envelope-level final flag inside the Message result. * fix(a2a): scope a2a lazy-feature so it doesn't subsume /v1/a2a/discover - _lazy_features.py: use /a2a prefix + /message/send suffix for the a2a feature so a request to /v1/a2a/discover no longer triggers the a2a_endpoints module to load alongside a2a_registration. - agent_endpoints/endpoints.py: drop the no-op description override kwarg from _build_merged_agent_card and its three call sites. The upstream card's description is already preserved by merge_agent_card's deepcopy, so passing it explicitly did nothing. * style: black-format litellm/a2a_protocol/litellm_completion_bridge/transformation.py * fix: address PR bugfix review for a2a discovery + metadata forwarding - agent create form (add_agent_form.tsx): drop the skills.length > 0 guard so an admin can clear all discovered skills during creation, matching the edit form's overlay behavior (consistency between create and edit flows). - agent_card_discovery.tsx: stop including savedAgentCard in the handleDiscover useCallback deps. Read it via a ref inside resetSelections instead, so a parent-driven re-render that hands us a new savedAgentCard object reference (e.g. a background refresh of the agent record) does not recreate handleDiscover and re-fire the auto-discover effect, which would otherwise overwrite in-progress user edits in parent-driven mode (debounceMs = 0). - a2a_endpoints.invoke_agent_a2a: skip 'metadata' when moving litellm params off of A2A MessageSendParams into body. The A2A protocol defines params.metadata as a first-class request-level field, and the completion bridge's get_forward_metadata is supposed to merge it with message.metadata. Previously the proxy always stripped params.metadata before constructing MessageSendParams, so the params-level branch in get_forward_metadata was dead code in the proxy flow. Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a): return 404 from get_agent_card when agent has no card * fix(agents): apply discovery overlay uniformly on create and dedupe ALLOWED_CAPABILITY_KEYS - buildAgentData now applies overlayDiscoveredCardParams after every non-custom branch (a2a, use_a2a_form_fields, dynamic) so types with credential_fields no longer silently drop discovered skills, capabilities, input/output modes, provider, and icon/doc URLs on submit. Mirrors the edit flow in agent_info.tsx. - Export ALLOWED_CAPABILITY_KEYS from agent_discovery_utils and import it in agent_card_discovery so the rendering and selection-filtering logic share a single source of truth. Co-authored-by: Yassin Kortam <[email protected]> * ci(proxy-endpoints): wire tests/test_litellm/proxy/a2a into the shard The two new test files (test_discovery.py, test_agent_card.py) were not picked up by any pytest path, so their coverage never reached codecov and patch coverage fell below the auto target. * fix(ui): overlay discovered name/description in create flow for dynamic agents Mirror the edit-form overlay in agent_info.tsx so dynamic agent types (e.g. LangGraph) whose forms don't register name/description as Form.Items don't silently lose those discovery-panel edits on save. Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a): default merged agent card version, null-guard runtime URL lookup, scope discovery auto-fire to A2A types - merge_agent_card now defaults version to 1.0.0 when upstream omits it (A2A v1.0 schema requires the field). - invoke_agent_a2a guards against agent_card_params being None so plain chat agents routed via the A2A path return a JSON-RPC error instead of AttributeError. - buildDiscoveryRequest no longer falls back to any URL-shaped credential field for non-A2A agent types (Azure AI Foundry, Bedrock AgentCore, Vertex). Discovery only auto-fires for pure A2A and use_a2a_form_fields runtimes; the manual URL input remains available as an escape hatch. * fix(ui): extract overlayDiscoveredCardParams + debounce parent-driven discovery Two findings from greptile review: 1. `overlayDiscoveredCardParams` was copy-pasted between `add_agent_form.tsx` and `agent_info.tsx`. Move it to `agent_discovery_utils.ts` so the create and edit flows share the same overlay logic and there's only one place to update when discovered fields change. 2. `agent_card_discovery.tsx` used a zero-debounce path for parent-driven mode, which fires one discovery HTTP request per keystroke when an admin types into the parent form's URL / api_base / assistant_id fields (the parent rebuilds the plan from watched form values every render). Apply the same 400ms debounce uniformly. * fix(a2a): preserve discovery name edit, default discovery headers, sync url on re-discover - _build_merged_agent_card: prefer card-supplied name over agent_name so the discovery panel's editable 'Name (shown to API clients)' value is not silently overwritten by the internal identifier. - async_safe_get call in fetch_well_known_card: pass headers or {} to avoid TypeError({**None, 'Host': ...}) when URL validation is enabled in production (default). - agent_info handleApplyDiscoveredCard: set url: selection.upstream_url in fieldsToSet so re-discovery during edit refreshes the form's URL field for pure A2A agents (matches add_agent_form). Co-authored-by: Yassin Kortam <[email protected]> * fix(a2a): scrub upstream url from /public/agent_hub cards Public agent_hub returned agent_card_params verbatim, exposing the retained upstream backend url to unauthenticated callers. Rewrite the url to the proxy /a2a/{agent_id} entrypoint on response, matching the behavior of the authenticated well-known agent-card endpoint, so the backend cannot be reached outside LiteLLM's auth, budget, and logging path. * fix(a2a): include suffix-matched routes in lazy warm openapi fragment --------- Co-authored-by: Cursor Agent <[email protected]> Co-authored-by: Yassin Kortam <[email protected]> Co-authored-by: mateo-berri <[email protected]>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds a registration-time discovery flow for A2A agents so admins can paste an upstream URL, see what skills/capabilities it advertises, pick what to expose, and have the proxy front it with a LiteLLM-shaped card.
litellm/proxy/a2a/with two discovery modes:well_known_fallback(default) tries/.well-known/agent-card.json→/.well-known/agent.json→/agent.json.langgraph_platformhits the canonical well-known path with?assistant_id=<id>(LangGraph serves one shared endpoint per deployment, not a per-assistant subpath).merge_agent_cardoverlays LiteLLM overrides on the upstream card: drops upstreamurl, forcesprotocolVersion=1.0, replacessecuritySchemeswithLiteLLMKeybearer, emitssupportedInterfacespointing at the proxy, filters capabilities to a small allowlist, strips non-v1.0 fields.POST /v1/a2a/discover(admin-only) returns the raw upstream card so the UI can render skills + capabilities for selection.agent_id, runmerge_agent_card, and store the merged result.AgentCardDiscoverycomponent with a parent-driven discovery plan (mode + params + display URL). For LangGraph the parent composes(api_base, assistant_id); for pure A2A it uses theurlfield. Overlays the admin's selections ontoagent_card_paramsat submit so dynamic agent forms (LangGraph, Bedrock, Azure) actually carry the picked skills/capabilities.kind: "message"to A2A response messages and unwrapresultso it's aMessagedirectly per spec — old shape failedSendMessageResponse.model_validate. Forward A2Ametadatato LangGraph runs viaextra_body.metadata.Testing
Note
Medium Risk
Touches agent registration, admin-only SSRF-guarded upstream fetches, and A2A JSON-RPC response/metadata behavior that existing clients may depend on.
Overview
Adds A2A agent registration via well-known card discovery: a new
litellm/proxy/a2a/stack (fetch_well_known_cardwithwell_known_fallbackvslanggraph_platform, admin-onlyPOST /v1/a2a/discover, andmerge_agent_cardto store a proxy-fronted card with LiteLLM bearer security, filtered capabilities, andsupportedInterfacespointing at/a2a/{agent_id}). Create/update/patch on/v1/agentspre-generatesagent_idwhen needed and persists the merged card; lazy routing splits/v1/a2a/discoverfrom message-send paths.The dashboard gains
AgentCardDiscovery(auto/manual discover, skill/capability selection, LangGraph-aware discovery plans) and overlays selections ontoagent_card_paramsat submit/edit.Runtime/protocol fixes: non-streaming A2A responses use
resultas akind: "message"object (Pydantic AI + completion bridge); A2Ametadatais forwarded on runs viaextra_body.metadatawith server-config keys winning; proxy invoke no longer stripsmetadatafrom params; LangGraph message transform keeps per-messagemetadata; duplicateproviders/litellm_completionbridge code is removed in favor oflitellm_completion_bridge.Reviewed by Cursor Bugbot for commit 65484a8. Bugbot is set up for automated code reviews on this repo. Configure here.