test/a2a/test_card.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0
from pathlib import Path
from a2a.client.client_factory import TransportProtocol
from a2a.types import (
AgentCard,
AgentProvider,
AgentSkill,
ClientCredentialsOAuthFlow,
HTTPAuthSecurityScheme,
OAuthFlows,
SecurityScheme,
)
from a2a.utils.proto_utils import validate_proto_required_fields
from ag2 import Agent
from ag2.a2a import build_card
from ag2.a2a.security import bearer_scheme, oauth2_scheme, require
from ag2.testing import TestConfig
from ag2.tools.skills import LocalRuntime, SkillsToolkit
def _agent() -> Agent:
return Agent("agent-x", config=TestConfig("ok"))
class TestCapabilities:
def test_push_notifications_default_false(self) -> None:
card = build_card(_agent(), url="http://test")
assert card.capabilities.push_notifications is False
def test_push_notifications_flag_propagates(self) -> None:
card = build_card(_agent(), url="http://test", push_notifications=True)
assert card.capabilities.push_notifications is True
def test_streaming_always_true(self) -> None:
card = build_card(_agent(), url="http://test")
assert card.capabilities.streaming is True
def test_round_trip_via_protobuf(self) -> None:
card = build_card(_agent(), url="http://test", push_notifications=True)
decoded = AgentCard.FromString(card.SerializeToString())
assert decoded.capabilities.push_notifications is True
class TestSecurity:
def test_no_security_by_default(self) -> None:
card = build_card(_agent(), url="http://test")
assert dict(card.security_schemes) == {}
assert list(card.security_requirements) == []
def test_schemes_auto_derived_from_requirements(self) -> None:
bearer = bearer_scheme(name="bearer", description="JWT auth")
card = build_card(_agent(), url="http://test", security=[require(bearer)])
assert dict(card.security_schemes) == {
"bearer": SecurityScheme(
http_auth_security_scheme=HTTPAuthSecurityScheme(
scheme="bearer",
bearer_format="JWT",
description="JWT auth",
),
),
}
assert list(card.security_requirements) == [require(bearer).to_proto()]
def test_round_trip_preserves_schemes(self) -> None:
bearer = bearer_scheme(name="bearer")
card = build_card(_agent(), url="http://test", security=[require(bearer)])
decoded = AgentCard.FromString(card.SerializeToString())
assert decoded == card
def test_oauth2_scoped_requirement(self) -> None:
flows = OAuthFlows(client_credentials=ClientCredentialsOAuthFlow(token_url="https://x/token"))
oauth = oauth2_scheme(name="oauth", flows=flows)
card = build_card(
_agent(),
url="http://test",
security=[require(oauth.with_scopes("read", "write"))],
)
assert list(card.security_requirements) == [require(oauth.with_scopes("read", "write")).to_proto()]
assert dict(card.security_schemes) == {"oauth": oauth.scheme}
def test_scheme_deduped_across_requirements(self) -> None:
bearer = bearer_scheme(name="bearer")
oauth = oauth2_scheme(name="oauth", flows=OAuthFlows())
card = build_card(
_agent(),
url="http://test",
security=[
require(bearer),
require(oauth.with_scopes("read")),
require(bearer, oauth.with_scopes("write")),
],
)
assert set(card.security_schemes.keys()) == {"bearer", "oauth"}
assert len(card.security_requirements) == 3
class TestProviderAndBranding:
def test_provider_passthrough(self) -> None:
provider = AgentProvider(organization="AG2ai", url="https://ag2.dev")
card = build_card(_agent(), url="http://test", provider=provider)
assert card.provider == provider
def test_documentation_and_icon_urls(self) -> None:
card = build_card(
_agent(),
url="http://test",
documentation_url="https://docs.example/agent-x",
icon_url="https://docs.example/agent-x.png",
)
assert card.documentation_url == "https://docs.example/agent-x"
assert card.icon_url == "https://docs.example/agent-x.png"
def test_omitted_by_default(self) -> None:
card = build_card(_agent(), url="http://test")
assert card.documentation_url == ""
assert card.icon_url == ""
assert card.provider == AgentProvider()
class TestSkills:
def test_default_single_skill_from_agent(self) -> None:
card = build_card(_agent(), url="http://test")
assert list(card.skills) == [
AgentSkill(id="agent-x", name="agent-x", description="agent-x", tags=["agent-x"]),
]
def test_custom_skills_replace_default(self) -> None:
skills = [
AgentSkill(
id="search",
name="Search",
description="Search the web",
tags=["search", "web"],
examples=["Find recent papers on X"],
input_modes=["text/plain"],
output_modes=["application/json"],
),
AgentSkill(id="summarize", name="Summarize", description="Summarize docs"),
]
card = build_card(_agent(), url="http://test", skills=skills)
assert list(card.skills) == skills
def test_empty_skills_list_replaces_default(self) -> None:
card = build_card(_agent(), url="http://test", skills=[])
assert list(card.skills) == []
class TestRequiredFields:
def test_default_card_sets_every_required_field(self) -> None:
# Spec §5.7 / a2a.proto: AgentCard.description and AgentSkill.tags are REQUIRED.
card = build_card(_agent(), url="http://test")
validate_proto_required_fields(card)
def test_card_from_skills_toolkit_sets_every_required_field(self, local_skills_dir: Path) -> None:
agent = Agent(
"agent-x",
config=TestConfig("ok"),
tools=[SkillsToolkit(LocalRuntime(str(local_skills_dir)))],
)
card = build_card(agent, url="http://test")
validate_proto_required_fields(card)
class TestSkillsAutoDiscovery:
def test_skills_picked_up_from_skills_toolkit(self, local_skills_dir: Path) -> None:
agent = Agent(
"agent-x",
config=TestConfig("ok"),
tools=[SkillsToolkit(LocalRuntime(str(local_skills_dir)))],
)
card = build_card(agent, url="http://test")
assert list(card.skills) == [
AgentSkill(
id="code-review", name="code-review", description="Review code for bugs and style", tags=["code-review"]
),
AgentSkill(
id="data-analysis",
name="data-analysis",
description="Analyse CSV/JSON datasets",
tags=["data-analysis"],
),
]
def test_explicit_skills_override_auto_discovery(self, local_skills_dir: Path) -> None:
agent = Agent(
"agent-x",
config=TestConfig("ok"),
tools=[SkillsToolkit(LocalRuntime(str(local_skills_dir)))],
)
override = [AgentSkill(id="override", name="override", description="custom")]
card = build_card(agent, url="http://test", skills=override)
assert list(card.skills) == override
def test_falls_back_to_default_skill_when_no_toolkit(self) -> None:
agent = Agent("agent-x", config=TestConfig("ok"))
card = build_card(agent, url="http://test")
assert list(card.skills) == [
AgentSkill(id="agent-x", name="agent-x", description="agent-x", tags=["agent-x"]),
]
def test_falls_back_when_toolkit_has_no_skills(self, tmp_path: Path) -> None:
agent = Agent(
"agent-x",
config=TestConfig("ok"),
tools=[SkillsToolkit(LocalRuntime(str(tmp_path)))],
)
card = build_card(agent, url="http://test")
assert list(card.skills) == [
AgentSkill(id="agent-x", name="agent-x", description="agent-x", tags=["agent-x"]),
]
class TestInterfaceTenants:
def test_no_tenant_by_default(self) -> None:
card = build_card(_agent(), url="http://test")
[iface] = list(card.supported_interfaces)
assert iface.tenant == ""
def test_tenant_propagates_to_jsonrpc_interface(self) -> None:
card = build_card(_agent(), url="http://test", tenants={"jsonrpc": "tenant-A"})
[iface] = list(card.supported_interfaces)
assert iface.tenant == "tenant-A"
assert iface.protocol_binding == TransportProtocol.JSONRPC.value
def test_tenant_per_transport(self) -> None:
card = build_card(
_agent(),
url="http://j",
transports=("jsonrpc", "rest", "grpc"),
rest_url="http://r",
grpc_url="grpc.example:50051",
tenants={"jsonrpc": "t-j", "grpc": "t-g"},
)
tenants = {iface.protocol_binding: iface.tenant for iface in card.supported_interfaces}
assert tenants == {
TransportProtocol.JSONRPC.value: "t-j",
TransportProtocol.HTTP_JSON.value: "",
TransportProtocol.GRPC.value: "t-g",
}