Skip to content

blockether.vis.extension

Tool declarations, host APIs and extension lifecycle contracts.

For authoring workflows and examples, use the extension tutorial and extension API guide.

@runtime_checkable
class Host(typing.Protocol):

Operations every injected or outside vis host must implement.

Host(*args, **kwargs)
def workspace_root(self) -> str:

Read the active session working copy, or the outside process directory.

def project_root(self) -> str:

Read this extension's project in the session working copy, or the outside process directory.

def python_environment(self) -> str:

Read the project's python.environment: project or shared.

def workspace_files(self) -> list[str] | None:

List the files that Git sees in the working copy, or None without Git.

def setting_declaration(self, spec: Mapping[str, typing.Any]) -> Mapping[str, typing.Any]:

Validate a declaration and fill defaults from the contract schema.

def setting(self, id: str, default: ~_SettingValue) -> ~_SettingValue:

Read one setting from the current callback snapshot, as the type of default.

def state_get(self, key: str) -> Any:

Read one value out of the extension's durable state.

def state_put(self, key: str, value: Any) -> Any:

Write one JSON value into the extension's durable state.

def state_del(self, key: str) -> Any:

Drop one key from the extension's durable state.

def state_keys(self) -> Any:

List every key the extension's durable state holds.

def log(self, level: str, message: str) -> Any:

Emit one engine log line at a level.

def notify(self, text: str, level: str) -> Any:

Show one notification on the user's channel.

def request(self, frame: str) -> str:

Send one SDK request frame as JSON. Return the reply envelope as JSON.

def bound_session(self) -> str | None:

Return the ID of the bound Vis session, or None.

def shell(self, options: Mapping[str, typing.Any]) -> Mapping[str, typing.Any]:

Run one shell operation. Return the contract's result shape.

def jailed_shell(self, options: Mapping[str, typing.Any]) -> Mapping[str, typing.Any]:

Run one shell op inside the workspace jail.

def request_input( self, request_json: str, validator_arities_json: str, run_validator: Callable[[str, int, str, str], str]) -> str:

Ask the human, and block until the answer settles or is cancelled.

def live(self, envelope_json: str, listener: Any = None) -> str:

Open, patch, read or close a live view.

On open, listener(event) gets each human event of a panel view.

state accepts timeout_ms (0..86400000) and after_seq (nonnegative). A positive timeout blocks until the sequence differs or the view closes. An unchanged timeout returns is_open=True, timed_out=True, without view. Ordinary state reads and changed waits return the current view with seq.

def activity(self, presentation: dict[str, typing.Any]) -> bool:

Replace the running symbol's headline, summary, content and sections.

def reveal_secret(self, handle: str) -> Any:

Resolve a vis-secret: handle to its plaintext.

def forget_secret(self, handle: str) -> Any:

Drop the plaintext a secret handle stands for.

def declare_env(self, declarations_json: str) -> str:

Resolve the environment variables the extension declared.

def call_tool( self, tool: str, args: Sequence[typing.Any], kwargs: Mapping[str, typing.Any]) -> Any:

Run one active session tool by the name python_execution calls.

def read_session(self, target: str | None) -> str:

Read one whole session as the sandbox read_session does. Return JSON.

def list_sessions(self, search: str | None) -> str:

List sessions as the sandbox list_sessions does. Return JSON.

def attach(self, options: Mapping[str, typing.Any]) -> str:

Store one artifact of the running block. Return its descriptor as JSON.

def list_attachments(self) -> str:

List the session's artifact descriptors as JSON, pending ones included.

def get_attachment(self, target: str, version: int | None) -> str:

Return the descriptor of one artifact as JSON.

def read_attachment(self, target: str, version: int | None) -> str:

Return the bytes of one artifact as base64 text.

def show_attachment(self, target: str, version: int | None) -> Sequence[typing.Any]:

Queue a stored image for the next model request: id, filename, type, size.

class Sessions:

Create, find and read Vis sessions from an extension, as vis.sessions.

Each method returns blockether.vis.engine.Session handles that use the host. The host records this extension as the origin of each session that it creates. An extension can change only its bound session, its child sessions and its own sessions.

def create( self, *, provider: str | None = None, model: str | None = None, **options) -> Any:

Create an independent session and return its Session.

Without root, the session starts in the root of the bound session. The other options are the options of ExecutionLayer.create_session. Only an extension callback can create a session.

def get(self, session_id: str) -> Any:

Return the Session handle of session_id without a request.

def models( self, *, provider: str | None = None, is_available: bool | None = None) -> tuple[typing.Any, ...]:

Return the Model records that the model pickers show, in picker order.

The filters are those of ExecutionLayer.models. A thread of your extension can also call it.

def list(self, search: str | None = None) -> 'list[dict[str, Any]]':

Return the session records of the first page, or the matches of search.

def export(self, session_id: str | None = None) -> bytes:

Return a .vissnap snapshot of a session and its children as bytes.

Without session_id, export the bound session. See Session.export.

def restore( self, data: bytes, *, root: str | None = None, group_id: str | None = None, key: str | None = None) -> Any:

Restore snapshot data as a new session and return its Session.

The arguments are those of ExecutionLayer.restore_session. Only an extension callback can restore a session.

sessions = <Sessions object>
def run( request: str, *, provider: str | None = None, model: str | None = None, timeout: float = 300, response_model: Any = None, max_corrections: int = 2, **options) -> Any:

Create an independent session, run one turn in it and return the result.

The arguments are those of blockether.vis.engine.Agent.run. provider and model pin the model of the new session. Without response_model, the result is the turn record; check its status. Only an extension callback can call it.

tools = <vis tools>
@dataclass(frozen=True, slots=True)
class ActivityText(_ActivityBlock):

Plain Activity text, never a model message or a question.

ActivityText(text: str)
text: str
type: ClassVar[str] = 'text'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityHeading(ActivityText):

Heading for a section of this invocation's current content.

ActivityHeading(text: str)
type: ClassVar[str] = 'heading'
Inherited methods and attributes

Inherited from ActivityText.text.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityMarkdown(ActivityText):

Markdown content for this invocation.

ActivityMarkdown(text: str)
type: ClassVar[str] = 'markdown'
Inherited methods and attributes

Inherited from ActivityText.text.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityCode(ActivityText):

Code with an optional language hint.

ActivityCode(text: str, language: str | None = None)
language: str | None
type: ClassVar[str] = 'code'
Inherited methods and attributes

Inherited from ActivityText.text.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityDiff(ActivityCode):

A diff, not engine-observed file-change evidence.

ActivityDiff(text: str, language: str | None = None)
type: ClassVar[str] = 'diff'
Inherited methods and attributes

Inherited from ActivityCode.language.

Inherited from ActivityText.text.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityTable(_ActivityBlock):

A rectangular table. Input lists are snapshotted as tuples.

paths is optional and, when given, holds one workspace path per row so clients can open that row's file. Use an empty string for a row without one.

ActivityTable( columns: tuple[str, ...], rows: tuple[tuple[str, ...], ...], paths: tuple[str, ...] | None = None)
columns: tuple[str, ...]
rows: tuple[tuple[str, ...], ...]
paths: tuple[str, ...] | None
type: ClassVar[str] = 'table'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityFile(_ActivityBlock):

Reference an existing attachment, never an external media URL.

ActivityFile(attachment_id: str, label: str)
attachment_id: str
label: str
type: ClassVar[str] = 'file'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityImage(ActivityFile):

Display an existing image attachment.

ActivityImage(attachment_id: str, label: str)
type: ClassVar[str] = 'image'
Inherited methods and attributes

Inherited from ActivityFile.attachment_id.

Inherited from ActivityFile.label.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityVideo(ActivityFile):

Display an existing video attachment.

ActivityVideo(attachment_id: str, label: str)
type: ClassVar[str] = 'video'
Inherited methods and attributes

Inherited from ActivityFile.attachment_id.

Inherited from ActivityFile.label.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityAudio(ActivityFile):

Display an existing audio attachment.

ActivityAudio(attachment_id: str, label: str)
type: ClassVar[str] = 'audio'
Inherited methods and attributes

Inherited from ActivityFile.attachment_id.

Inherited from ActivityFile.label.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityProgress(_ActivityBlock):

Progress with paired value/total, or an indeterminate indicator when omitted.

ActivityProgress( label: str, value: int | float | None = None, total: int | float | None = None)
label: str
value: int | float | None
total: int | float | None
type: ClassVar[str] = 'progress'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

@dataclass(frozen=True, slots=True)
class ActivityTerminalLine:

One output line of an ActivityTerminal. Vis shows "stderr" lines as errors.

ActivityTerminalLine(text: str, stream: Literal['stdout', 'stderr'] = 'stdout')
text: str
stream: Literal['stdout', 'stderr']
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh terminal line in contract form.

@dataclass(frozen=True, slots=True)
class ActivityTerminal(_ActivityBlock):

A command and its output, shown like a terminal.

command holds one or more stages. Start each later stage with its operator, for example ("cd app", "&& npm test", "| tail -8"). Vis highlights the operators. lines holds the output in order; a plain string is a stdout line. Keep the end of long output and count the dropped lines in omitted_lines. Remove ANSI escape codes first. exit is the exit code of a finished command.

ActivityTerminal( command: tuple[str, ...], lines: tuple[ActivityTerminalLine, ...] = (), omitted_lines: int | None = None, exit: int | None = None)
command: tuple[str, ...]
lines: tuple[ActivityTerminalLine, ...]
omitted_lines: int | None
exit: int | None
type: ClassVar[str] = 'terminal'
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh Activity content block in contract form.

Host lifecycle fields are not included.

SymbolTag: TypeAlias = Literal['observation', 'mutation', 'verification', 'external']
@dataclass(frozen=True, slots=True)
class ActivitySection:

A visible headline and one-line summary.

Only the content waits behind disclosure.

summary_format="markdown" opts into inline Markdown with HTTP(S) links. Unmarked or "inline" summaries are literal text. The 512-byte, one-line limit includes Markdown source. Images, HTML and block layout are not shown.

ActivitySection( headline: str, summary: str, content: tuple[ActivityText | ActivityHeading | ActivityMarkdown | ActivityCode | ActivityDiff | ActivityTable | ActivityFile | ActivityImage | ActivityVideo | ActivityAudio | ActivityProgress | ActivityTerminal, ...] = (), *, summary_format: Optional[Literal['inline', 'markdown']] = None)
headline: str
summary: str
summary_format: Optional[Literal['inline', 'markdown']]
def to_wire(self) -> dict[str, typing.Any]:

Return fresh portable data, without engine-owned lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityPresentation(ActivitySection):

One atomic symbol presentation, optionally linked to a persistent handle.

Use the same non-secret handle_id for later receipts of one operation in this extension and Python form. Vis shows one current outcome, and it keeps distinct details, errors and every invocation in history. Omit it for independent calls. A handle is not a label, tool argument or return value.

Set verdict when the call finished but reports on checks: "failed" for failing tests or lint findings, "passed" when every check passed. Vis pins a failed verdict in the settled turn's digest until a later run of the same tool passes.

outcome is one short result line. Vis shows it after an arrow on the closed row, for example the last line of command output. aside is short muted context at the right edge of the row, for example a directory or "exit 1".

ActivityPresentation( headline: str, summary: str, content: tuple[ActivityText | ActivityHeading | ActivityMarkdown | ActivityCode | ActivityDiff | ActivityTable | ActivityFile | ActivityImage | ActivityVideo | ActivityAudio | ActivityProgress | ActivityTerminal, ...] = (), sections: tuple[ActivitySection, ...] = (), *, summary_format: Optional[Literal['inline', 'markdown']] = None, handle_id: str | None = None, verdict: Optional[Literal['passed', 'failed']] = None, outcome: str | None = None, aside: str | None = None)
sections: tuple[ActivitySection, ...]
handle_id: str | None
verdict: Optional[Literal['passed', 'failed']]
outcome: str | None
aside: str | None
def to_wire(self) -> dict[str, typing.Any]:

Return fresh portable data, without engine-owned lifecycle fields.

Inherited methods and attributes

Inherited from ActivitySection.headline.

Inherited from ActivitySection.summary.

Inherited from ActivitySection.content.

@dataclass(frozen=True, slots=True)
class Activity:

Human-facing symbol presentation. The engine owns identity, timing and outcome.

Declare Activity on every exported callable, including each object method. Write understandable English for people, not Python identifiers or object reprs. Labels and headlines use sentence case ("Read file", "Run tests") and keep proper names and acronyms. Summaries explain the target, useful counts or outcome. Use render or publish_activity for selected content. Return values stay independent.

show_start=False makes a fast operation end-only: no running row or start callback. Use it for quick reads, patches and local lookups. Keep show_start=True for work people wait for, such as tests, network requests or transfers. Internal start/end tracking still preserves ordering, timing, errors and cancellation. Published content is kept but stays hidden until an end-only invocation settles.

The final presentation must stand alone: name the target and outcome, including empty results or failures reported as return values. A normally returned failed workflow is still a failed workflow. "Completed" alone is not enough. Label partial lists and excerpts, and keep useful counts, errors and changes.

render is an optional synchronous callback. It receives phase, args, kwargs, result and error as keyword arguments. It returns an ActivityPresentation, or None to keep the current presentation. It runs on success and failure, and on start only when show_start=True. Use publish_activity for intermediate stages of long-running tools. Rendering failures never change returns or errors.

Arguments:
  • presenter: Presentation category. Leave "generic" for ordinary tools.
  • label: Optional nonblank, single-line label of at most 96 characters.
  • render: Synchronous callback accepting phase, args, kwargs, result and error keyword arguments. Return an ActivityPresentation or None.
  • show_start: Whether readers see a running row before the call settles.
Raises:
  • TypeError: show_start is not boolean or render is not synchronous/callable.
  • ValueError: The presenter is unknown or the label is invalid.
Activity( presenter: str = 'generic', label: str | None = None, render: Callable[..., ActivityPresentation | None] | None = None, show_start: bool = True)
presenter: str
label: str | None
render: Callable[..., ActivityPresentation | None] | None
show_start: bool
def publish_activity(presentation: ActivityPresentation) -> bool:

Replace headline, summary, content and sections without authoring lifecycle.

Headline and summary stay visible when collapsed. Content uses typed text, Markdown, table, code, diff, attachment and progress blocks. Sections group multiple results with one blank line between them. To clear content, publish the same headline/summary with empty content. None from render keeps the last snapshot. Returns False outside an invocation or when publication is rejected.

@dataclass(frozen=True, slots=True, kw_only=True)
class Setting(typing.Generic[~_SettingValue]):

Declare a boolean, choice or text setting shared by the app and TUI.

scopes allows any non-empty combination of global, project, group and session. Omitting it uses the contract's global-only default. A project extension exists only in its project, so there global means the whole project. The setting shows in its extension's own settings section. value() reads the current callback's response snapshot. Its type follows default: bool for a boolean setting, str for a choice setting and for a text setting. Outside Vis, it returns the declared default. A string setting holds one line of text. is_secret=True makes it write-only: Settings show only whether a value is saved. A secret needs the default "" and only the global scope. Its value stays on this machine and out of trajectories. Read a secret while Vis calls its own extension, for example in a tool. Elsewhere, value() can return the default. parent names another setting of the same section. Settings shows this setting under that setting. Settings can nest at any depth. on_change(old, new, scope) runs once after a person changes the setting in the app or TUI and its effective value changes. scope is global, project, group or session. Vis keeps the saved value if the hook raises, and shows the error.

Setting( *, id: str, label: str, default: ~_SettingValue, type: str = 'boolean', choices: Sequence[str] = (), scopes: Sequence[str] | None = None, description: str | None = None, parent: str | None = None, on_change: Callable[[typing.Any, typing.Any, str], typing.Any] | None = None, is_secret: bool = False)
id: str
label: str
default: ~_SettingValue
type: str
choices: Sequence[str]
scopes: Sequence[str] | None
description: str | None
parent: str | None
on_change: Callable[[typing.Any, typing.Any, str], typing.Any] | None
is_secret: bool
def value(self) -> ~_SettingValue:

Read this setting in the current callback without changing any scope.

@dataclass(frozen=True, slots=True, kw_only=True)
class Extension:

Group tools and host capabilities in one immutable declaration.

Arguments:
  • name: Human-readable, nonblank extension name.
  • description: Nonblank description of what the extension does.
  • alias: Python namespace for exported symbols. Required with symbols.
  • symbols: Symbol declarations for functions or object namespaces.
  • version: Optional extension version string.
  • kind: Optional extension kind understood by the host.
  • activation: Optional host activation callback.
  • prompt: Instructions as text or a host callback.
  • slash_commands: SlashCommand declarations.
  • keybindings: Keybinding declarations: TUI keys after C-x.
  • op_hooks: OpHook declarations for host lifecycle operations.
  • ctx: Optional host context callback.
  • providers: Provider declarations.
  • network_filters: NetworkFilter declarations.
  • settings: Host-owned Setting declarations with explicit allowed scopes.
  • env: Environment variable names resolved at host registration.
  • grants: Grant declarations: toolchain paths for this extension's jailed_shell children.

Collection inputs are copied into tuples. Constructing an Extension is pure: it does not run callbacks or register anything. In an application, pass it to blockether.vis.engine.Agent or its register_extension method. In an installed extension entrypoint, call this module's register_extension once instead. The application callback bridge rejects host-only fields before connecting.

Raises:
  • ValueError: A required name, alias, callback or environment name is invalid.
  • TypeError: A field has the wrong type or a collection contains something other than its corresponding SDK declaration.
Extension( *, name: str, description: str, version: str | None = None, kind: str | None = None, alias: str | None = None, activation: Callable[..., typing.Any] | None = None, symbols: Sequence[Symbol] = (), prompt: str | Callable[..., typing.Any] | None = None, slash_commands: Sequence[SlashCommand] = (), keybindings: Sequence[Keybinding] = (), op_hooks: Sequence[OpHook] = (), ctx: Callable[..., typing.Any] | None = None, providers: Sequence[Provider] = (), network_filters: Sequence[NetworkFilter] = (), settings: Sequence[Setting] = (), env: Sequence[str] = (), grants: Sequence[Grant] = ())
name: str
description: str
version: str | None
kind: str | None
alias: str | None
activation: Callable[..., typing.Any] | None
symbols: Sequence[Symbol]
prompt: str | Callable[..., typing.Any] | None
slash_commands: Sequence[SlashCommand]
keybindings: Sequence[Keybinding]
op_hooks: Sequence[OpHook]
ctx: Callable[..., typing.Any] | None
providers: Sequence[Provider]
network_filters: Sequence[NetworkFilter]
settings: Sequence[Setting]
env: Sequence[str]
grants: Sequence[Grant]
def register_extension(extension: Extension) -> None:

Register one typed declaration and resolve its declared environment in this context.

Construction is pure, and registration is the only host boundary. A failure before completion leaves the context unregistered. No constructor registers itself.

def host_env(name, default=None):
def sequence(*, field: str) -> Callable[[~_SequenceClass], ~_SequenceClass]:

Declare a dataclass's public list or tuple field as its sandbox sequence.

Apply this outside @dataclass. The returned class is unchanged except for metadata: original methods are neither called nor transported. Each subclass must opt in separately. Values are checked when an extension returns them. The backing field must hold a built-in list or tuple, not a lazy iterable.

Sandbox records iterate over the received field and support length, truth testing, integer indices and slices. String keys still read named fields. Only the received items are exposed, and iteration never fetches another page.

def method( fn: Any = None, *, tag: Literal['observation', 'mutation', 'verification', 'external'] = 'observation', is_hidden: bool = False, activity: Activity | None = None) -> Any:

Describe a public method on an object exported through Symbol.

Arguments:
  • fn: Method to annotate. Omit it to use @method(...).
  • tag: What the method does, as Activity reports it. "observation" reads, "mutation" changes local state and "verification" checks work (tests, lint, CI status). "external" reaches people or systems beyond the session (messages, publications, remote hosts).
  • is_hidden: Hide the method from discovery without removing the callable.
  • activity: This method's human-facing presentation. Declare it on each exported method, not on the containing object namespace.
Returns:

The original method, or a decorator returning it. Calling your method directly still uses normal Python behavior. This decorator adds metadata.

Raises:
  • ValueError: The tag is unsupported or the decorated value is not callable.
  • TypeError: The Activity declaration has the wrong type.
from blockether.vis.extension import Activity, Symbol, method


class Greeter:
    @method(activity=Activity(label="Greet person", show_start=False))
    def hello(self, name: str) -> str:
        "Return a greeting for the named person."
        return f"Hello, {name}!"


greeter = Greeter()
tool = Symbol(greeter, name="greeter")
assert greeter.hello("Ada") == "Hello, Ada!"
assert tool.contract["members"][0]["name"] == "greeter.hello"
@dataclass(frozen=True, slots=True)
class Symbol:

Expose a function or an object's public methods as typed tools.

Arguments:
  • fn: Function, or an object whose public methods form a namespace.
  • name: Override the function name. Required for objects, and must then be a public Python identifier.
  • tag: Default operation classification, one of the method tags.
  • is_hidden: Hide the symbol from discovery without removing the callable.
  • activity: Human-facing presentation for a function. Declare one on every exported callable. For an object, leave this unset and put an Activity on every exported method instead.

Construction validates the declaration and derives its contract without calling the function or starting Vis. Function docstrings and type annotations become tool documentation, not runtime argument validation. Export narrow objects: public methods and nested namespace objects are traversed, not just methods bearing the decorator.

Raises:
  • ValueError: A callable lacks a docstring, or the name, operation tag, Activity placement or namespace is invalid, including cycles and repeated object references.
  • TypeError: A declaration field has an unsupported type.
Symbol( fn: Callable[..., typing.Any] | object, name: str | None = None, tag: Literal['observation', 'mutation', 'verification', 'external'] = 'observation', is_hidden: bool = False, activity: Activity | None = None)
fn: Callable[..., typing.Any] | object
name: str | None
tag: Literal['observation', 'mutation', 'verification', 'external']
is_hidden: bool
activity: Activity | None
contract: dict[str, typing.Any]

Fresh portable tool description, including literal defaults but no host access.

Namespace members carry their full public names. Strings in Annotated describe meaning. Unresolved annotations remain explicit and are never evaluated. This is documentation, not runtime argument or result validation.

@dataclass(frozen=True, slots=True)
class TypeSpec:

An inert Python type description. References bound recursive records.

TypeSpec( kind: Literal['any', 'null', 'scalar', 'opaque', 'unresolved', 'union', 'literal', 'generic', 'record', 'reference'], name: str, description: str = '', arguments: tuple[TypeSpec, ...] = (), fields: tuple[FieldSpec, ...] = (), values: tuple[str | int | bool | None, ...] = (), variadic: bool = False, sequence_field: str | None = None)
kind: Literal['any', 'null', 'scalar', 'opaque', 'unresolved', 'union', 'literal', 'generic', 'record', 'reference']
name: str
description: str
arguments: tuple[TypeSpec, ...]
fields: tuple[FieldSpec, ...]
values: tuple[str | int | bool | None, ...]
variadic: bool
sequence_field: str | None
@dataclass(frozen=True, slots=True)
class FieldSpec:

A dataclass field. Default values and factories are never exported or run.

FieldSpec( name: str, type: TypeSpec, required: bool, has_default: bool, default_is_none: bool)
name: str
type: TypeSpec
required: bool
has_default: bool
default_is_none: bool
@dataclass(frozen=True, slots=True)
class ParameterSpec(FieldSpec):

A callable parameter. default_source is inert source, or None when unavailable.

ParameterSpec( name: str, type: TypeSpec, required: bool, has_default: bool, default_is_none: bool, kind: Literal['positional_only', 'positional_or_keyword', 'var_positional', 'keyword_only', 'var_keyword'], default_source: str | None)
kind: Literal['positional_only', 'positional_or_keyword', 'var_positional', 'keyword_only', 'var_keyword']
default_source: str | None
Inherited methods and attributes

Inherited from FieldSpec.name.

Inherited from FieldSpec.type.

Inherited from FieldSpec.required.

Inherited from FieldSpec.has_default.

Inherited from FieldSpec.default_is_none.

@dataclass(frozen=True, slots=True)
class ToolSpec:

Typed view of one public callable's authoritative Symbol contract.

ToolSpec( version: int, name: str, tag: Literal['observation', 'mutation', 'verification', 'external'], description: str, signature: str, parameters: tuple[ParameterSpec, ...], returns: TypeSpec)
version: int
name: str
tag: Literal['observation', 'mutation', 'verification', 'external']
description: str
signature: str
parameters: tuple[ParameterSpec, ...]
returns: TypeSpec
@dataclass(frozen=True, slots=True)
class NamespaceSpec:

A public namespace and its callable descendants, with full dotted names.

NamespaceSpec(name: str, members: tuple[ToolSpec, ...])
name: str
members: tuple[ToolSpec, ...]
@dataclass(frozen=True, slots=True)
class HelpDocument:

Generated reference text, not captured CLI output or an operation result.

HelpDocument(tool: str, text: str)
tool: str
text: str
class Catalog:

Read-only snapshot of public Symbols. Construction and lookup do no IO.

Pass the same symbols to Catalog and Extension. Rebuild after changing declarations. This is an adapter, not a registry, dispatcher or runtime type validator.

Catalog(symbols: Sequence[Symbol])
@method(activity=Activity(label='Inspect tool catalog', show_start=False, render=_catalog_presentation))
def spec( self, name: Annotated[str | None, 'Full public name; None lists top-level tools and namespaces.'] = None) -> ToolSpec | NamespaceSpec | tuple[ToolSpec | NamespaceSpec, ...]:

Inspect declared tools without invoking them. Unknown or hidden names raise ValueError.

@method(activity=Activity(label='Read tool reference', show_start=False, render=_catalog_presentation))
def help( self, name: Annotated[str, 'Full public tool or namespace name.']) -> HelpDocument:

Render the same metadata as doc(). No configuration, authentication or operation runs.

Raises TypeError for a non-string name, and ValueError for unknown or hidden names.

@dataclass(frozen=True, slots=True)
class SlashCommand:

A user-facing slash command, not a model-facing tool.

SlashCommand( name: str, run: Callable[..., typing.Any], doc: str | None = None, usage: str | None = None, saveable: bool = False)
name: str
run: Callable[..., typing.Any]
doc: str | None
usage: str | None
saveable: bool
@dataclass(frozen=True, slots=True)
class Keybinding:

A TUI key binding: C-x, a group key and this key run a command, a verb or a function.

Arguments:
  • key: One printable ASCII character from ! to ~.
  • command: A slash command line that starts with /. The TUI runs it as if the person typed it and pressed Enter. Give it with label.
  • action: A built-in TUI verb id in kebab case, such as new-session. The verb moves to key.
  • label: The key menu and palette label. Required with command and run.
  • group: The C-x group of the key. A built-in group name, such as Session, places the key there, in any case. A command or run binding without a group uses Extensions (C-x e). Any other name makes a group of its own under C-x e, opened by the first free letter of the name. An action binding without a group stays in the group of its verb.
  • run: A function that the gateway calls when the person presses the key. It adds no turn to the transcript. It receives a dict with key, group, channel, session_id, workspace_root and project_root. It returns None, a text, vis.ok(...) or vis.err(...), and the TUI shows the title. Call vis.notify(...) in it to tell every client of the session.

Give exactly one of command, action and run.

Raises:
  • ValueError: A field breaks these rules.
Keybinding( key: str, command: str | None = None, action: str | None = None, label: str | None = None, group: str | None = None, run: Callable[..., typing.Any] | None = None)
key: str
command: str | None
action: str | None
label: str | None
group: str | None
run: Callable[..., typing.Any] | None
GATE_OPS = ('fs_access',)
@dataclass(frozen=True, slots=True)
class OpHook:

An operation observer or a fail-closed gate. Never mix both kinds in one hook.

OpHook( ops: Sequence[str], fn: Callable[..., typing.Any], phase: Literal['before', 'after'] = 'before')
ops: Sequence[str]
fn: Callable[..., typing.Any]
phase: Literal['before', 'after']
@dataclass(frozen=True, slots=True)
class NetworkFilter:

Request/response policy at the host's network-filter boundary.

NetworkFilter(fn: Callable[..., typing.Any])
fn: Callable[..., typing.Any]
@dataclass(frozen=True, slots=True)
class Grant:

Filesystem access that this extension's own jailed child processes need.

A toolchain often lives outside the workspace: a JDK, a package cache or a launcher. Declare each path once. The host checks every grant at registration. It adds the grant only to the jailed_shell children of this extension. The model's shell never receives it, and the jail deny rules still win.

Arguments:
  • path: Absolute path, or a path that starts with ~.
  • mode: "read" to read and run files, or "read_write" for a cache that the tool also writes.
  • reason: Short text that tells people why the extension needs the path.
Raises:
  • ValueError: The path or reason is blank, or the mode is unknown.
Grant( path: str | os.PathLike[str], mode: Literal['read', 'read_write'] = 'read', *, reason: str)
path: str | os.PathLike[str]
mode: Literal['read', 'read_write']
reason: str
ProviderJSON: TypeAlias = str | int | float | bool | None | Mapping[str, 'ProviderJSON'] | Sequence['ProviderJSON']
ProviderAPIStyle: TypeAlias = Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']
ProviderLimitStatus: TypeAlias = Literal['error', 'ok', 'unauthenticated', 'unknown-provider', 'unsupported']
ProviderLimitScope: TypeAlias = Literal['account', 'model', 'plan', 'workspace']
ProviderLimitKind: TypeAlias = Literal['credits', 'rate', 'requests', 'sessions', 'tokens', 'usd']
ProviderWindowKind: TypeAlias = Literal['calendar', 'lifetime', 'rolling']
ProviderWindowUnit: TypeAlias = Literal['day', 'hour', 'minute', 'month', 'week', 'year']
ProviderLimitPrecision: TypeAlias = Literal['derived', 'estimate', 'exact', 'unknown']
ProviderLimitSource: TypeAlias = Literal['derived', 'local', 'provider-api', 'static']
@dataclass(frozen=True, slots=True, kw_only=True)
class ProviderPreset(_ProviderValue):

Endpoint defaults, not credentials. Opaque API payload keys remain unchanged.

extra carries additional router settings as JSON, never overriding named fields. Use enrich_models_fn for typed model metadata beyond default model names.

ProviderPreset( *, base_url: str | None = None, api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']] = None, default_models: Sequence[str] = (), responses_path: str | None = None, llm_headers: Mapping[str, str] | None = None, extra_body: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None, is_hidden: bool | None = None, extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None)
base_url: str | None
api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']]
default_models: Sequence[str]
responses_path: str | None
llm_headers: Mapping[str, str] | None
extra_body: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
is_hidden: bool | None
extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderCredential(_ProviderValue):

A usable credential. Tokens and headers are excluded from repr.

Return None instead when no credential is available.

ProviderCredential( token: str, api_url: str | None = None, api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']] = None, responses_path: str | None = None, llm_headers: Mapping[str, str] | None = None, source: str | None = None)
token: str
api_url: str | None
api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']]
responses_path: str | None
llm_headers: Mapping[str, str] | None
source: str | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderStatus(_ProviderValue):

Connection verdict, separate from usage limits. extra is display-only JSON metadata.

ProviderStatus( is_authenticated: bool, error: str | None = None, source: str | None = None, provider_id: str | None = None, status: str | None = None, base_url: str | None = None, label: str | None = None, config_path: str | None = None, extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None)
is_authenticated: bool
error: str | None
source: str | None
provider_id: str | None
status: str | None
base_url: str | None
label: str | None
config_path: str | None
extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderModel(_ProviderValue):

Model metadata returned by enrichment. extra keeps additional router fields.

is_tool_call and is_image_input are old names of is_tool_calls_supported and is_vision_supported. They still work. A new name wins over its old name.

ProviderModel( name: str, context: int | None = None, is_tool_calls_supported: bool | None = None, is_vision_supported: bool | None = None, extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None, is_tool_call: bool | None = None, is_image_input: bool | None = None)
name: str
context: int | None
is_tool_calls_supported: bool | None
is_vision_supported: bool | None
extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
is_tool_call: bool | None
is_image_input: bool | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderLimitWindow(_ProviderValue):

The contract's calendar, rolling or lifetime window for one limit.

ProviderLimitWindow( kind: Literal['calendar', 'lifetime', 'rolling'], unit: Optional[Literal['day', 'hour', 'minute', 'month', 'week', 'year']] = None, size: int | None = None, resets_at_ms: int | None = None)
kind: Literal['calendar', 'lifetime', 'rolling']
unit: Optional[Literal['day', 'hour', 'minute', 'month', 'week', 'year']]
size: int | None
resets_at_ms: int | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderLimit(_ProviderValue):

One usage row in contract form.

Finite measurements keep their original precision.

ProviderLimit( id: str, label: str, scope: Literal['account', 'model', 'plan', 'workspace'], kind: Literal['credits', 'rate', 'requests', 'sessions', 'tokens', 'usd'], precision: Literal['derived', 'estimate', 'exact', 'unknown'], source: Literal['derived', 'local', 'provider-api', 'static'], is_unlimited: bool = False, used: int | float | None = None, limit: int | float | None = None, remaining: int | float | None = None, window: ProviderLimitWindow | None = None, subject: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None, note: str | None = None)
id: str
label: str
scope: Literal['account', 'model', 'plan', 'workspace']
kind: Literal['credits', 'rate', 'requests', 'sessions', 'tokens', 'usd']
precision: Literal['derived', 'estimate', 'exact', 'unknown']
source: Literal['derived', 'local', 'provider-api', 'static']
is_unlimited: bool
used: int | float | None
limit: int | float | None
remaining: int | float | None
window: ProviderLimitWindow | None
subject: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
note: str | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderError(_ProviderValue):

A limits failure, not an exception carrying a credential.

ProviderError( type: str, message: str, data: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None)
type: str
message: str
data: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True, kw_only=True)
class ProviderLimits(_ProviderValue):

Usage snapshot. The host fills omitted provider_id and fetched_at_ms values.

ProviderLimits( *, status: Literal['error', 'ok', 'unauthenticated', 'unknown-provider', 'unsupported'] = 'ok', limits: Sequence[ProviderLimit] = (), rpm: int | None = None, tpm: int | None = None, note: str | None = None, error: ProviderError | None = None, provider_id: str | None = None, fetched_at_ms: int | None = None)
status: Literal['error', 'ok', 'unauthenticated', 'unknown-provider', 'unsupported']
limits: Sequence[ProviderLimit]
rpm: int | None
tpm: int | None
note: str | None
error: ProviderError | None
provider_id: str | None
fetched_at_ms: int | None
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data. Declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class Provider:

Pure provider declaration. register_extension adapts its callbacks to the host.

Credential reads are passive, and only auth_fn may start a login. Callbacks are synchronous and may run without a session. Refresh accepts zero arguments or one rejected token (None if unknown), chosen without retrying callback errors. Enrichment and selection inputs remain JSON mappings owned by the router or config, not a second SDK schema for those domains. Callback outputs use typed records.

Provider( id: str, label: str, preset: ProviderPreset | None = None, is_managed: bool = False, get_token_fn: Callable[[], ProviderCredential | None] | None = None, detect_fn: Callable[[], ProviderCredential | None] | None = None, status_fn: Callable[[], ProviderStatus | None] | None = None, logout_fn: Callable[[], None] | None = None, limits_fn: Callable[[], ProviderLimits | None] | None = None, refresh_token_fn: Callable[[str | None], ProviderCredential | None] | Callable[[], ProviderCredential | None] | None = None, auth_fn: Callable[[Callable[[str], None]], str | bool | None] | None = None, auth_prompt_fn: Callable[[], Sequence[str] | str | None] | None = None, enrich_models_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Sequence[ProviderModel] | None] | None = None, on_selected_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], None] | None = None)
id: str
label: str
preset: ProviderPreset | None
is_managed: bool
get_token_fn: Callable[[], ProviderCredential | None] | None
detect_fn: Callable[[], ProviderCredential | None] | None
status_fn: Callable[[], ProviderStatus | None] | None
logout_fn: Callable[[], None] | None
limits_fn: Callable[[], ProviderLimits | None] | None
refresh_token_fn: Callable[[str | None], ProviderCredential | None] | Callable[[], ProviderCredential | None] | None
auth_fn: Callable[[Callable[[str], None]], str | bool | None] | None
auth_prompt_fn: Callable[[], Sequence[str] | str | None] | None
enrich_models_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Sequence[ProviderModel] | None] | None
on_selected_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], None] | None
def ok(title, body=None, data=None):
def err(title, body=None, data=None):
def block(reason, hint=None):

Refuse the call a before op hook is deciding.

The model reads reason as the tool failure. hint tells it what to do next; without one, the failure advises asking the user before a retry.

def repair(source, *, notes):

Return repaired source from a before hook, without writing or running it.

Only patch accepts this decision. Vis repairs python_execution blocks itself with parinferish. The host validates the request and reports the notes with the result. A patch still checks the complete original file before it writes the repaired text once.

def strings_of(value):
state = <blockether.vis.extension._State object>
def workspace_root() -> pathlib._local.Path:

Return the active session workspace as a path, including its current draft.

This is the primary working copy, not a language-specific project marker. Outside Vis, it is the process directory. A hosted call without a bound session fails instead of silently using the gateway process directory.

def project_root() -> pathlib._local.Path:

Return this extension's project root in the active session workspace.

The project root is the nearest directory above the extension entry that holds a .vis directory, mapped into the session working copy. For an extension in the workspace's own .vis/extensions, it is workspace_root(). Outside Vis, it is the process directory. A hosted call without a bound session fails, like workspace_root().

def workspace_files() -> list[pathlib._local.Path] | None:

Return the files under workspace_root() that Git sees.

The list holds tracked files and untracked files that Git does not ignore. The host lists the working copy once and gives the same list to each extension that asks within a short time. Use it instead of a child process for each extension. The result is None when the workspace is not in a Git working tree. Outside Vis, it lists the process directory.

def python_environment() -> Literal['project', 'shared']:

Return the Python environment that the session's project selects.

The value comes from python.environment in the merged configuration. shared means that project Python runs like vis-agent python --shared, with Vis shared packages and without the project's .venv. project, the default, means that the project's own environment applies. Outside Vis, the value is project.

def log(level, msg):
def notify(text, level='info'):
def write_private_file(path: str | os.PathLike[str], text: str) -> None:

Replace path with text in one atomic step, readable only by its owner.

A failure keeps the previous file. A symbolic link writes the file it names. Missing parent directories are created owner-only.

class Shell(builtins.dict):
def logs(self, offset=None, limit=None):
def type(self, text, is_enter=True):
def stop(self):
def wait(self, seconds=120):
def shell(opts):
def jailed_shell(opts):
fs = <blockether.vis.extension._Fs object>
class Answer:
Answer(raw)
is_submitted
reason
request_id
values
def get(self, name, default=None):
def reveal(self, name):
def ask(title, nodes, **options):
def reveal(handle):
def forget(handle):
def attach( source, filename=None, *, kind=None, media_type=None, label=None, audience='both', on_event=None):

Store an artifact as a durable attachment of the running block, or open an app.

source is a path, bytes, a Matplotlib figure or a Pillow image. The engine stores the artifact with the python_execution block that called this tool, exactly like the sandbox attach verb. A second attachment under the same filename is the next version of that artifact. audience is both, user or model; a PDF or HTML document is for the user only.

Returns the descriptor dict: id, filename, version, media_type, kind, size and audience. Every read function takes it as its target.

An app declaration from vis.ui.app(...) opens an app and answers its LiveAttachment handle. The app row from list_attachments() answers a handle for that app again. An app takes only on_event(event), which then runs for each human change, submit, activate or shortcut, as with vis.live.

def list_attachments(name=None):

List the artifact descriptors of this session, oldest first.

The list includes the artifacts that the running block attached. With name, it lists the versions of that one artifact.

def get_attachment(target, version=None):

Return the descriptor of one artifact, without its bytes.

target is a filename, an id or a descriptor. A filename means the latest version, unless you give version. A negative version counts back.

def read_attachment(target, version=None):

Return the bytes of one artifact. target works as in get_attachment.

def show_attachment(target, version=None):

Show a stored image to the model for the next request only.

target works as in get_attachment. Returns the id, filename, media_type and size of the queued image.

def read_session(target=None):

Read one whole session, exactly as the sandbox read_session does.

Without target, it reads the session that runs the extension. target takes a bare id, an unambiguous prefix or the copied vis_session_id#<uuid> marker. The introspection toggle does not apply, because an extension is trusted code.

Returns the same string-keyed dict: session, current_turn, failures, diagnosis, session_forks, turn_retries, usage and transcript.

def list_sessions(search=None):

List sessions newest first, exactly as the sandbox list_sessions does.

With search, it runs the session search of the TUI and the Companion app. The rows come in rank order: title, request, reply, then thinking, newest first in each rank. Keep that order. The introspection toggle does not apply.

Returns string-keyed rows: id, channel, title, goal, turn_count, created_at and modified_at. A match adds rank, the is_in_* flags and optional snippets.

class Interrupted(builtins.Exception):

The live view this handle drives is no longer open.

Raised by the next push after the human stopped watching — Escape in the terminal, Stop in the app — so an unattended loop ends by itself. A loop that would rather finish its own work reads view.is_interrupted instead and decides.

note is the comment the person left with the stop, when they left one: the reason it is being stopped, in their words.

Interrupted(view_id, reason=None, note=None)
view_id
reason
note
@dataclass(frozen=True)
class LiveEvent:

One human action on a live view: change, submit, activate, select or shortcut.

value is the new value of a changed input; a secret input gives only is_set. values holds every input that a submit button collected. A Send in an app gives submission_id, the id of its receipt, and comments: the thread messages that it posted, each with node_id, thread_id, message_id, body and the anchor of a diff line. A pressed app shortcut gives action, the id that the shortcut sends, and key, the key that the person pressed. A shortcut that names a node gives an activate event of that node instead. A selection in a selectable table gives item_ids, the selected row ids.

LiveEvent( type: str, node_id: str, seq: int, value: Any = None, values: Mapping[str, typing.Any] | None = None, is_set: bool | None = None, submission_id: str | None = None, comments: list[Mapping[str, typing.Any]] | None = None, action: str | None = None, key: str | None = None, item_ids: list[str] | None = None)
type: str
node_id: str
seq: int
value: Any = None
values: Mapping[str, typing.Any] | None = None
is_set: bool | None = None
submission_id: str | None = None
comments: list[Mapping[str, typing.Any]] | None = None
action: str | None = None
key: str | None = None
item_ids: list[str] | None = None
class TextNode(_Node):

A paragraph, heading or code block, replaced in place.

def set(self, text, **spec):
Inherited methods and attributes
TextNode(view, node_id, type_name)
node_id
type
class Spinner(_Node):

One node of a live view, addressed by its id.

def set(self, text=None, *, variant=None, is_active=None):
Inherited methods and attributes
Spinner(view, node_id, type_name)
node_id
type
class Button(_Node):

One node of a live view, addressed by its id.

def set(self, label=None, *, is_disabled=None):
clicks

How often the human pressed this button.

Inherited methods and attributes
Button(view, node_id, type_name)
node_id
type
class Card(_Node):

A titled card. A selectable card counts activations like a button.

def set( self, title=None, *, text=None, tone=None, meta=None, is_selectable=None):
clicks

How often the human activated this card.

Inherited methods and attributes
Card(view, node_id, type_name)
node_id
type
class Input(_Node):

One input component of a live view, read and set by id.

A secret input never shows its value: value is None for it, and its value arrives only in the submit event, as a vis-secret: handle.

def set(self, value):
def clear(self):
value

The current value, as the human or the extension last set it.

Inherited methods and attributes
Input(view, node_id, type_name)
node_id
type
class Status(_Node):

One node of a live view, addressed by its id.

def set(self, text, tone=None, detail=None, label=None):
Inherited methods and attributes
Status(view, node_id, type_name)
node_id
type
class Progress(_Node):

One node of a live view, addressed by its id.

def set(self, value=None, done=None, total=None, label=None):
Inherited methods and attributes
Progress(view, node_id, type_name)
node_id
type
class Stat(_Node):

One node of a live view, addressed by its id.

def set(self, stat_id, value_text, label, tone=None):
def clear(self):
def remove(self, *item_ids):
Inherited methods and attributes
Stat(view, node_id, type_name)
node_id
type
class Steps(_Node):

One node of a live view, addressed by its id.

def set(self, step_id, label, tone=None, detail=None, value=None):
def clear(self):
def remove(self, *item_ids):
Inherited methods and attributes
Steps(view, node_id, type_name)
node_id
type
LogTone: TypeAlias = Literal['idle', 'running', 'ok', 'warn', 'error']
class Log(_Node):

One node of a live view, addressed by its id.

def write( self, *lines: str | Sequence[str], tone: Optional[Literal['idle', 'running', 'ok', 'warn', 'error']] = None):

Append complete lines, optionally styled by severity. Redact before writing.

Each argument is one stored line, not a raw byte fragment. A tone applies only to this call, and omitted tones are plain. Controls are displayed literally.

def clear(self):
Inherited methods and attributes
Log(view, node_id, type_name)
node_id
type
class Table(_KeyedNode):

A node holding items the extension addresses by id: it can drop them.

def upsert(self, row_id, cells, tone=None, parent=None):
def group(self, group_id, label=None, tone=None, order=None, is_open=None):
def select(self, *item_ids):
def replace(self, rows):

Make rows the whole table, in their order, in one step.

Each row is a vis.ui.table_row(row_id, cells, tone=, parent=). Rows not given are dropped. A selected row that comes back stays selected, and a selected row that is gone loses its selection. An empty list empties the table.

Inherited methods and attributes
Table(view, node_id, type_name)
node_id
type
def remove(self: blockether.vis.extension._Node, *item_ids):
def clear(self):
class DiffNode(_Node):

A unified diff that people review line by line, with threads on lines.

def set(self, patch=None, *, old_revision=None, new_revision=None):

Replace the patch or its revisions. Drafts of an older revision go stale.

threads

The threads on the lines of this diff, as the surfaces paint them.

def reply(self, body, **where):

Answer in a thread of this diff. See LiveAttachment.reply.

Inherited methods and attributes
DiffNode(view, node_id, type_name)
node_id
type
class ThreadNode(_Node):

One conversation between people and the owner of the app.

messages

The messages of this thread, oldest first.

def reply(self, body, **where):

Answer in this thread. See LiveAttachment.reply.

Inherited methods and attributes
ThreadNode(view, node_id, type_name)
node_id
type
class Scene(_KeyedNode):

A 3D world that surfaces draw. A tick sends only what changed.

def set(self, *, background=None, fov=None, is_wrapping=None, tick_ms=None):

Change the background, the field of view, wrapping or the tick length.

def put(self, object_id, **changes):

Add an object, or change the given keys of one: mesh, x, y, z, rx, ry, rz, scale, color, is_emissive. A new object needs mesh.

def mesh(self, mesh_id, vertices, faces):

Add or replace a custom mesh. See vis.ui.scene_mesh.

objects

The objects of this scene as the host holds them, by id.

Inherited methods and attributes
Scene(view, node_id, type_name)
node_id
type
def remove(self: blockether.vis.extension._Node, *item_ids):
def clear(self):
class LiveView:

A live view the human WATCHES, driven by the extension that opened it.

vis.live(...) mounts one and answers this handle. Nodes are addressed by id, as view['jobs'] or view.node('jobs'), and each answers the typed handle its own type declares. The view-level shortcuts (view.status(...), view.log(...), view.row(...)) resolve to the one node of that type. When the view holds several such nodes, they raise and name the candidate ids. So an ambiguous call fails where it was written, and does not quietly patch the wrong table.

Pushes are BATCHED. Ops buffer and cross on the next push after flush_ms, or when a coalesced push fills. They always cross before the view is read or closed. with view.batch(): groups one logical picture explicitly, including structural add and drop operations. Repeated writes to the same row or node collapse into the last one. So a per-row progress counter costs one wire row per tick, not one per write.

Closing is the point: close() answers either the structured verdict or the compact model_result that the extension chose. Used as a context manager, the view closes itself. It closes as completed on the way out, and as failed with the error when the body raised. A run that died midway still owes the model what happened.

LiveView(request, flush_ms=None, on_event=None)
view_id
@contextmanager
def batch(self):

Send one complete picture for a related group of node changes.

Node handles still coalesce exactly as usual, but neither the leading edge nor structural add/drop operations cross the host seam until the outermost batch ends. Reads and close remain explicit flush points.

def flush(self):

Send everything buffered. Called for you before any read or close.

def state(self):

What the view looks like right now, as the surfaces paint it.

def sleep(self, seconds):

Block until the view changes, ends, or seconds elapse.

One host wait replaces periodic state reads. An unchanged timeout returns no view payload. Returns True on change or close, False on timeout. Nonpositive durations do not flush or call the host.

is_interrupted

True once the human stopped watching.

Asks the engine at most once per flush window. So a compute loop can poll it on every iteration and still cost one host call per tick.

reason

Why the view ended, or None while open or after a compact result.

is_from_human

True when a PERSON ended it, rather than the run itself or a deadline.

A view is always stoppable. Nothing is asked of the human, so stopping it leaves nothing unanswered. This is how the run finds out that a human stopped it.

note

The comment the human left with their stop, or None.

The stop always lands. The note says WHY in their own words, and the same words reach the model in the verdict.

result

The structured verdict or compact model result, once ended.

def values(self):

Every visible input value by node id. Secret inputs are left out.

def events(self, timeout=None):

Return the human actions that arrived since the last call.

Without timeout this never waits. With timeout seconds it waits up to that long for the first new event. Each event also goes to every callback that on_event registered, in arrival order.

def on_event(self, callback):

Call callback(event) for each event that events or sleep reads.

A panel view opened with vis.live(..., on_event=...) gets its events from the host instead. Each one runs as an extension callback with the view session bound, also when no extension code waits on the view. Returns the callback, so you can use this method as a decorator.

def node(self, node_id):

The typed handle for one node, by id.

def status(self, text, tone=None, detail=None):
def progress(self, value=None, done=None, total=None):
def stat(self, stat_id, value_text, label, tone=None):
def step(self, step_id, label, tone=None, detail=None, value=None):
def write( self, *lines: str | Sequence[str], tone: Optional[Literal['idle', 'running', 'ok', 'warn', 'error']] = None):
def row(self, row_id, cells, tone=None, parent=None):
def group(self, group_id, label=None, tone=None, order=None, is_open=None):
def add(self, node, after=None):

Add a whole node to a running view — a scan that discovers a seventh device should not have to have declared it.

def drop(self, node_id):

Drop a whole node, its items with it. A group drops its nodes too.

def close( self, reason=None, summary=None, error=None, artifact_id=None, selection_snapshots=None, model_result=None):

End the view and answer the result the model reads.

model_result is an optional compact string returned instead of the full structured verdict. The finished picture and close metadata remain in the durable artifact and on human-facing close events.

selection_snapshots are finished pictures keyed by a selectable table and its selected rows. They are sealed only into the artifact record, so a reopened run can still switch rows without keeping its extension alive.

Closing twice is a no-op that answers the first result. A finally that closes what an interrupt already closed must not overwrite the reason that the human chose.

class LiveAttachment(LiveView):

An app: a live view that is attached to the session and keeps its state.

vis.attach(vis.ui.app(title, *nodes)) opens one and answers this handle. The app stays in the session after the call returns, also across gateway restarts. People use it in the TUI and in the Companion app with the same vis.ui components. They save drafts, and each Send gives a receipt in submissions(). The owner answers with reply(...) and moves a receipt with set_status(...).

id is the app identity. vis.attach(row) with the app row from list_attachments() answers a new handle for the same app, also after a restart.

LiveAttachment(request, flush_ms=None, on_event=None, door=None)
def flush(self):

Send everything buffered. Called for you before any read or close.

id

The app identity: the view id that every surface and record uses.

def reply( self, node_id, body, *, thread_id=None, anchor=None, submission_id=None):

Answer people in a thread of node node_id, as the owner of the app.

thread_id names an existing thread. On a diff without it, anchor ({"path", "side", "line"}) starts a thread on that line. submission_id names the Send that you answer, and its receipt moves to answered. The host sets the author. Answers thread_id, message_id and the new seq.

def set_status(self, submission_id, status, *, note=None):

Move the receipt of Send submission_id to status, and answer it.

status is delivered, processing, answered or failed. People see the stage, and note (at most 500 characters) tells them why.

def submissions(self, status=None):

The Send receipts of this app, oldest first, as dicts.

Each has id, idempotency_key, status, created_at and updated_at, and node_id, values, comments, revision or note when it has them. With status, only the receipts in that stage.

def submission(self, submission_id):

One Send receipt by id. An unknown id raises KeyError.

def thread(self, node_id, thread_id=None):

One thread as a dict: id, anchor, messages, draft, is_resolved.

A thread node is its own thread. On a diff, thread_id names the thread.

def set_shortcuts(self, shortcuts):

Replace the shortcuts of this app with vis.ui.shortcut(...) values.

An empty list removes every shortcut. Answers the new seq.

def set_agent_note(self, text):

Tell the model the current state of this app, in plain text.

The model reads the newest note with the next user message, also when no event reached the agent. A blank text removes the note. Answers the new seq.

def run(self, callback, *, timeout=None):

Serve the events of this app with callback(event) and block.

It returns when the app closes, when callback returns False, after timeout seconds or on KeyboardInterrupt. Answers the number of events that it served. Callbacks of on_event still get each event too.

def archive(self):

Close the app and keep it only in the archived list of its session.

The record stays readable. A second archive changes nothing.

Inherited methods and attributes

Inherited from LiveView.view_id.

Inherited from LiveView.batch.

Inherited from LiveView.state.

Inherited from LiveView.sleep.

Inherited from LiveView.is_interrupted.

Inherited from LiveView.reason.

Inherited from LiveView.is_from_human.

Inherited from LiveView.note.

Inherited from LiveView.result.

Inherited from LiveView.values.

Inherited from LiveView.events.

Inherited from LiveView.on_event.

Inherited from LiveView.node.

Inherited from LiveView.status.

Inherited from LiveView.progress.

Inherited from LiveView.stat.

Inherited from LiveView.step.

Inherited from LiveView.write.

Inherited from LiveView.row.

Inherited from LiveView.group.

Inherited from LiveView.add.

Inherited from LiveView.drop.

Inherited from LiveView.close.

def live(title, nodes, placement=None, key=None, on_event=None, **options):

Open a live view and answer the handle that drives it.

View options: description, source, session_id, channel_ids, plus flush_ms for the batching window. EVERY key is a snake_case string, exactly as vis.ask documents. There is no cancellable flag: a human can always stop watching, and the verdict says they did (is_from_human) and why (note).

placement is inline (the default), attachment (an app, which answers a LiveAttachment) or panel (the side panel of one session). Only extension code opens a panel view, and it needs a key: the extension and the key own one view, so opening the same key again replaces it. Such a view stays open after the call returns. It closes on close() or when the extension unloads. on_event(event) then runs for each human change, submit or activate, with the view session bound.

The view is mounted at once and nothing blocks — use it as a context manager so it closes itself:

with vis.live('Deploy', [vis.ui.steps('plan', steps=[...])]) as view:
    view['plan'].set('build', tone='running')

Closing answers the verdict: is_completed, reason, the finished picture as data, and whatever summary the extension chose to end with.

testing = <blockether.vis.extension._Testing object>
ui = <blockether.vis.extension._UI object>