a2a_pack.context

Runtime context handed to skill handlers.

The same agent code runs unchanged on local dev, Docker, Kubernetes, and hosted runtimes — the runtime provides a concrete :class:RunContext that implements artifact storage, secret access, streaming, and cancellation.

AgentEvent (class)

AgentEvent(kind: 'str', payload: 'dict[str, Any]' = <factory>) -> None

A structured event emitted during a skill run.

__init__ (method)

__init__(self, kind: 'str', payload: 'dict[str, Any]' = <factory>) -> None

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

ArtifactRef (class)

ArtifactRef(name: 'str', uri: 'str', mime_type: 'str', size_bytes: 'int') -> None

Opaque handle to a stored artifact (blob, file, etc.).

__init__ (method)

__init__(self, name: 'str', uri: 'str', mime_type: 'str', size_bytes: 'int') -> None

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

CancelledByCaller (class)

Raised by :meth:RunContext.check_cancelled when the caller cancelled.

LocalRunContext (class)

LocalRunContext(*, auth: 'AuthT', task_id: 'str' = 'local-task', secrets: 'dict[str, str] | None' = None, workspace: 'WorkspaceClient | None' = None, sandbox: 'SandboxClient | None' = None, a2a: 'A2AClient | None' = None, discover: 'DiscoveryClient | None' = None, on_event: 'Any | None' = None) -> 'None'

In-memory context for local dev and tests.

Stores events and artifacts in lists/dicts. Secrets come from a plain mapping. Cancellation is driven by an :class:asyncio.Event.

__init__ (method)

__init__(self, *, auth: 'AuthT', task_id: 'str' = 'local-task', secrets: 'dict[str, str] | None' = None, workspace: 'WorkspaceClient | None' = None, sandbox: 'SandboxClient | None' = None, a2a: 'A2AClient | None' = None, discover: 'DiscoveryClient | None' = None, on_event: 'Any | None' = None) -> 'None'

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

cancel (method)

cancel(self) -> 'None'

(no docstring)

check_cancelled (method)

check_cancelled(self) -> 'None'

Raise :class:CancelledByCaller if the caller cancelled.

emit_event (method)

emit_event(self, event: 'AgentEvent') -> 'None'

Publish a structured event to subscribers (UI, logs, traces).

secret (method)

secret(self, name: 'str') -> 'str'

Look up a runtime-injected secret by logical name.

write_artifact (method)

write_artifact(self, name: 'str', data: 'bytes', mime_type: 'str') -> 'ArtifactRef'

Persist data as a named artifact and return a reference.

MissingScopes (class)

MissingScopes(missing: 'Sequence[str]') -> 'None'

Raised by :meth:RunContext.require_scopes when caller lacks scopes.

__init__ (method)

__init__(self, missing: 'Sequence[str]') -> 'None'

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

RunContext (class)

RunContext()

Per-invocation context.

A new context is constructed by the runtime for every skill call. It carries caller identity (auth), the task identity, and runtime capabilities (artifacts, secrets, streaming, cancellation).

Agents must depend only on this abstract interface, never on a concrete runtime implementation.

answer (method)

answer(question_id: 'str', answer: 'str') -> 'bool'

Resolve a pending :meth:ask from outside (e.g. an HTTP handler).

ask (method)

ask(self, prompt: 'str', *, timeout: 'float' = 180.0) -> 'str'

Pause the skill until the caller answers a free-text question.

Emits an event with kind="question". The runtime is responsible for routing the answer back via :meth:answer. If no answer arrives within timeout seconds, raises :class:asyncio.TimeoutError.

call (method)

call(self, target: 'str', skill: 'str', *, args: 'dict[str, Any] | None' = None, grant: 'str | None' = None, timeout: 'float | None' = None) -> 'CallResult'

Invoke another agent's skill via the runtime's :class:A2AClient.

target is whatever the underlying client expects — an HTTP URL for :class:HttpA2AClient, an agent name for in-process routing. Pair with :meth:WorkspaceClient.delegate to hand a scoped workspace grant to the callee.

check_cancelled (method)

check_cancelled(self) -> 'None'

Raise :class:CancelledByCaller if the caller cancelled.

deny_scope (method)

deny_scope(request_id: 'str', reason: 'str') -> 'bool'

Resolve a pending :meth:request_scope with a deny reason.

emit_artifact (method)

emit_artifact(self, ref: 'ArtifactRef') -> 'None'

Notify subscribers that a new artifact is available.

emit_error (method)

emit_error(self, message: 'str', *, code: 'str | None' = None) -> 'None'

Emit a structured error event (does not raise).

emit_event (method)

emit_event(self, event: 'AgentEvent') -> 'None'

Publish a structured event to subscribers (UI, logs, traces).

emit_progress (method)

emit_progress(self, message: 'str') -> 'None'

Emit a human-readable progress event.

emit_text_delta (method)

emit_text_delta(self, text: 'str') -> 'None'

Emit a streamed token chunk (for LLM-style streaming output).

request_scope (method)

request_scope(self, *, reason: 'str', read: 'Sequence[str]' = (), write_prefix: 'str | None' = None, ttl_seconds: 'int' = 60, mode: 'str' = 'read_only', timeout: 'float' = 60.0) -> 'Grant'

Ask the platform for a scope expansion mid-skill.

Pauses execution until the platform replies. On approve, the new :class:Grant is verified, installed on ctx.workspace, and returned. On deny, raises :class:ScopeDenied.

The skill must opt in by declaring allow_scope_expansion=True on its @skill decorator — otherwise raises :class:ScopeExpansionNotAllowed.

require_scopes (method)

require_scopes(self, required: 'Sequence[str]') -> 'None'

Raise :class:MissingScopes if self.auth lacks any required scope.

Auth models without a scopes attribute (e.g. :class:NoAuth) are treated as having an empty scope set.

resolve_scope_grant (method)

resolve_scope_grant(request_id: 'str', grant_token: 'str') -> 'bool'

Resolve a pending :meth:request_scope with a fresh signed grant.

Called by the runtime adapter when the platform POSTs the new grant back to the agent (see serve/asgi.py's /scope-grants/{id}).

secret (method)

secret(self, name: 'str') -> 'str'

Look up a runtime-injected secret by logical name.

write_artifact (method)

write_artifact(self, name: 'str', data: 'bytes', mime_type: 'str') -> 'ArtifactRef'

Persist data as a named artifact and return a reference.

ScopeDenied (class)

Raised by :meth:RunContext.request_scope when the platform refuses.

ScopeExpansionNotAllowed (class)

Raised when a skill calls request_scope without opting in via @skill.

Source: apps/a2a/a2a_pack/context.py