a2a docsreferenceagent

a2a_pack.agent

A2AAgent (class)

A2AAgent(config: 'ConfigT | dict[str, Any] | None' = None) -> 'None'

Base class for A2A agents.

Subclasses declare:

  • name, description (and optional version),
  • optional config_model / auth_model (default to empty / NoAuth),
  • deployment metadata: required_secrets, required_env, capabilities, input_modes, output_modes,
  • one or more methods decorated with :func:skill.

__init__ (method)

__init__(self, config: 'ConfigT | dict[str, Any] | None' = None) -> 'None'

Initialize self. See help(type(self)) for accurate signature.

card (method)

card(self) -> 'AgentCard'

(no docstring)

health (method)

health(self) -> 'bool'

Lightweight liveness check. Override to add real probes.

invoke (method)

invoke(self, skill_name: 'str', ctx: 'RunContext[AuthT]', /, **kwargs: 'Any') -> 'Any'

Invoke a skill with caller-supplied kwargs.

Inputs are validated and coerced via the skill's pydantic schema. Required scopes are enforced against ctx.auth before the handler runs. The raw handler return value is returned (Python-typed).

invoke_json (method)

invoke_json(self, skill_name: 'str', ctx: 'RunContext[AuthT]', payload: 'dict[str, Any]') -> 'Any'

Runtime-facing invoke: takes JSON-shaped payload, returns JSON-shaped result.

local_invoke (method)

local_invoke(self, skill_name: 'str', /, *, auth: 'AuthT | None' = None, secrets: 'dict[str, str] | None' = None, task_id: 'str' = 'local-task', workspace: 'Any' = None, sandbox: 'Any' = None, a2a: 'Any' = None, discover: 'Any' = None, **kwargs: 'Any') -> 'Any'

Convenience harness: build a :class:LocalRunContext and invoke.

Useful in tests and notebooks. Pass workspace=, sandbox=, a2a=, and/or discover= to bind concrete runtime clients.

runtime (method)

runtime() -> 'AgentRuntime'

Aggregate the class-level runtime declaration.

sandbox is always :attr:Sandbox.MICROSANDBOX; it is set here rather than on the class so developers cannot weaken isolation.

shutdown (method)

shutdown(self, ctx: 'RunContext[AuthT]') -> 'None'

Called once before the agent process exits. Override to tear down.

startup (method)

startup(self, ctx: 'RunContext[AuthT]') -> 'None'

Called once before the first invocation. Override to set up state.

ParamSpec (class)

ParamSpec(name: 'str', adapter: 'TypeAdapter[Any]', has_default: 'bool', default: 'Any' = None) -> None

Validation metadata for a single skill parameter.

__init__ (method)

__init__(self, name: 'str', adapter: 'TypeAdapter[Any]', has_default: 'bool', default: 'Any' = None) -> None

Initialize self. See help(type(self)) for accurate signature.

SkillInputError (class)

Raised when invoke() inputs fail validation against the skill schema.

SkillInvocationError (class)

Raised when a skill handler raises during invoke().

SkillNotFound (class)

Raised when invoke() is called with an unknown skill name.

SkillSpec (class)

SkillSpec(name: 'str', description: 'str', tags: 'tuple[str, ...]', scopes: 'tuple[str, ...]', stream: 'bool', policy: 'SkillPolicy', input_schema: 'dict[str, Any]', output_schema: 'dict[str, Any]', handler: 'Callable[..., Awaitable[Any]]', params: 'tuple[ParamSpec, ...]' = <factory>, output_adapter: 'TypeAdapter[Any] | None' = None) -> None

Static metadata about a single skill, captured at decoration time.

__init__ (method)

__init__(self, name: 'str', description: 'str', tags: 'tuple[str, ...]', scopes: 'tuple[str, ...]', stream: 'bool', policy: 'SkillPolicy', input_schema: 'dict[str, Any]', output_schema: 'dict[str, Any]', handler: 'Callable[..., Awaitable[Any]]', params: 'tuple[ParamSpec, ...]' = <factory>, output_adapter: 'TypeAdapter[Any] | None' = None) -> None

Initialize self. See help(type(self)) for accurate signature.

skill (function)

skill(*, name: 'str | None' = None, description: 'str' = '', tags: 'Sequence[str]' = (), scopes: 'Sequence[str]' = (), stream: 'bool' = False, timeout_seconds: 'float | None' = None, idempotent: 'bool' = False, max_retries: 'int' = 0, cost_class: 'str | None' = None, allow_scope_expansion: 'bool' = False) -> 'Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]'

Mark an :class:A2AAgent method as a discoverable skill.

Conventions:

  • The handler MUST be async def.
  • Its first parameter (after self) MUST be a :class:RunContext; the context is supplied by the runtime and is omitted from the published input schema.
  • Remaining parameters MUST be type-annotated. *args and **kwargs are rejected.

Source: apps/a2a/a2a_pack/agent.py