Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 17 additions & 18 deletions samples/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,18 +11,16 @@

from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.types import Message, Part, Role, SendMessageRequest, TaskState
from a2a.utils import get_artifact_text, get_message_text
from a2a.utils.agent_card import display_agent_card


async def _handle_stream( # noqa: PLR0912
async def _handle_stream(
stream: Any, current_task_id: str | None
) -> str | None:
async for event in stream:
if event.HasField('message'):
print('Message:', end=' ')
for part in event.message.parts:
if part.text:
print(part.text, end=' ')
print()
print('Message:', get_message_text(event.message, delimiter=' '))
return None

if not current_task_id:
Expand All @@ -35,12 +33,15 @@ async def _handle_stream( # noqa: PLR0912

if event.HasField('status_update'):
state_name = TaskState.Name(event.status_update.status.state)
print(f'TaskStatusUpdate [state={state_name}]:', end=' ')
if event.status_update.status.HasField('message'):
for part in event.status_update.status.message.parts:
if part.text:
print(part.text, end=' ')
print()
message_text = (
': '
Comment thread
sokoliva marked this conversation as resolved.
+ get_message_text(
event.status_update.status.message, delimiter=' '
)
if event.status_update.status.HasField('message')
else ''
)
print(f'TaskStatusUpdate [state={state_name}]{message_text}')
if state_name in (
'TASK_STATE_COMPLETED',
'TASK_STATE_FAILED',
Expand All @@ -52,12 +53,10 @@ async def _handle_stream( # noqa: PLR0912
elif event.HasField('artifact_update'):
print(
f'TaskArtifactUpdate [name={event.artifact_update.artifact.name}]:',
end=' ',
get_artifact_text(
event.artifact_update.artifact, delimiter=' '
),
)
for part in event.artifact_update.artifact.parts:
if part.text:
print(part.text, end=' ')
print()
return current_task_id


Expand Down Expand Up @@ -86,7 +85,7 @@ async def main() -> None:
resolver = A2ACardResolver(httpx_client, args.url)
card = await resolver.get_agent_card()
print('\n✓ Agent Card Found:')
print(f' Name: {card.name}')
display_agent_card(card)

client = await create_client(card, client_config=config)

Expand Down
2 changes: 2 additions & 0 deletions src/a2a/utils/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Utility functions for the A2A Python SDK."""

from a2a.utils import proto_utils
from a2a.utils.agent_card import display_agent_card
from a2a.utils.artifact import (
get_artifact_text,
new_artifact,
Expand Down Expand Up @@ -44,6 +45,7 @@
'build_text_artifact',
'completed_task',
'create_task_obj',
'display_agent_card',
'get_artifact_text',
'get_data_parts',
'get_file_parts',
Expand Down
76 changes: 76 additions & 0 deletions src/a2a/utils/agent_card.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
"""Utility functions for inspecting AgentCard instances."""

from a2a.types.a2a_pb2 import AgentCard


def display_agent_card(card: AgentCard) -> None:
Comment thread
sokoliva marked this conversation as resolved.
"""Print a human-readable summary of an AgentCard to stdout.

Args:
card: The AgentCard proto message to display.
"""
width = 52
Comment thread
sokoliva marked this conversation as resolved.
sep = '=' * width
thin = '-' * width

lines: list[str] = [sep, 'AgentCard'.center(width), sep]

lines += [
'--- General ---',
f'Name : {card.name}',
f'Description : {card.description}',
f'Version : {card.version}',
]
if card.documentation_url:
lines.append(f'Docs URL : {card.documentation_url}')
if card.icon_url:
lines.append(f'Icon URL : {card.icon_url}')
if card.HasField('provider'):
url_suffix = f' ({card.provider.url})' if card.provider.url else ''
lines.append(f'Provider : {card.provider.organization}{url_suffix}')

lines += ['', '--- Interfaces ---']
for i, iface in enumerate(card.supported_interfaces):
binding = f'{iface.protocol_binding} {iface.protocol_version}'.strip()
parts = [
p
for p in [binding, f'tenant={iface.tenant}' if iface.tenant else '']
if p
]
suffix = f' ({", ".join(parts)})' if parts else ''
line = f' [{i}] {iface.url}{suffix}'
lines.append(line)

lines += [
'',
'--- Capabilities ---',
f'Streaming : {card.capabilities.streaming}',
f'Push notifications : {card.capabilities.push_notifications}',
f'Extended agent card : {card.capabilities.extended_agent_card}',
]

lines += [
'',
'--- I/O Modes ---',
f'Input : {", ".join(card.default_input_modes) or "(none)"}',
f'Output : {", ".join(card.default_output_modes) or "(none)"}',
]

lines += ['', '--- Skills ---']
if card.skills:
for skill in card.skills:
lines += [
thin,
f' ID : {skill.id}',
f' Name : {skill.name}',
f' Description : {skill.description}',
f' Tags : {", ".join(skill.tags) or "(none)"}',
]
if skill.examples:
for ex in skill.examples:
lines.append(f' Example : {ex}')
else:
lines.append(' (none)')

lines.append(sep)
print('\n'.join(lines))
Comment thread
sokoliva marked this conversation as resolved.
194 changes: 194 additions & 0 deletions tests/utils/test_agent_card_display.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,194 @@
"""Tests for display_agent_card utility."""

import pytest

from a2a.types.a2a_pb2 import (
AgentCapabilities,
AgentCard,
AgentInterface,
AgentProvider,
AgentSkill,
)
from a2a.utils.agent_card import display_agent_card


@pytest.fixture
def full_agent_card() -> AgentCard:
return AgentCard(
name='Sample Agent',
description='A sample agent.',
version='1.0.0',
documentation_url='https://docs.example.com',
icon_url='https://example.com/icon.png',
provider=AgentProvider(
organization='Example Org', url='https://example.com'
),
supported_interfaces=[
AgentInterface(
url='http://localhost:9999/a2a/jsonrpc',
protocol_binding='JSONRPC',
protocol_version='1.0',
),
AgentInterface(
url='http://localhost:9999/a2a/rest',
protocol_binding='HTTP+JSON',
protocol_version='1.0',
tenant='tenant-a',
),
],
capabilities=AgentCapabilities(
streaming=True,
push_notifications=False,
extended_agent_card=True,
),
default_input_modes=['text'],
default_output_modes=['text', 'task-status'],
skills=[
AgentSkill(
id='skill-1',
name='My Skill',
description='Does something useful.',
tags=['foo', 'bar'],
examples=['Do the thing', 'Another example'],
),
AgentSkill(
id='skill-2',
name='Other Skill',
description='Does something else.',
tags=['baz'],
),
],
)


class TestDisplayAgentCard:
def test_full_card_output(
self, full_agent_card: AgentCard, capsys: pytest.CaptureFixture[str]
) -> None:
"""Golden test: exact output for a fully-populated card."""
display_agent_card(full_agent_card)
assert capsys.readouterr().out == (
'====================================================\n'
' AgentCard \n'
'====================================================\n'
'--- General ---\n'
'Name : Sample Agent\n'
'Description : A sample agent.\n'
'Version : 1.0.0\n'
'Docs URL : https://docs.example.com\n'
'Icon URL : https://example.com/icon.png\n'
'Provider : Example Org (https://example.com)\n'
'\n'
'--- Interfaces ---\n'
' [0] http://localhost:9999/a2a/jsonrpc (JSONRPC 1.0)\n'
' [1] http://localhost:9999/a2a/rest (HTTP+JSON 1.0, tenant=tenant-a)\n'
'\n'
'--- Capabilities ---\n'
'Streaming : True\n'
'Push notifications : False\n'
'Extended agent card : True\n'
'\n'
'--- I/O Modes ---\n'
'Input : text\n'
'Output : text, task-status\n'
'\n'
'--- Skills ---\n'
'----------------------------------------------------\n'
' ID : skill-1\n'
' Name : My Skill\n'
' Description : Does something useful.\n'
' Tags : foo, bar\n'
' Example : Do the thing\n'
' Example : Another example\n'
'----------------------------------------------------\n'
' ID : skill-2\n'
' Name : Other Skill\n'
' Description : Does something else.\n'
' Tags : baz\n'
'====================================================\n'
)

def test_empty_card_output(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""Golden test: exact output for a card with only default/empty fields.

An empty supported_interfaces section signals a malformed card —
the bare header with no entries is intentional and visible to the user.
"""
display_agent_card(AgentCard())
assert capsys.readouterr().out == (
'====================================================\n'
' AgentCard \n'
'====================================================\n'
'--- General ---\n'
'Name : \n'
'Description : \n'
'Version : \n'
'\n'
'--- Interfaces ---\n'
'\n'
'--- Capabilities ---\n'
'Streaming : False\n'
'Push notifications : False\n'
'Extended agent card : False\n'
'\n'
'--- I/O Modes ---\n'
'Input : (none)\n'
'Output : (none)\n'
'\n'
'--- Skills ---\n'
' (none)\n'
'====================================================\n'
)

def test_interface_without_protocol_version_has_no_trailing_space(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""No trailing space in the binding field when protocol_version is not set."""
card = AgentCard(
supported_interfaces=[
AgentInterface(
url='127.0.0.1:50051',
protocol_binding='GRPC',
)
]
)
display_agent_card(card)
assert ' [0] 127.0.0.1:50051 (GRPC)' in capsys.readouterr().out

def test_interface_without_binding_or_version_has_no_parentheses(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""No parentheses when neither protocol_binding nor protocol_version are set."""
card = AgentCard(
supported_interfaces=[AgentInterface(url='127.0.0.1:50051')]
)
display_agent_card(card)
assert ' [0] 127.0.0.1:50051\n' in capsys.readouterr().out

def test_provider_with_url(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""Provider shows organization and URL in parentheses when both are set."""
card = AgentCard(
provider=AgentProvider(
organization='Example Org',
url='https://example.com',
),
)
display_agent_card(card)
assert (
'Provider : Example Org (https://example.com)'
in capsys.readouterr().out
)

def test_provider_without_url_has_no_empty_parentheses(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""No empty parentheses when provider URL is not set."""
card = AgentCard(provider=AgentProvider(organization='Example Org'))
display_agent_card(card)
out = capsys.readouterr().out
assert 'Provider : Example Org' in out
assert '()' not in out
Loading