---

# Page: (home)

# a2a — Docs

One Python class becomes a deployable, discoverable, sandboxed agent
service. Agents call each other through scoped grants. Users never see
Docker, Kubernetes, Gitea, or ArgoCD.

```bash
pip install a2a-pack
a2a init research-agent
cd research-agent
a2a deploy
# → https://research-agent.a2acloud.io
```

## What's here

- **[Quickstart](/quickstart)** — install, scaffold, deploy in 90 seconds.
- **[Concepts](/concepts)** — agents, skills, grants, sandbox, marketplace.
- **[Reference](/reference)** — every public symbol in `a2a_pack` + every `a2a` CLI command, auto-generated from source.
- **[Examples](/examples)** — copy-paste-able agents.
- **[For LLMs](/llms.txt)** — index file you can feed to a coding agent.
- **[Full corpus](/llms-full.txt)** — every doc page concatenated into one plaintext blob.

## The shape of an agent

```python
from pydantic import BaseModel
from a2a_pack import A2AAgent, NoAuth, RunContext, skill


class GreeterConfig(BaseModel):
    suffix: str = "!"


class Greeter(A2AAgent[GreeterConfig, NoAuth]):
    name = "greeter"
    description = "Say hi."
    version = "0.1.0"
    config_model = GreeterConfig
    auth_model = NoAuth

    @skill(description="Greet someone.")
    async def greet(self, ctx: RunContext[NoAuth], who: str) -> str:
        await ctx.emit_progress(f"greeting {who}")
        return f"hello {who}{self.config.suffix}"
```

That's it. `a2a deploy` ships it. Other agents can discover and call your
`greet` skill with a scoped grant; the CP mints + verifies; matplotlib /
pandas / whatever can run in a sandboxed microVM if your skill asks for it.



---

# Page: concepts/agents

# Agents & skills

An a2a agent is one Python class that subclasses
[`A2AAgent`](/reference/agent) and exposes one or more `@skill`-decorated
methods.

## Required class attributes

```python
class MyAgent(A2AAgent[MyConfig, MyAuth]):
    name = "my-agent"        # globally unique on the platform
    description = "..."
    version = "0.1.0"        # semver, surfaces on Card + dashboard
    config_model = MyConfig  # Pydantic; passed as `self.config`
    auth_model = MyAuth      # Pydantic; bound to `ctx.auth` per call
```

## Skills

Decorate `async` methods. The first parameter (after `self`) must be a
[`RunContext`](/reference/context). The rest must be type-annotated;
their types become the skill's input JSON Schema.

```python
@skill(
    description="Generate a chart.",
    tags=["visualization"],
    timeout_seconds=120,
    allow_scope_expansion=False,
)
async def chart(
    self,
    ctx: RunContext[NoAuth],
    prompt: str,
    data_path: str | None = None,
) -> dict:
    await ctx.emit_progress(f"working on: {prompt}")
    return {"ok": True, "path": "outputs/chart.png"}
```

The decorator:

- captures the signature with `inspect`
- builds a `pydantic.TypeAdapter` per parameter
- emits a JSON Schema with `additionalProperties: false` and
  `required` populated from non-default args
- publishes the resulting `SkillCard` on your agent's `AgentCard`

## Input validation, two levels

1. **Pydantic at the agent.** Every `/invoke/{skill}` call runs your
   args through `adapter.validate_python(...)` before the handler
   fires. Bad types → HTTP 400 (`SkillInputError`).
2. **Strict tool schema at the LLM.** The orchestrator can register
   each skill as an OpenAI-strict function tool — the model gateway
   refuses to emit calls that don't conform.

## Optional opt-ins

| Attribute / flag                    | What it enables                                            |
|-------------------------------------|------------------------------------------------------------|
| `@skill(allow_scope_expansion=True)`| Skill may call `ctx.request_scope(...)` mid-flight         |
| `@skill(stream=True)`               | Skill emits multiple events; the platform SSE-bridges them |
| `workspace_access = WorkspaceAccess.dynamic(...)` | The platform attaches a workspace client to `ctx` |
| `egress = EgressPolicy(allow_hosts=...)` | Limit outbound network from the sandbox             |
| `tools_used = ("sandbox", "matplotlib")` | Surfaces on the Card; informational                |

## Lifecycle

`A2AAgent.startup(ctx)` and `A2AAgent.shutdown(ctx)` run once each per
process. Use for warm-up (load a model, hydrate a cache) or graceful
shutdown.

## Reference

- [`A2AAgent`](/reference/agent)
- [`@skill`](/reference/agent)
- [`SkillSpec`, `SkillPolicy`](/reference/agent)



---

# Page: concepts/grants

# Grants — scoped, signed, time-bound

Every cross-agent call carries a **grant token** — an HMAC-signed blob
that names *who* called, *whom*, against *which bucket*, with *what
read/write patterns*, at *what mode*, for *how long*.

## Wire format

```
<base64url(json(payload))>.<base64url(hmac_sha256(secret, payload))>
```

Payload shape:

```json
{
  "grant_id": "abc12345",
  "issuer": "main-agent:user-7",
  "audience": "graph-agent",
  "bucket": "user-7-files",
  "mode": "read_write_overlay",
  "allow_patterns": ["**"],
  "deny_patterns": [],
  "outputs_prefix": "outputs/",
  "expires_at": 1717440000,
  "issued_at": 1717439700,
  "nonce": "e7f9..."
}
```

Receiving agents call [`verify_grant`](/reference/grants) — bad
signature, missing audience, or expired TTL → 403.

## Runtime scope negotiation

Skills that opt in via `@skill(allow_scope_expansion=True)` can call
`await ctx.request_scope(...)` mid-execution. The orchestrator runs
[`decide_extension`](/reference/scope_policy) — hard ceilings always
deny (cross-bucket); risk-0 read-only short-TTL auto-approves in
auto-mode; risk-2 (mode upgrade / new write_prefix) always asks the
user.

The flow surfaces on the dashboard as a `ScopeCard` with reason +
diff (old vs requested patterns) + approve/deny buttons.

## Audit trail

Every minted grant and every extension writes a row to `GrantAudit`,
linked via `parent_grant_id`. Query the chain:

```
GET /v1/me/grants/{grant_id}
```

returns the row and every transitive extension in one shot.

## Reference

- [`a2a_pack.grants`](/reference/grants) — SDK mint + verify
- [`a2a_pack.workspace`](/reference/workspace) — `WorkspaceClient`, `install_grant`
- [Concepts: scope negotiation](/concepts/scope-negotiation) *(coming)*



---

# Page: quickstart

# Quickstart

90-second path from nothing to a deployed agent.

## Install the CLI

```bash
pip install a2a-pack
a2a --help
```

## Sign up

```bash
a2a signup --email you@example.com --password ...
a2a whoami
```

Stores credentials at `~/.a2a/credentials.json`. Default API is
`https://api.a2acloud.io` — override with `A2A_API_URL` or `--api`.

## Scaffold

```bash
a2a init research-agent
cd research-agent
```

You get:

```
research-agent/
├── agent.py            # one A2AAgent subclass + one @skill method
├── a2a.yaml            # name, version, entrypoint
└── requirements.txt
```

## Run locally

```bash
a2a card                  # print the Card JSON the platform will see
a2a run --entrypoint agent:ResearchAgent
```

## Deploy

```bash
a2a deploy
```

Behind the scenes the CP:
- tarballs your source
- pushes it to a gitea repo it manages for you
- triggers a Gitea Actions build (Dockerfile FROM `a2a-pack-base`)
- pushes the image to the cluster registry
- bumps `deploy/20-deployment.yaml`, ArgoCD reconciles
- returns the public URL when it's live

Standard output:

```
shipped
  agent:   research-agent
  version: 0.1.0
  status:  building
  url:     https://research-agent.a2acloud.io
waiting for https://research-agent.a2acloud.io ...
live: https://research-agent.a2acloud.io
```

## What you got

- Public URL with TLS
- `/healthz`, `/.well-known/agent-card`, `/invoke/{skill}` (SSE-capable)
- Auto-derived JSON Schema input validation per skill (pydantic
  `TypeAdapter`, `additionalProperties: false`)
- MCP-compatible exposure
- Discoverable from any other agent on the platform via
  `discover_agent` + `call_agent` (HMAC-signed grant)

## Next

- [Concepts: agents + skills](/concepts/agents)
- [Concepts: grants](/concepts/grants)
- [Reference: `a2a_pack`](/reference)



---

# Page: reference/a2a_client

# `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)*

```python
A2AClient()
```

Transport-shaped agent-to-agent client.

### `call`  *(method)*

```python
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)*

```python
CallResult(result: 'Any', events: 'tuple[dict[str, Any], ...]' = (), artifacts: 'tuple[dict[str, Any], ...]' = (), grant_id: 'str | None' = None) -> None
```

What an A2A invocation returns to the calling skill.

### `__init__`  *(method)*

```python
__init__(self, result: 'Any', events: 'tuple[dict[str, Any], ...]' = (), artifacts: 'tuple[dict[str, Any], ...]' = (), grant_id: 'str | None' = None) -> None
```

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

## `HttpA2AClient`  *(class)*

```python
HttpA2AClient(default_timeout: 'float' = 60.0) -> None
```

A2A client that POSTs to the standard /invoke/{skill} endpoint.

### `__init__`  *(method)*

```python
__init__(self, default_timeout: 'float' = 60.0) -> None
```

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

### `call`  *(method)*

```python
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)*

```python
InMemoryA2AClient(agents: "dict[str, 'A2AAgent']", ctx_factory: 'Any' = None) -> None
```

Routes 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)*

```python
__init__(self, agents: "dict[str, 'A2AAgent']", ctx_factory: 'Any' = None) -> None
```

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

### `call`  *(method)*

```python
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`*


---

# Page: reference/agent

# `a2a_pack.agent`



## `A2AAgent`  *(class)*

```python
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)*

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

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

### `card`  *(method)*

```python
card(self) -> 'AgentCard'
```

*(no docstring)*

### `health`  *(method)*

```python
health(self) -> 'bool'
```

Lightweight liveness check. Override to add real probes.

### `invoke`  *(method)*

```python
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)*

```python
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)*

```python
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)*

```python
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)*

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

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

### `startup`  *(method)*

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

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

## `ParamSpec`  *(class)*

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

Validation metadata for a single skill parameter.

### `__init__`  *(method)*

```python
__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)*

```python
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)*

```python
__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)*

```python
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`*


---

# Page: reference/auth

# `a2a_pack.auth`

Pluggable auth principal models.

These describe *who* is invoking a skill. The runtime auth provider produces
an instance of the agent's declared ``auth_model`` and hands it to the
:class:`RunContext`.

## `APIKeyAuth`  *(class)*

```python
APIKeyAuth(*, api_key_id: str, scopes: list[str] = <factory>) -> None
```

Caller authenticated by a long-lived API key.

## `JWTAuth`  *(class)*

```python
JWTAuth(*, sub: str, org_id: str | None = None, email: str | None = None, scopes: list[str] = <factory>) -> None
```

Caller authenticated by a JWT (typically from a user-facing login).

## `NoAuth`  *(class)*

```python
NoAuth() -> None
```

Public agent: no caller identity required.


*Source: `apps/a2a/a2a_pack/auth.py`*


---

# Page: reference/card

# `a2a_pack.card`



## `AgentCard`  *(class)*

```python
AgentCard(*, name: str, description: str, version: str, skills: list[a2a_pack.card.SkillCard], capabilities: dict[str, typing.Any] = <factory>, input_modes: list[str] = <factory>, output_modes: list[str] = <factory>, required_secrets: list[str] = <factory>, required_env: list[str] = <factory>, runtime: a2a_pack.runtime.AgentRuntime = <factory>, state_schema: dict[str, typing.Any] | None = None, workspace_access: a2a_pack.workspace.WorkspaceAccess = <factory>) -> None
```

Public description of an agent.

Mirrors the A2A Agent Card spec: identity, capabilities, IO modes, and
the catalog of skills the agent advertises.

### `from_agent`  *(method)*

```python
from_agent(agent: "'A2AAgent'") -> "'AgentCard'"
```

*(no docstring)*

## `SkillCard`  *(class)*

```python
SkillCard(*, id: str, name: str, description: str, tags: list[str] = <factory>, scopes: list[str] = <factory>, stream: bool = False, policy: a2a_pack.runtime.SkillPolicy = <factory>, input_schema: dict[str, typing.Any], output_schema: dict[str, typing.Any]) -> None
```

Public description of a single skill, shaped for the A2A spec.


*Source: `apps/a2a/a2a_pack/card.py`*


---

# Page: reference/cli

# `a2a` CLI

Build, package, and deploy A2A agents.

## Top-level help

```
Usage: a2a [OPTIONS] COMMAND [ARGS]...                                         
                                                                                
 Build, package, and deploy A2A agents.                                         
                                                                                
╭─ Options ────────────────────────────────────────────────────────────────────╮
│ --help          Show this message and exit.                                  │
╰──────────────────────────────────────────────────────────────────────────────╯
╭─ Commands ───────────────────────────────────────────────────────────────────╮
│ signup    Create an account on the control plane and store the JWT locally.  │
│ login     Authenticate with the control plane and cache the JWT.             │
│ logout    Forget the cached JWT.                                             │
│ whoami    Show the currently logged-in user.                                 │
│ agents    List agents visible to the current user.                           │
│ init      Scaffold a new agent project.                                      │
│ validate  Load the agent and print its Card schema. Exits non-zero on        │
│           errors.                                                            │
│ card      Print the Agent Card JSON for the project's agent.                 │
│ run       Run the agent's HTTP server locally (used inside the container     │
│           too).                                                              │
│ build     Build (and optionally push) the container image for the agent.     │
│ deploy    Ship the agent.                                                    │
╰──────────────────────────────────────────────────────────────────────────────╯
```



---

# Page: reference/context

# `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)*

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

A structured event emitted during a skill run.

### `__init__`  *(method)*

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

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

## `ArtifactRef`  *(class)*

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

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

### `__init__`  *(method)*

```python
__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)*

```python
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)*

```python
__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)*

```python
cancel(self) -> 'None'
```

*(no docstring)*

### `check_cancelled`  *(method)*

```python
check_cancelled(self) -> 'None'
```

Raise :class:`CancelledByCaller` if the caller cancelled.

### `emit_event`  *(method)*

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

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

### `secret`  *(method)*

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

Look up a runtime-injected secret by logical name.

### `write_artifact`  *(method)*

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

Persist ``data`` as a named artifact and return a reference.

## `MissingScopes`  *(class)*

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

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

### `__init__`  *(method)*

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

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

## `RunContext`  *(class)*

```python
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)*

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

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

### `ask`  *(method)*

```python
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)*

```python
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)*

```python
check_cancelled(self) -> 'None'
```

Raise :class:`CancelledByCaller` if the caller cancelled.

### `deny_scope`  *(method)*

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

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

### `emit_artifact`  *(method)*

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

Notify subscribers that a new artifact is available.

### `emit_error`  *(method)*

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

Emit a structured error event (does not raise).

### `emit_event`  *(method)*

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

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

### `emit_progress`  *(method)*

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

Emit a human-readable progress event.

### `emit_text_delta`  *(method)*

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

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

### `request_scope`  *(method)*

```python
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)*

```python
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)*

```python
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)*

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

Look up a runtime-injected secret by logical name.

### `write_artifact`  *(method)*

```python
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`*


---

# Page: reference/discovery

# `a2a_pack.discovery`

Agent discovery surface available via ``ctx.discover``.

Agents find each other by capability/tag/skill — *never* by hardcoded URL.
The runtime attaches a :class:`DiscoveryClient`; the canonical impl
queries the platform's agent registry (control plane).

## `ControlPlaneDiscovery`  *(class)*

```python
ControlPlaneDiscovery(api_url: 'str', *, token: 'str | None' = None, timeout: 'float' = 10.0) -> 'None'
```

Hits the platform's agent registry (control plane).

### `__init__`  *(method)*

```python
__init__(self, api_url: 'str', *, token: 'str | None' = None, timeout: 'float' = 10.0) -> 'None'
```

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

### `find_agents`  *(method)*

```python
find_agents(self, *, tags: 'Sequence[str]' = (), capability: 'str | None' = None, skill: 'str | None' = None, limit: 'int' = 10) -> 'list[DiscoveredAgent]'
```

*(no docstring)*

### `get_agent`  *(method)*

```python
get_agent(self, name: 'str') -> 'DiscoveredAgent'
```

*(no docstring)*

## `DiscoveredAgent`  *(class)*

```python
DiscoveredAgent(name: 'str', url: 'str | None', card: 'AgentCard') -> None
```

A registry hit. ``url`` is what the caller hands to ``ctx.call``.

### `__init__`  *(method)*

```python
__init__(self, name: 'str', url: 'str | None', card: 'AgentCard') -> None
```

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

## `DiscoveryClient`  *(class)*

```python
DiscoveryClient()
```

Discovery surface.

### `find_agents`  *(method)*

```python
find_agents(self, *, tags: 'Sequence[str]' = (), capability: 'str | None' = None, skill: 'str | None' = None, limit: 'int' = 10) -> 'list[DiscoveredAgent]'
```

*(no docstring)*

### `get_agent`  *(method)*

```python
get_agent(self, name: 'str') -> 'DiscoveredAgent'
```

*(no docstring)*

## `InMemoryDiscovery`  *(class)*

```python
InMemoryDiscovery(agents: 'dict[str, DiscoveredAgent]') -> 'None'
```

Discovery surface.

### `__init__`  *(method)*

```python
__init__(self, agents: 'dict[str, DiscoveredAgent]') -> 'None'
```

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

### `find_agents`  *(method)*

```python
find_agents(self, *, tags: 'Sequence[str]' = (), capability: 'str | None' = None, skill: 'str | None' = None, limit: 'int' = 10) -> 'list[DiscoveredAgent]'
```

*(no docstring)*

### `get_agent`  *(method)*

```python
get_agent(self, name: 'str') -> 'DiscoveredAgent'
```

*(no docstring)*


*Source: `apps/a2a/a2a_pack/discovery.py`*


---

# Page: reference/grants

# `a2a_pack.grants`

Signed grant tokens for cross-agent workspace handoff.

A grant is a small, self-contained, signed claim issued by one agent that
the platform (or the receiving agent) can verify without a registry round-trip.

Wire format::

    "<base64url(json(payload))>.<base64url(hmac_sha256(secret, payload))>"

The payload describes *what* the callee is allowed to do, *whose* workspace
they can see, and *for how long*. The runtime on the receiving side
materializes a :class:`WorkspaceClient` scoped to that grant.

Auth model is intentionally simple for v1: a shared platform secret signs
every grant. Swap for asymmetric (X.509 / JWKS) when crossing trust domains.

## `Grant`  *(class)*

```python
Grant(*, grant_id: str, issuer: str, audience: str, bucket: str, mode: a2a_pack.workspace.WorkspaceMode = <WorkspaceMode.READ_ONLY: 'read_only'>, allow_patterns: tuple[str, ...] = ('**',), deny_patterns: tuple[str, ...] = (), outputs_prefix: str | None = None, expires_at: typing.Annotated[int, Ge(ge=0)] = 0, issued_at: typing.Annotated[int, Ge(ge=0)] = 0, nonce: str = <factory>) -> None
```

The payload of a signed grant token.

A grant binds *who* (issuer) gave *whom* (audience) access to *which*
workspace files (bucket + allow/deny patterns) under *what* mode and
*how long*. The runtime enforces every line of this payload.

## `GrantInvalid`  *(class)*



Raised by :func:`verify_grant` when a grant is bad/expired/forged.

## `mint_grant`  *(function)*

```python
mint_grant(*, issuer: 'str', audience: 'str', bucket: 'str', mode: 'WorkspaceMode' = <WorkspaceMode.READ_ONLY: 'read_only'>, allow_patterns: 'tuple[str, ...]' = ('**',), deny_patterns: 'tuple[str, ...]' = (), outputs_prefix: 'str | None' = None, ttl_seconds: 'int' = 300, secret: 'bytes | None' = None) -> 'tuple[Grant, str]'
```

Build a :class:`Grant` and return it together with its signed token.

## `sign_grant`  *(function)*

```python
sign_grant(grant: 'Grant', *, secret: 'bytes | None' = None) -> 'str'
```

*(no docstring)*

## `verify_grant`  *(function)*

```python
verify_grant(token: 'str', *, secret: 'bytes | None' = None) -> 'Grant'
```

Parse + verify ``token``. Raises :class:`GrantInvalid` on any failure.

Checks signature, expiry, and minimal structural shape. Caller-specific
audience checks are layered on top by the server adapter.


*Source: `apps/a2a/a2a_pack/grants.py`*


---

# Page: reference/runtime

# `a2a_pack.runtime`

Declarative runtime/deployment metadata.

These types describe *how* the platform should run an agent: lifecycle,
state needs, isolation level, resource budget, egress policy. They are
read by the deployer and by the registry; agent code itself should not
depend on which runtime is selected.

## `AgentRuntime`  *(class)*

```python
AgentRuntime(*, lifecycle: a2a_pack.runtime.Lifecycle = <Lifecycle.EPHEMERAL: 'ephemeral'>, state: a2a_pack.runtime.State = <State.NONE: 'none'>, sandbox: a2a_pack.runtime.Sandbox = <Sandbox.MICROSANDBOX: 'microsandbox'>, resources: a2a_pack.runtime.Resources = <factory>, concurrency: typing.Annotated[int, Gt(gt=0)] = 1, egress: a2a_pack.runtime.EgressPolicy = <factory>, tools_used: tuple[str, ...] = ()) -> None
```

Aggregate runtime declaration; published on the Agent Card.

## `EgressPolicy`  *(class)*

```python
EgressPolicy(*, allow_hosts: tuple[str, ...] = (), allow_internal_services: tuple[str, ...] = (), deny_internet_by_default: bool = True) -> None
```

What external hosts the agent is allowed to talk to.

## `Lifecycle`  *(class)*

```python
Lifecycle(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)
```

How long an instance of the agent process lives.

## `Resources`  *(class)*

```python
Resources(*, cpu: str = '100m', memory: str = '256Mi', gpu: typing.Annotated[int, Ge(ge=0)] = 0, max_runtime_seconds: typing.Annotated[int, Gt(gt=0)] = 600) -> None
```

Resource budget hint for the deployer.

## `Sandbox`  *(class)*

```python
Sandbox(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)
```

Isolation level. The platform always runs agents under microsandbox.

Modeled as an enum (rather than a constant) so the wire format stays
stable if more isolation tiers are added later, but only one value is
currently valid: every agent runs in a microvm-class sandbox.

## `SkillPolicy`  *(class)*

```python
SkillPolicy(*, timeout_seconds: float | None = None, idempotent: bool = False, max_retries: typing.Annotated[int, Ge(ge=0)] = 0, cost_class: str | None = None, allow_scope_expansion: bool = False) -> None
```

Per-skill operational policy advertised on the Agent Card.

## `State`  *(class)*

```python
State(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)
```

What kind of state the agent retains between invocations.


*Source: `apps/a2a/a2a_pack/runtime.py`*


---

# Page: reference/sandbox

# `a2a_pack.sandbox`

Code-execution sandbox surface available to agents via ``ctx.sandbox``.

The abstract :class:`SandboxClient` is what agent code programs against. The
runtime layer (host-side microsandbox + FUSE-mounted MinIO, in-cluster
DaemonSet, hosted SaaS) supplies a concrete implementation.

The sandbox is **general-purpose code execution**, not Python-only. Agents
can:

  * run arbitrary shell pipelines: ``await ctx.sandbox.run_shell("git clone … && cargo build")``
  * exec a binary with explicit args (no shell parsing): ``await sb.exec("/usr/bin/git", ["clone", url])``
  * pick any OCI image: ``run_shell("npx @openai/codex …", image="node:20-slim")``

``run_python`` is just a convenience for the common Python-snippet case.

Why an abstract here when ``microsandbox`` itself already has a Python SDK?
The platform owns the *policy* layer — bucket selection, network egress,
write-path restrictions, resource caps, audit logging. Agents must depend on
the policy-respecting surface, not on the raw SDK, so the same agent code
runs unchanged across local dev / cluster / hosted environments.

## `ExecResult`  *(class)*

```python
ExecResult(stdout: 'str', stderr: 'str' = '', exit_code: 'int' = 0, truncated: 'bool' = False) -> None
```

Result of a command run inside a sandbox.

### `__init__`  *(method)*

```python
__init__(self, stdout: 'str', stderr: 'str' = '', exit_code: 'int' = 0, truncated: 'bool' = False) -> None
```

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

## `SandboxClient`  *(class)*

```python
SandboxClient()
```

Negotiation surface handed to agents via ``ctx.sandbox``.

### `create`  *(method)*

```python
create(self, spec: 'SandboxSpec') -> 'SandboxHandle'
```

*(no docstring)*

### `get`  *(method)*

```python
get(self, name: 'str') -> 'SandboxHandle'
```

*(no docstring)*

### `list`  *(method)*

```python
list(self) -> 'list[str]'
```

*(no docstring)*

### `remove`  *(method)*

```python
remove(self, name: 'str') -> 'None'
```

*(no docstring)*

### `run_python`  *(method)*

```python
run_python(self, code: 'str', *, image: 'str' = 'python:3.11-slim', **kwargs: 'Any') -> 'ExecResult'
```

Convenience: spin a one-shot sandbox, run inline Python, tear down.

Equivalent to ``create(SandboxSpec(image=image)).exec("python", ["-c", code])``.
Use the lower-level surface when you need persistence, multiple
commands, or non-Python tools.

### `run_shell`  *(method)*

```python
run_shell(self, script: 'str', *, image: 'str' = 'python:3.11-slim', **kwargs: 'Any') -> 'ExecResult'
```

Convenience: spin a one-shot sandbox, run an arbitrary shell script,
tear down.

Pass ``image=`` to pick the toolchain (e.g. ``"node:20-slim"`` for
npm-based tools like codex, ``"rust:1-slim"`` for cargo,
``"alpine/git"`` for plain git ops). The default ``python:3.11-slim``
already has bash/coreutils/curl/git so most one-liners just work.

## `SandboxHandle`  *(class)*

```python
SandboxHandle()
```

Live handle to a running sandbox VM.

### `exec`  *(method)*

```python
exec(self, cmd: 'str', args: 'Sequence[str] | None' = None, *, timeout: 'float | None' = None) -> 'ExecResult'
```

*(no docstring)*

### `kill`  *(method)*

```python
kill(self) -> 'None'
```

*(no docstring)*

### `logs`  *(method)*

```python
logs(self, *, tail: 'int | None' = None) -> 'str'
```

*(no docstring)*

### `shell`  *(method)*

```python
shell(self, script: 'str', *, timeout: 'float | None' = None) -> 'ExecResult'
```

*(no docstring)*

### `stop`  *(method)*

```python
stop(self) -> 'None'
```

*(no docstring)*

## `SandboxSpec`  *(class)*

```python
SandboxSpec(name: 'str', image: 'str' = 'python:3.11-slim', memory_mib: 'int' = 512, cpus: 'int' = 1, workspace: 'str | None' = None, secrets: 'tuple[str, ...]' = (), egress: 'tuple[str, ...]' = (), labels: 'dict[str, str]' = <factory>) -> None
```

Caller request shape for :meth:`SandboxClient.create`.

### `__init__`  *(method)*

```python
__init__(self, name: 'str', image: 'str' = 'python:3.11-slim', memory_mib: 'int' = 512, cpus: 'int' = 1, workspace: 'str | None' = None, secrets: 'tuple[str, ...]' = (), egress: 'tuple[str, ...]' = (), labels: 'dict[str, str]' = <factory>) -> None
```

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

## `SandboxUnavailable`  *(class)*



Raised when ``ctx.sandbox`` is accessed but no runtime is attached.


*Source: `apps/a2a/a2a_pack/sandbox.py`*


---

# Page: reference/serve-asgi

# `a2a_pack.serve.asgi`

HTTP adapter that turns any :class:`A2AAgent` into a service.

This is intentionally minimal: it covers the surface needed to plug into
the wider A2A ecosystem and the platform's control plane.

Endpoints:
  GET  /healthz                     -> liveness
  GET  /.well-known/agent-card      -> Agent Card JSON
  POST /invoke/{skill}              -> invoke skill (JSON in, JSON out)

Auth: a single bearer token is read from the ``A2A_API_KEY`` env var. If set,
all routes except ``/healthz`` and the card require ``Authorization: Bearer
<key>``. The bearer token is materialized into the agent's declared
``auth_model`` (best-effort: APIKeyAuth, otherwise NoAuth).

## `build_app`  *(function)*

```python
build_app(agent: 'A2AAgent') -> 'FastAPI'
```

Build a FastAPI app for the given agent instance.

## `serve`  *(function)*

```python
serve(agent: 'A2AAgent', *, host: 'str' = '0.0.0.0', port: 'int' = 8000) -> 'None'
```

Run the agent's HTTP server with uvicorn (blocking).


*Source: `apps/a2a/a2a_pack/asgi.py`*


---

# Page: reference/workspace

# `a2a_pack.workspace`

Workspace capability negotiation.

Agents never receive a filesystem path. They negotiate a *view* by intent::

    view = await ctx.workspace.open_view(
        purpose="Fix failing payment test",
        hints=["payment", "checkout"],
        file_types=["python"],
        max_files=10,
        mode=WorkspaceMode.READ_WRITE_OVERLAY,
    )
    for path in view.files:
        content = await view.read(path)

The runtime resolves the request (semantic search + dependency graph + git
metadata + policy + optional human approval) and returns a bounded grant.
Writes are staged as :class:`WorkspacePatch` objects, never applied directly
to the host filesystem from inside the sandbox.

## `FileMatch`  *(class)*

```python
FileMatch(*, path: str, file_type: a2a_pack.workspace.FileType, score: float = 0.0, summary: str | None = None, size_bytes: typing.Annotated[int, Ge(ge=0)] = 0) -> None
```

Result row from :meth:`WorkspaceClient.search`.

## `FileType`  *(class)*

```python
FileType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)
```

str(object='') -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or
errors is specified, then the object must expose a data buffer
that will be decoded using the given encoding and error handler.
Otherwise, returns the result of object.__str__() (if defined)
or repr(object).
encoding defaults to sys.getdefaultencoding().
errors defaults to 'strict'.

## `LocalWorkspaceClient`  *(class)*

```python
LocalWorkspaceClient(files: 'dict[str, bytes]', *, access: 'WorkspaceAccess', bucket: 'str' = 'local', issuer: 'str' = 'local') -> 'None'
```

In-memory workspace for local dev and tests.

Search is naive substring match; ranking is keyword-overlap. Real
runtime implementations replace this with embeddings/dep-graph search.
Policy enforcement (:class:`WorkspaceAccess`) IS applied here so tests
cover the rejection paths.

### `__init__`  *(method)*

```python
__init__(self, files: 'dict[str, bytes]', *, access: 'WorkspaceAccess', bucket: 'str' = 'local', issuer: 'str' = 'local') -> 'None'
```

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

### `install_grant`  *(method)*

```python
install_grant(self, grant: "'Grant'") -> 'None'
```

Replace this client's access policy with a superseding grant.

Called by :meth:`RunContext.request_scope` after the platform mints a
new grant covering additional read patterns / longer TTL / new write
prefix. Subsequent reads/writes use the new policy.

Default impl raises — concrete clients (LocalWorkspaceClient or the
runtime's MinIO-backed variant) override.

### `list_grants`  *(method)*

```python
list_grants(self) -> 'list[WorkspaceGrant]'
```

*(no docstring)*

### `open_view`  *(method)*

```python
open_view(self, *, purpose: 'str', hints: 'Sequence[str]' = (), file_types: 'Sequence[FileType]' = (), max_files: 'int' = 10, mode: 'WorkspaceMode' = <WorkspaceMode.READ_ONLY: 'read_only'>, reason: 'str | None' = None) -> 'WorkspaceView'
```

*(no docstring)*

### `request_access`  *(method)*

```python
request_access(self, *, files: 'Sequence[FileMatch | str]', mode: 'WorkspaceMode', reason: 'str', purpose: 'str' = '') -> 'WorkspaceGrant'
```

*(no docstring)*

### `search`  *(method)*

```python
search(self, *, query: 'str', types: 'Sequence[FileType]' = (), limit: 'int' = 20) -> 'list[FileMatch]'
```

*(no docstring)*

## `LocalWorkspaceView`  *(class)*

```python
LocalWorkspaceView(grant: 'WorkspaceGrant', client: "'LocalWorkspaceClient'") -> 'None'
```

A bounded view over a granted set of files.

Returned by :meth:`WorkspaceClient.open_view`. Reads always go to the
granted view; writes return :class:`WorkspacePatch` objects that the
runtime will commit (or reject) outside the sandbox.

### `__init__`  *(method)*

```python
__init__(self, grant: 'WorkspaceGrant', client: "'LocalWorkspaceClient'") -> 'None'
```

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

### `delete`  *(method)*

```python
delete(self, path: 'str') -> 'WorkspacePatch'
```

*(no docstring)*

### `patches`  *(method)*

```python
patches(self) -> 'tuple[WorkspacePatch, ...]'
```

*(no docstring)*

### `read`  *(method)*

```python
read(self, path: 'str') -> 'bytes'
```

*(no docstring)*

### `write`  *(method)*

```python
write(self, path: 'str', content: 'bytes') -> 'WorkspacePatch'
```

*(no docstring)*

## `WorkspaceAccess`  *(class)*

```python
WorkspaceAccess(*, enabled: bool = False, max_files: typing.Annotated[int, Ge(ge=0)] = 0, allowed_modes: tuple[a2a_pack.workspace.WorkspaceMode, ...] = (), require_reason: bool = True, deny_patterns: tuple[str, ...] = (), require_human_approval: bool = False, max_total_size_bytes: typing.Annotated[int, Gt(gt=0)] = 104857600) -> None
```

Class-level workspace policy.

Use :meth:`none` for agents that do not touch any workspace, or
:meth:`dynamic` to allow capability negotiation under bounds.

### `dynamic`  *(method)*

```python
dynamic(*, max_files: 'int' = 25, allowed_modes: 'Sequence[WorkspaceMode]' = (<WorkspaceMode.READ_ONLY: 'read_only'>,), require_reason: 'bool' = True, deny_patterns: 'Sequence[str]' = (), require_human_approval: 'bool' = False, max_total_size_bytes: 'int' = 104857600) -> "'WorkspaceAccess'"
```

*(no docstring)*

### `none`  *(method)*

```python
none() -> "'WorkspaceAccess'"
```

*(no docstring)*

## `WorkspaceClient`  *(class)*

```python
WorkspaceClient()
```

Negotiation surface handed to the agent via ``ctx.workspace``.

The concrete implementation is provided by the runtime; agents must
program against this interface only.

### `delegate`  *(method)*

```python
delegate(self, *, audience: 'str', allow_patterns: 'Sequence[str]' = ('**',), deny_patterns: 'Sequence[str]' = (), mode: 'WorkspaceMode' = <WorkspaceMode.READ_ONLY: 'read_only'>, outputs_prefix: 'str | None' = None, ttl_seconds: 'int' = 300) -> 'str'
```

Mint a signed grant token the caller can hand to ``ctx.call``.

The default implementation requires the workspace to expose
``self.bucket`` and ``self.issuer`` — override in concrete clients
that don't fit that shape.

### `install_grant`  *(method)*

```python
install_grant(self, grant: "'Grant'") -> 'None'
```

Replace this client's access policy with a superseding grant.

Called by :meth:`RunContext.request_scope` after the platform mints a
new grant covering additional read patterns / longer TTL / new write
prefix. Subsequent reads/writes use the new policy.

Default impl raises — concrete clients (LocalWorkspaceClient or the
runtime's MinIO-backed variant) override.

### `list_grants`  *(method)*

```python
list_grants(self) -> 'list[WorkspaceGrant]'
```

*(no docstring)*

### `open_view`  *(method)*

```python
open_view(self, *, purpose: 'str', hints: 'Sequence[str]' = (), file_types: 'Sequence[FileType]' = (), max_files: 'int' = 10, mode: 'WorkspaceMode' = <WorkspaceMode.READ_ONLY: 'read_only'>, reason: 'str | None' = None) -> 'WorkspaceView'
```

*(no docstring)*

### `request_access`  *(method)*

```python
request_access(self, *, files: 'Sequence[FileMatch | str]', mode: 'WorkspaceMode', reason: 'str', purpose: 'str' = '') -> 'WorkspaceGrant'
```

*(no docstring)*

### `search`  *(method)*

```python
search(self, *, query: 'str', types: 'Sequence[FileType]' = (), limit: 'int' = 20) -> 'list[FileMatch]'
```

*(no docstring)*

## `WorkspaceDenied`  *(class)*



Raised when a workspace request violates the agent's policy.

## `WorkspaceGrant`  *(class)*

```python
WorkspaceGrant(*, grant_id: str, purpose: str, files: tuple[a2a_pack.workspace.FileMatch, ...], mode: a2a_pack.workspace.WorkspaceMode, reason: str, expires_at: datetime.datetime | None = None, requires_human_approval: bool = False) -> None
```

An approved access grant for a bounded set of files.

## `WorkspaceMode`  *(class)*

```python
WorkspaceMode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)
```

str(object='') -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or
errors is specified, then the object must expose a data buffer
that will be decoded using the given encoding and error handler.
Otherwise, returns the result of object.__str__() (if defined)
or repr(object).
encoding defaults to sys.getdefaultencoding().
errors defaults to 'strict'.

## `WorkspacePatch`  *(class)*

```python
WorkspacePatch(*, grant_id: str, path: str, operation: Literal['create', 'update', 'delete'], content: bytes | None = None) -> None
```

A staged write. Not applied until the runtime/approver commits it.

## `WorkspaceView`  *(class)*

```python
WorkspaceView()
```

A bounded view over a granted set of files.

Returned by :meth:`WorkspaceClient.open_view`. Reads always go to the
granted view; writes return :class:`WorkspacePatch` objects that the
runtime will commit (or reject) outside the sandbox.

### `delete`  *(method)*

```python
delete(self, path: 'str') -> 'WorkspacePatch'
```

*(no docstring)*

### `patches`  *(method)*

```python
patches(self) -> 'tuple[WorkspacePatch, ...]'
```

*(no docstring)*

### `read`  *(method)*

```python
read(self, path: 'str') -> 'bytes'
```

*(no docstring)*

### `write`  *(method)*

```python
write(self, path: 'str', content: 'bytes') -> 'WorkspacePatch'
```

*(no docstring)*


*Source: `apps/a2a/a2a_pack/workspace.py`*