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)

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)

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)

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)

__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)

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)

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

(no docstring)

open_view (method)

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)

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

(no docstring)

search (method)

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

(no docstring)

LocalWorkspaceView (class)

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)

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

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

delete (method)

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

(no docstring)

patches (method)

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

(no docstring)

read (method)

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

(no docstring)

write (method)

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

(no docstring)

WorkspaceAccess (class)

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)

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)

none() -> "'WorkspaceAccess'"

(no docstring)

WorkspaceClient (class)

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)

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)

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)

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

(no docstring)

open_view (method)

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)

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

(no docstring)

search (method)

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)

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)

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)

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)

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)

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

(no docstring)

patches (method)

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

(no docstring)

read (method)

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

(no docstring)

write (method)

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

(no docstring)

Source: apps/a2a/a2a_pack/workspace.py