[Feat] Agent Gateway - allow adding langgraph, bedrock agent core agents (#17802)

* fix: langgraph bridge streaming

* add public/agents/fields

* test_a2a_completion_bridge_non_streaming

* TestA2AStreamingTransformation

* AgentCredentialFieldMetadata

* add new logo

* refactor add agent

* fix add dynamic fields

* feat allow adding langgraph agent

* add langgraph provider

* stash

* add AgentCreateInfo

* agent_create_fields

* fix fields

* test_a2a_completion_bridge_bedrock_agentcore

* test_a2a_completion_bridge_bedrock_agentcore

* add public endpoints

* fix a2a endpoints

* fix dynamic fields
This commit is contained in:
Ishaan Jaff 2025-12-10 19:13:50 -08:00 • committed by GitHub
parent ad5d94a82f
commit 4a7437ba5f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 918 additions and 68 deletions

View file

@ -2,6 +2,12 @@
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
@ -10,6 +16,7 @@ import litellm
from litellm._logging import verbose_logger
from litellm.a2a_protocol.litellm_completion_bridge.transformation import (
A2ACompletionBridgeTransformation,
A2AStreamingContext,
)
@ -50,7 +57,8 @@ class A2ACompletionBridgeHandler:
model = litellm_params.get("model", "agent")
# Build full model string if provider specified
if custom_llm_provider:
# 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
@ -87,6 +95,12 @@ class A2ACompletionBridgeHandler:
"""
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
@ -94,11 +108,17 @@ class A2ACompletionBridgeHandler:
api_base: API base URL from agent_card_params
Yields:
A2A streaming response chunks
A2A streaming response events
"""
# 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
@ -109,7 +129,8 @@ class A2ACompletionBridgeHandler:
model = litellm_params.get("model", "agent")
# Build full model string if provider specified
if custom_llm_provider:
# 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
@ -118,6 +139,19 @@ class A2ACompletionBridgeHandler:
f"A2A completion bridge streaming: model={full_model}, api_base={api_base}"
)
# 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(
model=full_model,
@ -126,27 +160,37 @@ class A2ACompletionBridgeHandler:
stream=True,
)
# 3. Accumulate content and emit artifact update
accumulated_text = ""
chunk_count = 0
async for chunk in response: # type: ignore[union-attr]
chunk_count += 1
a2a_chunk = A2ACompletionBridgeTransformation.openai_chunk_to_a2a_chunk(
chunk=chunk,
request_id=request_id,
is_final=False,
)
if a2a_chunk:
yield a2a_chunk
# Send final chunk
final_chunk = A2ACompletionBridgeTransformation.openai_chunk_to_a2a_chunk(
chunk=None,
request_id=request_id,
is_final=True,
# 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,
)
if final_chunk:
# Clear content for final chunk
final_chunk["result"]["message"]["parts"][0]["text"] = ""
yield final_chunk
yield completed_event
verbose_logger.info(
f"A2A completion bridge streaming completed: request_id={request_id}, chunks={chunk_count}"

View file

@ -10,14 +10,36 @@ A2A Message Format:
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.
@ -108,6 +130,114 @@ class A2ACompletionBridgeTransformation:
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,
}
return {
"id": ctx.request_id,
"jsonrpc": "2.0",
"result": {
"contextId": ctx.context_id,
"final": final,
"kind": "status-update",
"status": status,
"taskId": ctx.task_id,
},
}
@staticmethod
def create_artifact_update_event(
ctx: A2AStreamingContext,
text: str,
) -> Dict[str, Any]:
"""
Create an artifact update event with content.
Args:
ctx: Streaming context
text: The text content for the artifact
"""
return {
"id": ctx.request_id,
"jsonrpc": "2.0",
"result": {
"artifact": {
"artifactId": str(uuid4()),
"name": "response",
"parts": [{"kind": "text", "text": text}],
},
"contextId": ctx.context_id,
"kind": "artifact-update",
"taskId": ctx.task_id,
},
}
@staticmethod
def openai_chunk_to_a2a_chunk(
chunk: Any,
@ -117,6 +247,10 @@ class A2ACompletionBridgeTransformation:
"""
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
@ -135,7 +269,7 @@ class A2ACompletionBridgeTransformation:
if not content and not is_final:
return None
# Build A2A streaming chunk
# Build A2A streaming chunk (legacy format)
a2a_chunk = {
"jsonrpc": "2.0",
"id": request_id,

View file

@ -186,8 +186,7 @@ async def asend_message(
if custom_llm_provider:
if request is None:
raise ValueError("request is required for completion bridge")
if api_base is None:
raise ValueError("api_base is required for completion bridge")
# api_base is optional for providers that derive endpoint from model (e.g., bedrock/agentcore)
verbose_logger.info(
f"A2A using completion bridge: provider={custom_llm_provider}, api_base={api_base}"
@ -334,8 +333,7 @@ async def asend_message_streaming(
if custom_llm_provider:
if request is None:
raise ValueError("request is required for completion bridge")
if api_base is None:
raise ValueError("api_base is required for completion bridge")
# api_base is optional for providers that derive endpoint from model (e.g., bedrock/agentcore)
verbose_logger.info(
f"A2A streaming using completion bridge: provider={custom_llm_provider}"

View file

@ -46,7 +46,7 @@ def _get_agent(agent_id: str):
async def _handle_stream_message(
api_base: str,
api_base: Optional[str],
request_id: str,
params: dict,
litellm_params: Optional[dict] = None,
@ -213,13 +213,17 @@ async def invoke_agent_a2a(
# Get backend URL and agent name
agent_url = agent.agent_card_params.get("url")
agent_name = agent.agent_card_params.get("name", agent_id)
if not agent_url:
return _jsonrpc_error(request_id, -32000, f"Agent '{agent_id}' has no URL configured", 500)
verbose_proxy_logger.info(f"Proxying A2A request to agent '{agent_id}' at {agent_url}")
# Get litellm_params (may include custom_llm_provider for completion bridge)
litellm_params = agent.litellm_params or {}
custom_llm_provider = litellm_params.get("custom_llm_provider")
# URL is required unless using completion bridge with a provider that derives endpoint from model
# (e.g., bedrock/agentcore derives endpoint from ARN in model string)
if not agent_url and not custom_llm_provider:
return _jsonrpc_error(request_id, -32000, f"Agent '{agent_id}' has no URL configured", 500)
verbose_proxy_logger.info(f"Proxying A2A request to agent '{agent_id}' at {agent_url or 'completion-bridge'}")
# Set up data dict for litellm processing
body.update({

View file

@ -4,6 +4,10 @@ model_list:
model: openai/gpt-4o-mini
tpm: 1000
# LangGraph models
- model_name: langgraph/*
litellm_params:
model: langgraph/*
litellm_settings:
callbacks: ["dynamic_rate_limiter_v3"]

View file

@ -0,0 +1,76 @@
[
{
"agent_type": "a2a",
"agent_type_display_name": "A2A Standard",
"description": "Standard A2A protocol",
"logo_url": "/assets/logos/a2a_agent.png",
"credential_fields": [],
"litellm_params_template": {}
},
{
"agent_type": "langgraph",
"agent_type_display_name": "LangGraph",
"description": "Connect to LangGraph agents via the LangGraph Platform API",
"logo_url": "/assets/logos/langgraph.png",
"model_template": "langgraph/{assistant_id}",
"credential_fields": [
{
"key": "assistant_id",
"label": "Assistant ID",
"placeholder": "agent",
"tooltip": "The assistant/agent ID from your LangGraph deployment",
"required": true,
"field_type": "text",
"default_value": "agent",
"include_in_litellm_params": false
},
{
"key": "api_base",
"label": "LangGraph API Base",
"placeholder": "http://localhost:2024",
"tooltip": "The base URL for your LangGraph server (e.g., http://localhost:2024 or your deployed LangGraph Cloud URL)",
"required": true,
"field_type": "text",
"default_value": "http://localhost:2024",
"include_in_litellm_params": true
},
{
"key": "api_key",
"label": "LangGraph API Key",
"placeholder": null,
"tooltip": "API key for authenticating with your LangGraph server (optional for local development)",
"required": false,
"field_type": "password",
"default_value": null,
"include_in_litellm_params": true
}
],
"litellm_params_template": {
"custom_llm_provider": "langgraph"
}
},
{
"agent_type": "bedrock_agentcore",
"agent_type_display_name": "Bedrock AgentCore",
"description": "Connect to Amazon Bedrock AgentCore hosted agent runtimes",
"logo_url": "/assets/logos/bedrock.svg",
"inherit_credentials_from_provider": "Bedrock",
"model_template": "bedrock/agentcore/{agent_runtime_arn}",
"credential_fields": [
{
"key": "agent_runtime_arn",
"label": "Agent Runtime ARN",
"placeholder": "arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my-agent-runtime",
"tooltip": "The ARN of your Bedrock AgentCore runtime. Find this in your AWS Bedrock console under AgentCore.",
"required": true,
"field_type": "text",
"default_value": null,
"include_in_litellm_params": false
}
],
"litellm_params_template": {
"custom_llm_provider": "bedrock"
}
}
]

View file

@ -1,6 +1,6 @@
from typing import List
import os
import json
import os
from typing import List
from fastapi import APIRouter, Depends, HTTPException
@ -12,6 +12,7 @@ from litellm.types.proxy.management_endpoints.model_management_endpoints import
ModelGroupInfoProxy,
)
from litellm.types.proxy.public_endpoints.public_endpoints import (
AgentCreateInfo,
ProviderCreateInfo,
PublicModelHubInfo,
)
@ -167,3 +168,52 @@ async def get_litellm_model_cost_map():
status_code=500,
detail=f"Internal Server Error ({str(e)})",
)
@router.get(
"/public/agents/fields",
tags=["public", "[beta] Agents"],
response_model=List[AgentCreateInfo],
)
async def get_agent_fields() -> List[AgentCreateInfo]:
"""
Return agent type metadata required by the dashboard create-agent flow.
If an agent has `inherit_credentials_from_provider`, the provider's credential
fields are automatically appended to the agent's credential_fields.
"""
base_path = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
"proxy",
"public_endpoints",
)
agent_create_fields_path = os.path.join(base_path, "agent_create_fields.json")
provider_create_fields_path = os.path.join(base_path, "provider_create_fields.json")
with open(agent_create_fields_path, "r") as f:
agent_create_fields = json.load(f)
with open(provider_create_fields_path, "r") as f:
provider_create_fields = json.load(f)
# Build a lookup map for providers by name
provider_map = {p["provider"]: p for p in provider_create_fields}
# Merge inherited credential fields
for agent in agent_create_fields:
inherit_from = agent.get("inherit_credentials_from_provider")
if inherit_from and inherit_from in provider_map:
provider = provider_map[inherit_from]
# Copy provider fields and mark them for inclusion in litellm_params
inherited_fields = []
for field in provider.get("credential_fields", []):
field_copy = field.copy()
field_copy["include_in_litellm_params"] = True
inherited_fields.append(field_copy)
# Append provider credential fields after agent's own fields
agent["credential_fields"] = agent.get("credential_fields", []) + inherited_fields
# Remove the inherit field from response (not needed by frontend)
agent.pop("inherit_credentials_from_provider", None)
return agent_create_fields

View file

@ -27,3 +27,25 @@ class ProviderCreateInfo(BaseModel):
litellm_provider: str
credential_fields: List[ProviderCredentialField]
default_model_placeholder: Optional[str] = None
class AgentCredentialField(BaseModel):
key: str
label: str
placeholder: Optional[str] = None
tooltip: Optional[str] = None
required: bool = False
field_type: Literal["text", "password", "select", "upload", "textarea"] = "text"
options: Optional[List[str]] = None
default_value: Optional[str] = None
include_in_litellm_params: Optional[bool] = None
class AgentCreateInfo(BaseModel):
agent_type: str
agent_type_display_name: str
description: Optional[str] = None
logo_url: Optional[str] = None
credential_fields: List[AgentCredentialField]
litellm_params_template: Optional[Dict[str, str]] = None
model_template: Optional[str] = None

View file

@ -71,6 +71,12 @@ async def test_a2a_completion_bridge_non_streaming():
async def test_a2a_completion_bridge_streaming():
"""
Test streaming A2A request via the completion bridge with LangGraph provider.
Validates proper A2A streaming format with 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
"""
from litellm.a2a_protocol import asend_message_streaming
@ -98,19 +104,100 @@ async def test_a2a_completion_bridge_streaming():
chunks.append(chunk)
print(f"Chunk: {chunk}")
# Validate we received chunks
assert len(chunks) > 0
# Validate we received proper A2A streaming events
assert len(chunks) >= 4, f"Expected at least 4 chunks (task, working, artifact, completed), got {len(chunks)}"
# Validate chunk structure (chunks are dicts from bridge)
# Validate chunk structure follows A2A spec
for chunk in chunks:
assert "jsonrpc" in chunk
assert chunk["jsonrpc"] == "2.0"
assert "id" in chunk
assert "result" in chunk
assert "message" in chunk["result"]
message = chunk["result"]["message"]
assert "role" in message
assert message["role"] == "agent"
assert "parts" in message
print(f"Received {len(chunks)} chunks")
# Validate first chunk is task event
task_chunk = chunks[0]
assert task_chunk["result"]["kind"] == "task", "First chunk should be task event"
assert task_chunk["result"]["status"]["state"] == "submitted"
assert "contextId" in task_chunk["result"]
assert "id" in task_chunk["result"] # task id
assert "history" in task_chunk["result"]
# Validate second chunk is working status update
working_chunk = chunks[1]
assert working_chunk["result"]["kind"] == "status-update", "Second chunk should be status-update"
assert working_chunk["result"]["status"]["state"] == "working"
assert "taskId" in working_chunk["result"]
assert "contextId" in working_chunk["result"]
assert working_chunk["result"]["final"] is False
# Validate artifact update chunk
artifact_chunk = chunks[2]
assert artifact_chunk["result"]["kind"] == "artifact-update", "Third chunk should be artifact-update"
assert "artifact" in artifact_chunk["result"]
assert "artifactId" in artifact_chunk["result"]["artifact"]
assert "parts" in artifact_chunk["result"]["artifact"]
assert len(artifact_chunk["result"]["artifact"]["parts"]) > 0
assert artifact_chunk["result"]["artifact"]["parts"][0]["kind"] == "text"
# Validate final chunk is completed status update
final_chunk = chunks[-1]
assert final_chunk["result"]["kind"] == "status-update", "Last chunk should be status-update"
assert final_chunk["result"]["status"]["state"] == "completed"
assert final_chunk["result"]["final"] is True
print(f"Received {len(chunks)} chunks with proper A2A streaming format")
@pytest.mark.asyncio
async def test_a2a_completion_bridge_bedrock_agentcore():
"""
Test A2A request via the completion bridge with Bedrock AgentCore provider.
Uses the AgentCore runtime ARN to call a hosted agent.
"""
from litellm.a2a_protocol import asend_message_streaming
litellm._turn_on_debug()
# Bedrock AgentCore ARN (streaming-capable runtime)
agentcore_arn = "arn:aws:bedrock-agentcore:us-west-2:888602223428:runtime/hosted_agent_r9jvp-3ySZuRHjLC"
send_message_payload = {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "Explain machine learning in simple terms"}],
"messageId": uuid4().hex,
}
}
request = SendStreamingMessageRequest(
id=str(uuid4()),
params=MessageSendParams(**send_message_payload), # type: ignore
)
chunks = []
async for chunk in asend_message_streaming(
request=request,
api_base=None, # Not needed for Bedrock AgentCore
litellm_params={
"custom_llm_provider": "bedrock",
"model": f"bedrock/agentcore/{agentcore_arn}",
},
):
chunks.append(chunk)
print(f"Chunk: {chunk}")
# Validate we received proper A2A streaming events
assert len(chunks) >= 4, f"Expected at least 4 chunks, got {len(chunks)}"
# Validate first chunk is task event
assert chunks[0]["result"]["kind"] == "task"
assert chunks[0]["result"]["status"]["state"] == "submitted"
# Validate final chunk is completed status
assert chunks[-1]["result"]["kind"] == "status-update"
assert chunks[-1]["result"]["status"]["state"] == "completed"
assert chunks[-1]["result"]["final"] is True
print(f"Received {len(chunks)} chunks from Bedrock AgentCore")

View file

@ -0,0 +1,159 @@
"""
Test A2A completion bridge streaming transformation to proper A2A format.
Tests that the completion bridge 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
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
class TestA2AStreamingTransformation:
"""Test the A2A streaming transformation creates proper events."""
def test_create_task_event(self):
"""Test that create_task_event produces proper A2A task event structure."""
from litellm.a2a_protocol.litellm_completion_bridge.transformation import (
A2ACompletionBridgeTransformation,
A2AStreamingContext,
)
input_message = {
"role": "user",
"parts": [{"kind": "text", "text": "Hello"}],
"messageId": "msg-123",
}
ctx = A2AStreamingContext(request_id="req-456", input_message=input_message)
event = A2ACompletionBridgeTransformation.create_task_event(ctx)
# Validate structure
assert event["jsonrpc"] == "2.0"
assert event["id"] == "req-456"
assert event["result"]["kind"] == "task"
assert event["result"]["status"]["state"] == "submitted"
assert "contextId" in event["result"]
assert "id" in event["result"] # task id
assert "history" in event["result"]
assert len(event["result"]["history"]) == 1
assert event["result"]["history"][0]["role"] == "user"
def test_create_status_update_working(self):
"""Test that create_status_update_event produces proper working status."""
from litellm.a2a_protocol.litellm_completion_bridge.transformation import (
A2ACompletionBridgeTransformation,
A2AStreamingContext,
)
ctx = A2AStreamingContext(
request_id="req-456",
input_message={"role": "user", "parts": []},
)
event = A2ACompletionBridgeTransformation.create_status_update_event(
ctx=ctx,
state="working",
final=False,
message_text="Processing...",
)
assert event["result"]["kind"] == "status-update"
assert event["result"]["status"]["state"] == "working"
assert event["result"]["final"] is False
assert "taskId" in event["result"]
assert "contextId" in event["result"]
assert "timestamp" in event["result"]["status"]
def test_create_artifact_update(self):
"""Test that create_artifact_update_event produces proper artifact event."""
from litellm.a2a_protocol.litellm_completion_bridge.transformation import (
A2ACompletionBridgeTransformation,
A2AStreamingContext,
)
ctx = A2AStreamingContext(
request_id="req-456",
input_message={"role": "user", "parts": []},
)
event = A2ACompletionBridgeTransformation.create_artifact_update_event(
ctx=ctx,
text="Hello, I am an AI assistant.",
)
assert event["result"]["kind"] == "artifact-update"
assert "artifact" in event["result"]
assert "artifactId" in event["result"]["artifact"]
assert event["result"]["artifact"]["name"] == "response"
assert event["result"]["artifact"]["parts"][0]["kind"] == "text"
assert event["result"]["artifact"]["parts"][0]["text"] == "Hello, I am an AI assistant."
@pytest.mark.asyncio
async def test_handle_streaming_emits_proper_events():
"""Test that handle_streaming emits events in correct order with proper structure."""
from litellm.a2a_protocol.litellm_completion_bridge.handler import (
A2ACompletionBridgeHandler,
)
# Mock litellm.acompletion to return a streaming response
mock_chunk1 = MagicMock()
mock_chunk1.choices = [MagicMock()]
mock_chunk1.choices[0].delta = MagicMock()
mock_chunk1.choices[0].delta.content = "Hello"
mock_chunk2 = MagicMock()
mock_chunk2.choices = [MagicMock()]
mock_chunk2.choices[0].delta = MagicMock()
mock_chunk2.choices[0].delta.content = " world"
async def mock_streaming_response():
yield mock_chunk1
yield mock_chunk2
with patch("litellm.acompletion", new_callable=AsyncMock) as mock_acompletion:
mock_acompletion.return_value = mock_streaming_response()
params = {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "Hi"}],
"messageId": "msg-123",
}
}
events = []
async for event in A2ACompletionBridgeHandler.handle_streaming(
request_id="req-456",
params=params,
litellm_params={"custom_llm_provider": "langgraph", "model": "agent"},
api_base="http://localhost:2024",
):
events.append(event)
# Should have 4 events: task, working, artifact, completed
assert len(events) == 4
# Event 1: task submitted
assert events[0]["result"]["kind"] == "task"
assert events[0]["result"]["status"]["state"] == "submitted"
# Event 2: status working
assert events[1]["result"]["kind"] == "status-update"
assert events[1]["result"]["status"]["state"] == "working"
assert events[1]["result"]["final"] is False
# Event 3: artifact update with accumulated content
assert events[2]["result"]["kind"] == "artifact-update"
assert events[2]["result"]["artifact"]["parts"][0]["text"] == "Hello world"
# Event 4: status completed
assert events[3]["result"]["kind"] == "status-update"
assert events[3]["result"]["status"]["state"] == "completed"
assert events[3]["result"]["final"] is True

Binary file not shown.

After

Width:  |  Height:  |  Size: 5.4 KiB

View file

@ -1,7 +1,9 @@
import React, { useState } from "react";
import { Modal, Form, Button as AntButton, message } from "antd";
import { createAgentCall } from "../networking";
import React, { useState, useEffect } from "react";
import { Modal, Form, message, Select } from "antd";
import { Button } from "@tremor/react";
import { createAgentCall, getAgentCreateMetadata, AgentCreateInfo } from "../networking";
import AgentFormFields from "./agent_form_fields";
import DynamicAgentFormFields, { buildDynamicAgentData } from "./dynamic_agent_form_fields";
import { getDefaultFormValues, buildAgentDataFromForm } from "./agent_config";
interface AddAgentFormProps {
@ -19,6 +21,29 @@ const AddAgentForm: React.FC<AddAgentFormProps> = ({
}) => {
const [form] = Form.useForm();
const [isSubmitting, setIsSubmitting] = useState(false);
const [agentType, setAgentType] = useState<string>("a2a");
const [agentTypeMetadata, setAgentTypeMetadata] = useState<AgentCreateInfo[]>([]);
const [loadingMetadata, setLoadingMetadata] = useState(false);
// Fetch agent type metadata on mount
useEffect(() => {
const fetchMetadata = async () => {
setLoadingMetadata(true);
try {
const metadata = await getAgentCreateMetadata();
setAgentTypeMetadata(metadata);
} catch (error) {
console.error("Error fetching agent metadata:", error);
} finally {
setLoadingMetadata(false);
}
};
fetchMetadata();
}, []);
const selectedAgentTypeInfo = agentTypeMetadata.find(
(info) => info.agent_type === agentType
);
const handleSubmit = async (values: any) => {
if (!accessToken) {
@ -28,10 +53,18 @@ const AddAgentForm: React.FC<AddAgentFormProps> = ({
setIsSubmitting(true);
try {
const agentData = buildAgentDataFromForm(values);
let agentData: any;
if (agentType === "a2a") {
agentData = buildAgentDataFromForm(values);
} else if (selectedAgentTypeInfo) {
agentData = buildDynamicAgentData(values, selectedAgentTypeInfo);
}
await createAgentCall(accessToken, agentData);
message.success("Agent created successfully");
form.resetFields();
setAgentType("a2a");
onSuccess();
onClose();
} catch (error) {
@ -44,42 +77,114 @@ const AddAgentForm: React.FC<AddAgentFormProps> = ({
const handleCancel = () => {
form.resetFields();
setAgentType("a2a");
onClose();
};
const handleAgentTypeChange = (value: string) => {
setAgentType(value);
form.resetFields();
};
// Get the logo for the selected agent type for the header
const selectedLogo = selectedAgentTypeInfo?.logo_url || agentTypeMetadata.find(a => a.agent_type === "a2a")?.logo_url;
return (
<Modal
title="Add New Agent"
title={
<div className="flex items-center space-x-3 pb-4 border-b border-gray-100">
{selectedLogo && (
<img
src={selectedLogo}
alt="Agent"
className="w-6 h-6 object-contain"
/>
)}
<h2 className="text-xl font-semibold text-gray-900">Add New Agent</h2>
</div>
}
open={visible}
onCancel={handleCancel}
footer={null}
width={800}
width={900}
className="top-8"
styles={{
body: { padding: "24px" },
header: { padding: "24px 24px 0 24px", border: "none" },
}}
>
<Form
form={form}
layout="vertical"
onFinish={handleSubmit}
initialValues={getDefaultFormValues()}
>
<AgentFormFields showAgentName={true} />
<Form.Item>
<div style={{ display: "flex", justifyContent: "flex-end", gap: "8px" }}>
<AntButton onClick={handleCancel}>
Cancel
</AntButton>
<AntButton
htmlType="submit"
loading={isSubmitting}
<div className="mt-4">
<Form
form={form}
layout="vertical"
onFinish={handleSubmit}
initialValues={agentType === "a2a" ? getDefaultFormValues() : {}}
className="space-y-4"
>
{/* Agent Type Selection */}
<Form.Item
label={<span className="text-sm font-medium text-gray-700">Agent Type</span>}
required
tooltip="Select the type of agent you want to create"
>
<Select
value={agentType}
onChange={handleAgentTypeChange}
size="large"
style={{ width: "100%" }}
optionLabelProp="label"
>
Create Agent
</AntButton>
{agentTypeMetadata.map((info) => (
<Select.Option
key={info.agent_type}
value={info.agent_type}
label={
<div className="flex items-center gap-2">
<img src={info.logo_url || ""} alt="" className="w-4 h-4 object-contain" />
<span>{info.agent_type_display_name}</span>
</div>
}
>
<div className="flex items-center gap-3 py-1">
<img
src={info.logo_url || ""}
alt={info.agent_type_display_name}
className="w-5 h-5 object-contain"
/>
<div>
<div className="font-medium">{info.agent_type_display_name}</div>
{info.description && (
<div className="text-xs text-gray-500">{info.description}</div>
)}
</div>
</div>
</Select.Option>
))}
</Select>
</Form.Item>
{/* Conditional Form Fields */}
<div className="mt-6">
{agentType === "a2a" ? (
<AgentFormFields showAgentName={true} />
) : selectedAgentTypeInfo ? (
<DynamicAgentFormFields agentTypeInfo={selectedAgentTypeInfo} />
) : null}
</div>
</Form.Item>
</Form>
{/* Footer Buttons */}
<div className="flex items-center justify-end space-x-3 pt-6 border-t border-gray-100 mt-6">
<Button variant="secondary" onClick={handleCancel}>
Cancel
</Button>
<Button variant="primary" loading={isSubmitting}>
{isSubmitting ? "Creating..." : "Create Agent"}
</Button>
</div>
</Form>
</div>
</Modal>
);
};
export default AddAgentForm;

View file

@ -0,0 +1,125 @@
import React from "react";
import { Form, Input, Select } from "antd";
import { AgentCreateInfo, AgentCredentialFieldMetadata } from "../networking";
interface DynamicAgentFormFieldsProps {
agentTypeInfo: AgentCreateInfo;
}
/**
* Form fields for dynamic agent types (e.g., LangGraph).
* Renders common fields (agent name, display name, description) plus
* credential fields defined by the agent type metadata.
*/
const DynamicAgentFormFields: React.FC<DynamicAgentFormFieldsProps> = ({
agentTypeInfo,
}) => {
return (
<>
<Form.Item
label="Agent Name"
name="agent_name"
rules={[{ required: true, message: "Please enter a unique agent name" }]}
tooltip="Unique identifier for the agent"
>
<Input placeholder="e.g., my-langgraph-agent" />
</Form.Item>
<Form.Item
label="Description"
name="description"
tooltip="Brief description of what this agent does"
>
<Input.TextArea rows={2} placeholder="Describe what this agent does..." />
</Form.Item>
{agentTypeInfo.credential_fields.map((field: AgentCredentialFieldMetadata) => (
<Form.Item
key={field.key}
label={field.label}
name={field.key}
rules={field.required ? [{ required: true, message: `Please enter ${field.label}` }] : undefined}
tooltip={field.tooltip}
initialValue={field.default_value}
>
{field.field_type === "password" ? (
<Input.Password placeholder={field.placeholder || ""} />
) : field.field_type === "textarea" ? (
<Input.TextArea rows={3} placeholder={field.placeholder || ""} />
) : field.field_type === "select" && field.options ? (
<Select placeholder={field.placeholder || ""}>
{field.options.map((opt) => (
<Select.Option key={opt} value={opt}>
{opt}
</Select.Option>
))}
</Select>
) : (
<Input placeholder={field.placeholder || ""} />
)}
</Form.Item>
))}
</>
);
};
/**
* Builds agent data from form values for dynamic agent types.
* Uses configuration from agentTypeInfo to determine which fields to include.
*/
export const buildDynamicAgentData = (
values: any,
agentTypeInfo: AgentCreateInfo
) => {
// Build litellm_params from template
const litellmParams: Record<string, any> = {
...(agentTypeInfo.litellm_params_template || {}),
};
// Add credential fields marked with include_in_litellm_params
for (const field of agentTypeInfo.credential_fields) {
const value = values[field.key];
if (value && field.include_in_litellm_params !== false) {
litellmParams[field.key] = value;
}
}
// Apply model_template if defined (e.g., "bedrock/agentcore/{agent_runtime_arn}")
if (agentTypeInfo.model_template) {
let model = agentTypeInfo.model_template;
// Replace {field_key} placeholders with actual values
for (const field of agentTypeInfo.credential_fields) {
const placeholder = `{${field.key}}`;
if (model.includes(placeholder) && values[field.key]) {
model = model.replace(placeholder, values[field.key]);
}
}
litellmParams.model = model;
}
return {
agent_name: values.agent_name,
agent_card_params: {
protocolVersion: "1.0",
name: values.display_name || values.agent_name,
description: values.description || `${agentTypeInfo.agent_type_display_name} agent`,
url: values.api_base || "",
version: "1.0.0",
defaultInputModes: ["text"],
defaultOutputModes: ["text"],
capabilities: {
streaming: true,
},
skills: [{
id: "chat",
name: "Chat",
description: "General chat capability",
tags: ["chat", "conversation"],
}],
},
litellm_params: litellmParams,
};
};
export default DynamicAgentFormFields;

View file

@ -195,6 +195,28 @@ export interface ProviderCreateInfo {
credential_fields: ProviderCredentialFieldMetadata[];
}
export interface AgentCredentialFieldMetadata {
key: string;
label: string;
placeholder?: string | null;
tooltip?: string | null;
required?: boolean;
field_type?: "text" | "password" | "select" | "upload" | "textarea";
options?: string[] | null;
default_value?: string | null;
include_in_litellm_params?: boolean;
}
export interface AgentCreateInfo {
agent_type: string;
agent_type_display_name: string;
description?: string | null;
logo_url?: string | null;
credential_fields: AgentCredentialFieldMetadata[];
litellm_params_template?: Record<string, string> | null;
model_template?: string | null;
}
export interface PublicModelHubInfo {
docs_title: string;
custom_docs_description: string | null;
@ -255,6 +277,26 @@ export const getProviderCreateMetadata = async (): Promise<ProviderCreateInfo[]>
return jsonData;
};
export const getAgentCreateMetadata = async (): Promise<AgentCreateInfo[]> => {
/**
* Fetch agent type metadata from the proxy's public endpoint.
* This is used by the UI to dynamically render agent-specific credential fields.
*/
const url = proxyBaseUrl ? `${proxyBaseUrl}/public/agents/fields` : `/public/agents/fields`;
const response = await fetch(url, {
method: "GET",
});
if (!response.ok) {
const errorText = await response.text();
console.error("Failed to fetch agent create metadata:", response.status, errorText);
throw new Error("Failed to load agent configuration");
}
const jsonData: AgentCreateInfo[] = await response.json();
return jsonData;
};
// Global variable for the header name
let globalLitellmHeaderName: string = "Authorization";
const MCP_AUTH_HEADER: string = "x-mcp-auth";