feat: Allow users to pass their own agent card to to_a2a method

PiperOrigin-RevId: 802763510
This commit is contained in:
Xiang (Sean) Zhou
2025-09-03 16:54:33 -07:00
committed by Copybara-Service
parent a30851ee16
commit a1679dae3f
2 changed files with 235 additions and 4 deletions
+55 -4
View File
@@ -21,6 +21,7 @@ try:
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCard
except ImportError as e:
if sys.version_info < (3, 10):
raise ImportError(
@@ -29,6 +30,9 @@ except ImportError as e:
else:
raise e
from typing import Optional
from typing import Union
from starlette.applications import Starlette
from ...agents.base_agent import BaseAgent
@@ -43,6 +47,41 @@ from ..experimental import a2a_experimental
from .agent_card_builder import AgentCardBuilder
def _load_agent_card(
agent_card: Optional[Union[AgentCard, str]],
) -> Optional[AgentCard]:
"""Load agent card from various sources.
Args:
agent_card: AgentCard object, path to JSON file, or None
Returns:
AgentCard object or None if no agent card provided
Raises:
ValueError: If loading agent card from file fails
"""
if agent_card is None:
return None
if isinstance(agent_card, str):
# Load agent card from file path
import json
from pathlib import Path
try:
path = Path(agent_card)
with path.open("r", encoding="utf-8") as f:
agent_card_data = json.load(f)
return AgentCard(**agent_card_data)
except Exception as e:
raise ValueError(
f"Failed to load agent card from {agent_card}: {e}"
) from e
else:
return agent_card
@a2a_experimental
def to_a2a(
agent: BaseAgent,
@@ -50,6 +89,7 @@ def to_a2a(
host: str = "localhost",
port: int = 8000,
protocol: str = "http",
agent_card: Optional[Union[AgentCard, str]] = None,
) -> Starlette:
"""Convert an ADK agent to a A2A Starlette application.
@@ -58,6 +98,9 @@ def to_a2a(
host: The host for the A2A RPC URL (default: "localhost")
port: The port for the A2A RPC URL (default: 8000)
protocol: The protocol for the A2A RPC URL (default: "http")
agent_card: Optional pre-built AgentCard object or path to agent card
JSON. If not provided, will be built automatically from the
agent.
Returns:
A Starlette application that can be run with uvicorn
@@ -66,6 +109,9 @@ def to_a2a(
agent = MyAgent()
app = to_a2a(agent, host="localhost", port=8000, protocol="http")
# Then run with: uvicorn module:app --host localhost --port 8000
# Or with custom agent card:
app = to_a2a(agent, agent_card=my_custom_agent_card)
"""
# Set up ADK logging to ensure logs are visible when using uvicorn directly
setup_adk_logger(logging.INFO)
@@ -93,8 +139,10 @@ def to_a2a(
agent_executor=agent_executor, task_store=task_store
)
# Build agent card
# Use provided agent card or build one from the agent
rpc_url = f"{protocol}://{host}:{port}/"
provided_agent_card = _load_agent_card(agent_card)
card_builder = AgentCardBuilder(
agent=agent,
rpc_url=rpc_url,
@@ -105,12 +153,15 @@ def to_a2a(
# Add startup handler to build the agent card and configure A2A routes
async def setup_a2a():
# Build the agent card asynchronously
agent_card = await card_builder.build()
# Use provided agent card or build one asynchronously
if provided_agent_card is not None:
final_agent_card = provided_agent_card
else:
final_agent_card = await card_builder.build()
# Create the A2A Starlette application
a2a_app = A2AStarletteApplication(
agent_card=agent_card,
agent_card=final_agent_card,
http_handler=request_handler,
)