a2a_pack.a2a_client
Agent-to-agent invocation surface available via ctx.call(...).
An agent never speaks raw HTTP to another agent. It calls
ctx.call(target, skill, args, grant=...) and the runtime-attached
:class:A2AClient handles transport: HTTP for cross-pod, in-memory for
local tests, anything else (gRPC, message bus) for future runtimes.
The grant token (see :mod:a2a_pack.grants) is the only way to hand
workspace access across agents. Callee-side runtime validates it before
materializing a :class:WorkspaceClient.
A2AClient (class)
A2AClient()Transport-shaped agent-to-agent client.
call (method)
call(self, target: 'str', skill: 'str', *, args: 'dict[str, Any] | None' = None, grant: 'str | None' = None, timeout: 'float | None' = None) -> 'CallResult'Invoke skill on target and return its :class:CallResult.
target is opaque to this layer — for the HTTP impl it's an agent
URL; for the in-memory impl it's an agent name.
CallResult (class)
CallResult(result: 'Any', events: 'tuple[dict[str, Any], ...]' = (), artifacts: 'tuple[dict[str, Any], ...]' = (), grant_id: 'str | None' = None) -> NoneWhat an A2A invocation returns to the calling skill.
__init__ (method)
__init__(self, result: 'Any', events: 'tuple[dict[str, Any], ...]' = (), artifacts: 'tuple[dict[str, Any], ...]' = (), grant_id: 'str | None' = None) -> NoneInitialize self. See help(type(self)) for accurate signature.
HttpA2AClient (class)
HttpA2AClient(default_timeout: 'float' = 60.0) -> NoneA2A client that POSTs to the standard /invoke/{skill} endpoint.
__init__ (method)
__init__(self, default_timeout: 'float' = 60.0) -> NoneInitialize self. See help(type(self)) for accurate signature.
call (method)
call(self, target: 'str', skill: 'str', *, args: 'dict[str, Any] | None' = None, grant: 'str | None' = None, timeout: 'float | None' = None) -> 'CallResult'Invoke skill on target and return its :class:CallResult.
target is opaque to this layer — for the HTTP impl it's an agent
URL; for the in-memory impl it's an agent name.
InMemoryA2AClient (class)
InMemoryA2AClient(agents: "dict[str, 'A2AAgent']", ctx_factory: 'Any' = None) -> NoneRoutes calls to agent instances registered by name.
The receiving agent gets a new :class:RunContext built by the
ctx_factory callable, so caller and callee don't share state.
Pass ctx_factory=lambda agent, grant: ... to control how scoped
workspaces / sandboxes are wired in.
__init__ (method)
__init__(self, agents: "dict[str, 'A2AAgent']", ctx_factory: 'Any' = None) -> NoneInitialize self. See help(type(self)) for accurate signature.
call (method)
call(self, target: 'str', skill: 'str', *, args: 'dict[str, Any] | None' = None, grant: 'str | None' = None, timeout: 'float | None' = None) -> 'CallResult'Invoke skill on target and return its :class:CallResult.
target is opaque to this layer — for the HTTP impl it's an agent
URL; for the in-memory impl it's an agent name.
Source: apps/a2a/a2a_pack/a2a_client.py