Skip to content

blockether.vis.engine

Synchronous agent, session, execution-layer and Council APIs.

For installation, workflows and lifecycle choices, use the Python SDK guide.

class Agent:

Run one conversation, with optional application-owned tools.

Choose this entry point for sequential requests that should share history. Use send to get a blockether.vis.engine.Turn immediately, or run to wait for its result. Pass response_model to run for a validated Pydantic model. Access session for transcripts, attachments and progress.

Arguments:
  • project: Project directory. Local execution resolves an existing directory immediately. Gateway execution requires an absolute path on the gateway.
  • execution_layer: A blockether.vis.engine.LocalEngine or blockether.vis.engine.GatewayClient to borrow. When omitted, the agent creates and owns a local engine. Configure executable, credentials and transport timeouts on an explicit layer.
  • extensions: Iterable of blockether.vis.extension.Extension declarations. Their Python callables stay in your application and execute on the calling thread while you read, wait or iterate events.
  • tools: Optional iterable of Vis extension tool names for the conversation. Names such as "py.lint_code" or "py.repl_*" keep only those tools of the extensions they name. Names such as "-py.repl_stop" turn tools off. Do not mix the two forms. The gateway refuses an unknown name.

Construction does not start a process or connect. Session access, context entry, send or run starts the conversation. Closing the default agent stops its private process and discards its temporary session database, not file edits. An explicitly supplied layer is borrowed: you must close it yourself, after all agents using it have finished. An agent cannot be reused after close.

Raises:
  • TypeError: The supplied layer does not implement blockether.vis.engine.ExecutionLayer, an extension is unsupported, or tools is not an iterable of strings.
  • ValueError: A project or extension declaration is invalid.
  • FileNotFoundError: The local project does not exist.
  • NotADirectoryError: The local project is not a directory.
Agent( project='.', *, execution_layer: ExecutionLayer | None = None, extensions=(), tools=None)
execution_layer
project
session: Session

The same conversation for all requests. It connects on first access.

def register_extension(self, extension):

Add an application-owned Extension before this Agent's first request.

Callables keep their Python objects and execute on the SDK calling thread while you read, wait or iterate events. No code or closures are uploaded. Host-only extension fields are refused before connecting.

def send(self, request: str, **options) -> Turn:

Submit a request without waiting for the model to finish.

Arguments:
  • request: User message or an explicit slash command.
  • **options: Forwarded to blockether.vis.engine.Session.send, including provider, model, attachments and idempotency_key.
Returns:

A blockether.vis.engine.Turn bound to this conversation. Call its read, wait or cancel method to track or control this request. Further calls reuse the same conversation and its history.

Submission can make network calls and raise transport or gateway errors. No mutation is automatically retried. If you retry an uncertain submission, reuse an explicit idempotency key rather than accidentally starting it twice.

def run( self, request: str, *, timeout: float = 300, response_model: type[~ResponseModel] | None = None, max_corrections: int = 2, **options) -> Union[dict, ~ResponseModel]:

Submit a request and wait for its result in the same conversation.

Without response_model, return the turn dictionary in contract form. Check its status. Completed, failed, cancelled, suspended and error all end the wait. Failed model work is returned as a record, not raised.

With a Pydantic BaseModel subclass, the request includes its validation JSON Schema. The answer is validated against that schema, then by Pydantic with your Python validators. The call returns the validated instance.

This is a prompt, not provider-enforced JSON mode. The final prose block must be one complete JSON value, without fences or commentary. Other content block types are ignored. Duplicate object members and non-finite numbers are invalid.

An invalid answer starts a correction turn in the same conversation. The correction lists each problem and repeats the schema. This repeats up to max_corrections times, and 0 disables corrections. Each correction is another model call. The agent is asked to keep completed work, but tool calls and file edits are never undone.

StructuredOutputError reports an unsuccessful turn, an answer that is still invalid, or a deadline that left no time for a correction. Its errors and attempts say what went wrong.

timeout is a positive finite wait deadline in seconds for the answer and any corrections. It is separate from the layer's transport timeout. Submission options such as provider, model and attachments are passed unchanged to send. Corrections reuse them, except attachments and idempotency_key.

VisTimeout does not cancel the turn, and transport errors propagate. Closing an owned local agent stops unfinished work.

def close(self):

Detach callbacks and release only resources owned by this agent.

Repeated calls are safe. The default local engine stops, and its temporary session database is removed. Project file edits stay. A borrowed execution layer and other agents that use it stay open. After closing, requests through this agent raise blockether.vis.engine.TransportError.

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

Communicate with other Vis sessions.

Get this handle with Session.council. Do not construct it with internal binding fields. When acquired, the handle captures its session's group and current activation. It does not follow whichever run is latest.

Use members to discover active peers, publish to send a message and get to inspect replies. Publication, delivery and a completed answer are separate states. threads and read return cursor-based pages of conversation history.

Communication requires an available Council group. A disabled or missing group is reported when a communication operation or group_id needs it.

Council( _session: Any, _group_id: str, _activation_id: str | None, _binding_error: Exception | None = None)
group_id: str

Return the captured group or report why communication is unavailable.

def members(self) -> tuple[CouncilMember, ...]:

Active session ids, titles and states. This list is only a snapshot.

def publish( self, content: str, *, thread_id: int | None = None, title: str | None = None, ping: list[str] | str | None = None, idempotency_key: str | None = None, reply_required: bool = False, reply_to: int | None = None, files: dict[str, list[str]] | None = None) -> CouncilEntry:

Publish a message, optionally requesting replies from selected sessions.

Arguments:
  • content: Message text. Keep secrets and private data out of messages.
  • thread_id: Existing thread to continue. Omit it to start a new thread.
  • title: Optional readable title for the message.
  • ping: Session IDs, "all" for active peers, or no recipients. Explicit IDs wake an idle peer of the group or an eligible managed teammate.
  • idempotency_key: Reuse this key when retrying uncertain IO. The same publication is returned without notifying recipients twice.
  • reply_required: Track an answer obligation for each addressed recipient.
  • reply_to: Explicit unanswered request to answer. Without it, a no-ping thread continuation answers only the latest addressed entry, and only if that entry is an unanswered request. It never falls back to older requests.
  • files: File rules for the follow-up task of the one pinged idle child session. They replace the file grant of that child.
Returns:

CouncilEntry with its entry/thread IDs and per-recipient replies. Pending, delivered and unavailable states are not completed answers. Inspect the entry later with get to see replies and their states.

Each message owns its host-generated source_ref. That reference identifies this publication, not a reported incident. When you report an incident, include the original session and turn/iteration/form. Follow-ups and acknowledgements do not grant wake rights.

def threads( self, *, after: int = 0, limit: int = 50) -> CouncilPage:

List roots in entry-ID order. Replies do not reorder roots.

def read( self, *, thread_id: int | None = None, after: int = 0, limit: int = 50) -> CouncilPage:

Read a bounded page. This never consumes or acknowledges a ping.

def get(self, entry_id: int) -> CouncilEntry:

Read the full content behind a preview, in this handle's group.

@dataclass(frozen=True, slots=True)
class CouncilEntry:
CouncilEntry( entry_id: int, thread_id: int, group_id: str | None, content: str, author_session_id: str, created_at: int, source: str, ping: tuple[str, ...], title: str | None = None, source_ref: CouncilSource | None = None, reply_required: bool = False, reply_to: int | None = None, replies: tuple[CouncilReply, ...] = ())
entry_id: int
thread_id: int
group_id: str | None
content: str
author_session_id: str
created_at: int
source: str
ping: tuple[str, ...]
title: str | None
source_ref: CouncilSource | None
reply_required: bool
reply_to: int | None
replies: tuple[CouncilReply, ...]
@classmethod
def from_wire(cls, data):
@dataclass(frozen=True, slots=True)
class CouncilMember:
CouncilMember(session_id: str, title: str, state: str)
session_id: str
title: str
state: str
@dataclass(frozen=True, slots=True)
class CouncilPage:
CouncilPage( entries: tuple[CouncilEntry | CouncilThread, ...], after: int, has_more: bool)
entries: tuple[CouncilEntry | CouncilThread, ...]
after: int
has_more: bool
@classmethod
def from_wire(cls, data, row, definition):
@dataclass(frozen=True, slots=True)
class CouncilReply:
CouncilReply(session_id: str, state: str, reply_entry_id: int | None = None)
session_id: str
state: str
reply_entry_id: int | None
@dataclass(frozen=True, slots=True)
class CouncilSource:
CouncilSource( session_id: str, turn: int | None, iteration: int | None, form: int | None, operation_id: str | None = None, session_state_id: str | None = None, session_turn_soul_id: str | None = None, session_turn_state_id: str | None = None, session_turn_iteration_id: str | None = None, tool_call_id: str | None = None)
session_id: str
turn: int | None
iteration: int | None
form: int | None
operation_id: str | None
session_state_id: str | None
session_turn_soul_id: str | None
session_turn_state_id: str | None
session_turn_iteration_id: str | None
tool_call_id: str | None
@classmethod
def from_wire(cls, data):
@dataclass(frozen=True, slots=True)
class CouncilThread:
CouncilThread(thread_id: int, title: str, author_session_id: str, created_at: int)
thread_id: int
title: str
author_session_id: str
created_at: int
@dataclass(frozen=True, slots=True)
class Event:

Typed envelope that Events yields. Payload fields depend on type.

Attributes:
  • type: Event name identifying how to interpret the payload.
  • session_id: Conversation that emitted the event.
  • seq: Sequence number for a journal event, when supplied.
  • cursor: Replay position from a subscription control event, when supplied.
  • turn_id: Request associated with this event, or None for session events.
  • data: Event-specific dictionary. It is not a final turn result by itself.
  • activity: Validated activity projection, when this event carries one.
  • view: Validated view event, when this event carries one.

Stream readers validate wire records through from_wire. Use Turn.read or Turn.wait for the final result of a request. Do not treat each event as a completed response.

Event( type: str, session_id: str, seq: int | None = None, cursor: int | None = None, turn_id: str | None = None, data: dict[str, typing.Any] = <factory>, activity: blockether.vis.activity.ActivityProjection | None = None, view: blockether.vis.views.ViewEvent | None = None)
type: str
session_id: str
seq: int | None
cursor: int | None
turn_id: str | None
data: dict[str, typing.Any]
@classmethod
def from_wire(cls, value: Any) -> Event:
class Events(blockether.vis.engine._client._EventStream):

Iterate typed session events, with replay and bounded reconnection.

Get a stream from Session.events. It yields Event objects. Inspect type, turn_id and the event-specific data instead of parsing raw SSE. The stream covers the whole conversation and does not stop when one turn completes. Use it in a with block, or call close when you are done.

Arguments:
  • session: Conversation whose events you want to follow.
  • cursor: Nonnegative replay position, normally a saved cursor or the cursor from Session.send. Defaults to 0.
  • **options: reconnects limits retries (default 3). retry_delay is a nonnegative finite delay in seconds (default 0.2).

The stream replays every stored event after cursor, including the progress of turns that finished before you subscribed. Duplicate events are suppressed. subscription.ready resets the cursor, also after a daemon restart. So save the stream's current cursor, and do not assume that it always increases.

The idle timeout is the client's timeout. Closing the stream never cancels a turn. Invalid replay options raise ValueError. Malformed events raise ProtocolError.

Events( session: Session, *, cursor=0, **options)
Inherited methods and attributes
client
def close(self):

Release this event subscription without cancelling conversation work.

Close the active response and iterator, and detach the stream from its client. Repeated calls are safe, and further iteration stops. The client and session stay open. Context exit calls this method automatically.

class ExecutionLayer(abc.ABC):

Shared session API for local processes and explicit gateway connections.

Applications normally choose blockether.vis.engine.LocalEngine or blockether.vis.engine.GatewayClient, then use create_session to start a conversation or session to attach to an existing session ID. Agent wraps this API when you only need one conversation.

The named HTTP methods below expose lower-level engine routes. Their body and query values use the gateway's wire schema, not any Python object. For ordinary workflows, use Session and Turn methods. Route failures raise GatewayError. IO failures and malformed replies raise TransportError or its subclasses. Mutations are not automatically retried.

Arguments:
  • timeout: Positive finite transport timeout in seconds. Defaults to 30. This is independent of a turn's wait deadline.

Use each layer and its session handles on one calling thread. Context entry calls connect, and context exit calls close. A custom transport implements connect, close, session_options and _open. _open returns a context-managed binary response with status and headers. Path and channel defaults belong to the layer, not to Agent.

timeout
@abstractmethod
def connect(self) -> ExecutionLayer:

Connect or start the layer. Repeated calls keep the same connection.

@abstractmethod
def close(self) -> None:

Release this layer's resources without deleting remote sessions.

@abstractmethod
def session_options( self, project='.') -> dict[str, str | int | float | bool | None | list[str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]] | dict[str, str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]]]:

Validate a project and return its wire-ready session creation options.

speech: Speech

Return the Speech API: voices, synthesis and transcription.

def session(self, sid: str) -> Session:

Return a lightweight handle for an existing session ID.

This does not fetch or validate the remote session. Use Session.read to get its state. That request reports a missing session. The handle borrows this layer and does not own its lifetime.

def create_session( self, *, provider: str | None = None, model: str | None = None, timeout=None, **options) -> Session:

Create a conversation and return a Session handle.

Arguments:
  • provider: Optional provider ID from models. Give it with model.
  • model: Optional model name. The pair pins the model of the session before its first turn.
  • timeout: Optional timeout in seconds for this creation request.
  • **options: Session creation fields from the gateway contract. For a project path, pass **layer.session_options(project). This applies local and gateway path rules and channel defaults correctly.

This sends a creation request immediately. Unlike session, it creates a new record. To attach to it later, use the returned handle's id.

def restore_session( self, data: bytes, *, root: str | None = None, group_id: str | None = None, key: str | None = None, timeout: float | None = None) -> Session:

Restore a session snapshot as a new session and return its handle.

Arguments:
  • data: The bytes that Session.export returned, for example read from a .vissnap file.
  • root: Optional folder for the restored sessions. Without it, the gateway uses the project folder of group_id, else its own folder.
  • group_id: Optional session group for the restored session.
  • key: Optional restore key. A second restore with the same key returns the session of the first restore and creates nothing.
  • timeout: Optional timeout in seconds for this request.

Each restored session and child gets a new ID. Your own permissions apply, not the permissions in the snapshot. A child that was running fails, so you can resume it. Read the manifest first with snapshot_info.

@staticmethod
def snapshot_info(data: bytes) -> dict[str, typing.Any]:

Read the manifest of session snapshot data without a gateway request.

The manifest gives format_version, vis_version, created_at, root_session_id, sessions, counts, sizes and payload_sha256. Raises ValueError when data is not a Vis session snapshot.

def put_session_client_extensions( self, sid: str, *, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')], timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PUT /v1/sessions/:sid/client-extensions — bind client-owned declarations.

def delete_session_client_extensions( self, sid: str, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/sessions/:sid/client-extensions — detach owned callbacks.

def get_session_client_calls( self, sid: str, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/client-calls — retrieve pending owned calls.

def post_session_client_call_result( self, sid: str, call_id: str, *, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')], timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/client-calls/:call_id/result — acknowledge one call.

def post_session_client_call_activity( self, sid: str, call_id: str, *, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')], timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/client-calls/:call_id/activity — publish presentation.

def list_sessions(self, **query):

Return one page with next_cursor. It does not fetch every session.

def get_capabilities( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/capabilities — json response.

def get_devices( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/devices — json response.

def post_devices( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/devices — json response.

def delete_device( self, token: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/devices/:token — json response.

def post_devices_test( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/devices/actions/test — json response.

def get_fs( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/fs — json response.

def post_fs_mkdir( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/fs/actions/mkdir — json response.

def get_mcp_servers( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/mcp/servers — json response.

def post_mcp_servers( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers — json response.

def delete_mcp_server( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/mcp/servers/:name — json response.

def put_mcp_server( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PUT /v1/mcp/servers/:name — json response.

def post_mcp_server_enable( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/actions/enable — json response.

def post_mcp_server_kill( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/actions/kill — json response.

def post_mcp_server_start( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/actions/start — json response.

def post_mcp_server_auth_cancel( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/auth/cancel — json response.

def post_mcp_server_auth_complete( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/auth/complete — json response.

def post_mcp_server_auth_logout( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/auth/logout — json response.

def post_mcp_server_auth_poll( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/auth/poll — json response.

def post_mcp_server_auth_start( self, name: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/:name/auth/start — json response.

def post_mcp_servers_test( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/mcp/servers/actions/test — json response.

def get_models( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/models — json response.

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

Return the models that the gateway pickers show, in picker order.

Arguments:
  • provider: Optional provider ID. Only its models are returned.
  • is_available: Optional filter. True keeps models of signed-in providers. False keeps the others.
  • timeout: Optional timeout in seconds for this request.

Each Model carries its price, billing, capabilities and the limits report of its provider. The gateway reads provider limits in parallel and bounds each read, so a slow provider gives an error report.

def model_limits( self, provider: str, *, timeout: float | None = None) -> dict[str, str | int | float | bool | None | list[str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]] | dict[str, str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]]]:

Return the current limits report of one provider.

Use it to refresh the Model.limits of that provider's models. The report follows the provider limits schema. A failed read gives an error report, not an exception.

def get_projects( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/projects — json response.

def post_projects( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/projects — json response.

def delete_project( self, pid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> None:

DELETE /v1/projects/:pid — empty response.

def get_project( self, pid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/projects/:pid — json response.

def patch_project( self, pid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/projects/:pid — json response.

def patch_project_sessions( self, pid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/projects/:pid/sessions — json response.

def post_projects_ensure( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/projects/actions/ensure — json response.

def get_projects_overview( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/projects/overview — json response.

def get_provider_presets( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/provider-presets — json response.

def post_providers( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/providers — json response.

def delete_provider( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/providers/:provider-id — json response.

def post_provider_auth_cancel( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/providers/:provider-id/auth/cancel — json response.

def post_provider_auth_complete( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/providers/:provider-id/auth/complete — json response.

def post_provider_auth_poll( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/providers/:provider-id/auth/poll — json response.

def post_provider_auth_start( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/providers/:provider-id/auth/start — json response.

def get_provider_limits( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/providers/:provider-id/limits — json response.

def post_provider_reset_credits_consume( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/providers/:provider-id/reset-credits/consume — json response.

def post_provider_logout( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/providers/:provider-id/logout — json response.

def get_provider_models( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/providers/:provider-id/models — json response.

def get_provider_status( self, provider_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/providers/:provider-id/status — json response.

def get_router( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/router — json response.

def patch_router( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/router — json response.

def get_session_groups( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/session-groups — the groups of one project, in the human's order.

def post_session_groups( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/session-groups — create one group inside a project.

def delete_session_group( self, gid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/session-groups/:gid — drop a group and say what becomes of it.

def patch_session_group( self, gid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/session-groups/:gid — rename, recolour or reorder a group.

def get_sessions( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions — json response.

def post_sessions( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions — json response.

def delete_session( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> None:

DELETE /v1/sessions/:sid — empty response.

def get_session( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid — json response.

def patch_session( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/sessions/:sid — json response.

def get_session_activity( self, sid: str, aid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/activity/:aid — a bounded page of complete Activity records.

Query with after, limit and q. Follow history.next_after until it is null. When you combine pages into one copy, reject a changed revision.

def get_session_activity_export( self, sid: str, aid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/sessions/:sid/activity/:aid/export — complete unfiltered Activity text.

The gateway streams this response, and the returned Response buffers its bytes. Supply revision to reject a changed history before the export starts. Raises ProtocolError if a concurrent change marks the export incomplete.

def get_session_alert( self, sid: str, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/alert — the banner text for the latest turn.

def get_session_artifacts( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/artifacts — json response.

def post_session_attachments( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, content: bytes | None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/attachments — json response.

def post_session_cancel_current( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/cancel-current — json response.

def get_session_context( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/context — json response.

def post_session_drain_queue( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/drain-queue — json response.

def get_session_events_since( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/events-since — json response.

def get_session_forks( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/forks — json response.

def post_session_forks( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/forks — json response.

def post_session_fs_actions_open( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/fs/actions/open — open one file in the editor.

def get_session_fs_file( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/fs/file — preview one workspace file by its kind.

def put_session_group( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PUT /v1/sessions/:sid/group — file this session under a group.

def get_session_iteration_attachments( self, sid: str, iid: str, idx: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/sessions/:sid/iterations/:iid/attachments/:idx — binary response.

def put_session_iteration_attachment_agent_note( self, sid: str, iid: str, idx: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PUT /v1/sessions/:sid/iterations/:iid/attachments/:idx/agent-note — set the note of an HTML page for the model.

def get_session_model( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/model — json response.

def patch_session_model( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/sessions/:sid/model — json response.

def post_session_queue_send_now( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/queue/send-now — json response.

def put_session_read( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PUT /v1/sessions/:sid/read — how far this reader has read the session.

def post_session_release( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> None:

POST /v1/sessions/:sid/release — empty response.

def get_session_resources( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/resources — json response.

def get_session_resources_logs( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/resources/logs — json response.

def post_session_resources_stop( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/resources/stop — json response.

def post_session_resume_queue( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/resume-queue — json response.

def get_session_seq( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/seq — json response.

def get_session_slashes( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/slashes — json response.

def get_session_keybindings( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/keybindings — json response.

def post_session_keybindings_run( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/keybindings/run — json response.

def get_session_snapshot( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/sessions/:sid/snapshot — binary response.

def post_session_speech( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/speech — json response.

def delete_session_speech_job( self, sid: str, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/sessions/:sid/speech/jobs/:job-id — json response.

def get_session_speech_job( self, sid: str, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/speech/jobs/:job-id — json response.

def get_session_speech_job_audio( self, sid: str, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/sessions/:sid/speech/jobs/:job-id/audio — binary response.

def get_session_suggest( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/suggest — json response.

def get_session_transcript( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/transcript — json response.

def get_session_transcript_html( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/sessions/:sid/transcript.html — html response.

def get_session_transcript_md( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/sessions/:sid/transcript.md — markdown response.

def get_session_turns( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/turns — json response.

def post_session_turns( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/turns — json response.

def delete_session_turn( self, sid: str, tid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/sessions/:sid/turns/:tid — json response.

def get_session_turn( self, sid: str, tid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/turns/:tid — json response.

def patch_session_turn( self, sid: str, tid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/sessions/:sid/turns/:tid — json response.

def get_session_turn_attachments( self, sid: str, tid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/turns/:tid/attachments — json response.

def post_session_turn_cancel( self, sid: str, tid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/turns/:tid/cancel — json response.

def get_session_turn_trace( self, sid: str, tid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/turns/:tid/trace — json response.

def get_session_usage( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/usage — json response.

def post_session_view( self, sid: str, view_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/views/:view-id/actions — json response.

def get_session_views_input( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/views/input — json response.

def get_session_views_live( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/views/live — json response.

def get_session_views_live_view( self, sid: str, view_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/views/live/:view-id — json response.

def get_session_views_live_log( self, sid: str, view_id: str, node_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/views/live/:view-id/log — json response.

The node rides the query string: node ids are free text a surface chose and may hold /, which a path segment cannot carry.

def post_session_apps( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/apps — json response.

def post_session_voice( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, content: bytes | None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/:sid/voice — json response.

def delete_session_voice_job( self, sid: str, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/sessions/:sid/voice/jobs/:job-id — json response.

def get_session_voice_job( self, sid: str, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/voice/jobs/:job-id — json response.

def get_session_workspace( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/sessions/:sid/workspace — json response.

def patch_session_workspace_root( self, sid: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/sessions/:sid/workspace/root — json response.

def post_sessions_archive( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/actions/archive — archive many sessions; skip working ones.

def post_sessions_import( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, content: bytes | None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/sessions/actions/import — restore a session snapshot.

def post_machines_order( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/machines/order — register gateway IDs and return their durable order.

def get_automations( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/automations — json response.

def post_automations( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/automations — json response.

def get_automation_runs( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/automations/runs — json response.

def get_automation_run( self, run_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/automations/runs/:run-id — json response.

def get_automation( self, automation_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/automations/:automation-id — json response.

def patch_automation( self, automation_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/automations/:automation-id — json response.

def delete_automation( self, automation_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/automations/:automation-id — json response.

def post_automation_run( self, automation_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/automations/:automation-id/run — json response.

def post_automation_secrets( self, automation_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/automations/:automation-id/secrets — json response.

def post_evals_agreement( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/evals/agreement — json response.

def post_evals_compare( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/evals/compare — json response.

def post_evals_diagnose( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/evals/diagnose — json response.

def post_evals_labels( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/evals/labels — json response.

def post_evals_leaderboard( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/evals/leaderboard — json response.

def post_evals_report( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/evals/report — json response.

def get_settings( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/settings — json response.

def post_settings( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/settings — json response.

def patch_settings( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/settings — json response.

def get_setting( self, id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/settings/:id — json response.

def post_extensions_reload( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/extensions/reload — json response.

def post_speech( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/speech — json response.

def delete_speech_job( self, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/speech/jobs/:job-id — json response.

def get_speech_job( self, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/speech/jobs/:job-id — json response.

def get_speech_job_audio( self, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/speech/jobs/:job-id/audio — binary response.

def get_speech_model( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/speech/model — json response.

def post_speech_model( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/speech/model — json response.

def get_speech_voices( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/speech/voices — json response.

def post_speech_voices( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, content: bytes | None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/speech/voices — json response.

def delete_speech_voice( self, voice_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/speech/voices/:voice-id — json response.

def patch_speech_voice( self, voice_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')] = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PATCH /v1/speech/voices/:voice-id — json response.

def get_speech_voice_sample( self, voice_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> Response:

GET /v1/speech/voices/:voice-id/sample — binary response.

def get_decision_models( self, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/decisions/models — installed and resident decision versions.

def get_decision_model( self, model_ref: str, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/decisions/models/:model-ref — immutable status after upload.

def get_decision_alias( self, alias: str, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/decisions/aliases/:alias — current ref before a CAS update.

def put_decision_alias( self, alias: str, *, model_ref: str, expected_current: str | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

PUT /v1/decisions/aliases/:alias — explicit compare-and-swap update.

def post_decision_training_job( self, *, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')], timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/decisions/training/jobs — train on approved local data.

def get_decision_training_job( self, job_id: str, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/decisions/training/jobs/:job-id — durable progress.

def delete_decision_training_job( self, job_id: str, *, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/decisions/training/jobs/:job-id — cancel or discard.

def post_systemone( self, *, body: str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')], timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/systemone — typed decisions from an explicitly named model.

def post_voice( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None, content: bytes | None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/voice — json response.

def delete_voice_job( self, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

DELETE /v1/voice/jobs/:job-id — json response.

def get_voice_job( self, job_id: str, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/voice/jobs/:job-id — json response.

def get_voice_model( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

GET /v1/voice/model — json response.

def post_voice_model( self, *, query: Mapping[str, str | int | float | bool | None] | None = None, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/voice/model — json response.

def speech_events(self, job_id: str, **options) -> JobEvents:

Synthesis snapshots over SSE (HTTP) or bounded polling (local).

def voice_events(self, job_id: str, **options) -> JobEvents:

Transcription snapshots over SSE (HTTP) or bounded polling (local).

def session_speech_events( self, sid: str, job_id: str, **options) -> JobEvents:

Session-scoped synthesis snapshots. Close never cancels the job.

def session_voice_events( self, sid: str, job_id: str, **options) -> JobEvents:

Session-scoped transcription snapshots. Close never cancels the job.

class GatewayClient(blockether.vis.engine.ExecutionLayer):

Use an existing Vis gateway through an explicit HTTP(S) origin.

Choose this layer when sessions must outlive your Python process, or when the engine runs on another machine. It never starts or discovers a gateway, and it never reads locally saved credentials. Get an authorized gateway URL and token separately. Keep tokens out of source code and logs.

Arguments:
  • url: HTTP(S) origin such as https://gateway.example.com, without a path, query, fragment or embedded credentials.
  • token: Optional bearer token supplied by your application.
  • timeout: Positive finite transport timeout in seconds. Defaults to 30.

Construction makes no network request. Context entry calls connect to check protocol compatibility and acquire a client lease. Closing releases this client's lease and streams, not the gateway or its stored sessions. Reuse the same layer for multiple agents, then close it after those agents.

import os
from blockether.vis.engine import Agent, GatewayClient

with GatewayClient(
    "https://gateway.example.com", token=os.environ["VIS_GATEWAY_TOKEN"]
) as layer:
    with Agent(project="/workspace/project", execution_layer=layer) as agent:
        result = agent.run("Summarize this project without changing files.")
        print(result["status"])

The project path is absolute and belongs to the gateway machine, not this Python process. TLS certificate verification is enabled. Redirects are refused, so credentials cannot follow them to another origin. Each instance and its session handles use one calling thread. HTTP failures raise GatewayError. Connection, deadline and protocol failures use the TransportError exception family.

GatewayClient(url: str, *, token: str | None = None, timeout: float = 30)
def post_decision_model( self, *, content: <class 'BinaryIO'>, sha256: str, length: int, timeout: float | None = None) -> str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]:

POST /v1/decisions/models — stream a verified FP32 bundle once.

def connect(self) -> GatewayClient:

Check protocol compatibility and acquire this client's lease once.

Return this client. Repeated calls reuse its lease. An incompatible gateway raises ProtocolError. Authentication and other HTTP failures raise GatewayError.

A short outage of the gateway does not end the lease. When the gateway no longer holds the lease, for example after a restart, the client takes a new lease. A client with registered application extensions cannot move them to a new lease. It then raises TransportError for each request: create a new client. A refused keepalive has the same result.

def close(self) -> None:

Close streams and release this client's lease without stopping the gateway.

Stored remote sessions are not deleted. Repeated calls are safe, but this closed client cannot be reused. Use Turn.cancel when you need explicit cancellation rather than merely detaching a client.

def session_options( self, project='.') -> dict[str, str | int | float | bool | None | list[str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]] | dict[str, str | int | float | bool | None | list[ForwardRef('JSONValue')] | dict[str, ForwardRef('JSONValue')]]]:

Prepare an app-channel session using a path on the gateway machine.

Inherited methods and attributes

Inherited from ExecutionLayer.timeout.

Inherited from ExecutionLayer.speech.

Inherited from ExecutionLayer.session.

Inherited from ExecutionLayer.get_devices.

Inherited from ExecutionLayer.post_devices.

Inherited from ExecutionLayer.get_fs.

Inherited from ExecutionLayer.get_models.

Inherited from ExecutionLayer.models.

Inherited from ExecutionLayer.model_limits.

Inherited from ExecutionLayer.get_projects.

Inherited from ExecutionLayer.get_project.

Inherited from ExecutionLayer.get_router.

Inherited from ExecutionLayer.patch_router.

Inherited from ExecutionLayer.get_sessions.

Inherited from ExecutionLayer.get_session.

Inherited from ExecutionLayer.get_settings.

Inherited from ExecutionLayer.get_setting.

Inherited from ExecutionLayer.post_speech.

Inherited from ExecutionLayer.post_voice.

Inherited from ExecutionLayer.voice_events.

class GatewayError(builtins.RuntimeError):

HTTP failure with a safe code, never the gateway's free-form message.

Decision input-limit errors also expose input_tokens and max_input_tokens. These are None when the gateway does not supply valid numeric diagnostics. A refused view action, such as a press on a disconnected app, has its reason token as code.

GatewayError( status: int, code: str, *, input_tokens: int | None = None, max_input_tokens: int | None = None)
status
code
input_tokens
max_input_tokens
class InstallError(builtins.RuntimeError):

The SDK could not find or download the Vis engine.

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

Current speech job snapshot, not a session event or a replay cursor.

JobEvent( type: str, id: str, phase: str, is_done: bool, data: dict[str, typing.Any] = <factory>)
type: str
id: str
phase: str
is_done: bool
data: dict[str, typing.Any]
class JobEvents(blockether.vis.engine._client._EventStream):

Snapshot-first job SSE.

A reconnect reads the state again. A terminal is_done ends the stream.

Job streams have no cursor and make no exactly-once claim. Identical snapshots are suppressed within this iterator. Other fields stay as the JSON that the contract defines.

JobEvents(client, route, job_id, event_name, **options)
Inherited methods and attributes
client
def close(self):

Release this event subscription without cancelling conversation work.

Close the active response and iterator, and detach the stream from its client. Repeated calls are safe, and further iteration stops. The client and session stay open. Context exit calls this method automatically.

JSONValue = str | int | float | bool | None | list['JSONValue'] | dict[str, 'JSONValue']
class LocalEngine(blockether.vis.engine.ExecutionLayer):

Own a private Vis process without starting an HTTP server.

Use this layer when you need to choose an executable, share one local process between agents, or call the session API directly. With the default vis-agent, connect uses the installed command when it runs this SDK's release. Otherwise it downloads that release once into $VIS_HOME/sdk and checks it against the release checksums. Set VIS_NO_AUTO_INSTALL=1 to turn off the download. This class never discovers a gateway. Supported platforms are Linux and macOS.

Arguments:
  • executable: Executable path/name or nonempty argument sequence. The SDK appends stdio. It does not parse shell commands.
  • root: Existing local working directory, resolved at construction.
  • timeout: Positive finite timeout in seconds for each transport operation.
  • startup_timeout: Positive finite deadline in seconds for initial boot.

Construction is lazy. connect or context entry starts the subprocess, and repeated connection calls reuse it. close stops only this owned process and discards its temporary session database. It does not revert project edits or stop any separately running Vis server. Explicitly injected agents borrow this layer. Close those agents before you leave the layer's context.

from blockether.vis.engine import Agent, LocalEngine

with LocalEngine(executable="vis-agent", root=".") as layer:
    with Agent(project=".", execution_layer=layer) as agent:
        result = agent.run("Summarize this project without changing files.")
        print(result["status"])

A pipe timeout stops the process: a late reply must never be mistaken for the next request's answer. This differs from a blockether.vis.engine.Turn.wait deadline, which does not itself request cancellation. Startup, protocol and transport failures propagate as the corresponding engine exception. When the engine exited by itself or wrote to stderr, the message ends with that exit status and the last 4 KiB of its stderr.

LocalEngine(*, executable='vis-agent', root='.', timeout=30, startup_timeout=120)
def session_options(self, project='.') -> dict:

Resolve an existing local directory before the caller changes directory.

def connect(self):

Start the owned process once and validate its protocol handshake.

Return this layer for chaining. A closed layer cannot be restarted. After a failure or close, create a new one. A startup failure cleans up the process and temporary database before the error propagates.

def session(self, sid):

Return a lightweight handle for an existing session ID.

This does not fetch or validate the remote session. Use Session.read to get its state. That request reports a missing session. The handle borrows this layer and does not own its lifetime.

def close(self):

Stop this process, close its streams and remove its temporary database.

Repeated calls are safe, including retrying cleanup after a signal fails. A failed close still prevents new requests. Unfinished local work cannot continue after the process stops. File edits and separately running gateways are untouched.

Inherited methods and attributes

Inherited from ExecutionLayer.timeout.

Inherited from ExecutionLayer.speech.

Inherited from ExecutionLayer.get_devices.

Inherited from ExecutionLayer.post_devices.

Inherited from ExecutionLayer.get_fs.

Inherited from ExecutionLayer.get_models.

Inherited from ExecutionLayer.models.

Inherited from ExecutionLayer.model_limits.

Inherited from ExecutionLayer.get_projects.

Inherited from ExecutionLayer.get_project.

Inherited from ExecutionLayer.get_router.

Inherited from ExecutionLayer.patch_router.

Inherited from ExecutionLayer.get_sessions.

Inherited from ExecutionLayer.get_session.

Inherited from ExecutionLayer.get_settings.

Inherited from ExecutionLayer.get_setting.

Inherited from ExecutionLayer.post_speech.

Inherited from ExecutionLayer.post_voice.

Inherited from ExecutionLayer.voice_events.

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

One model with its price, billing, capabilities and provider limits.

pricing gives USD rates for 1M tokens: input, output and the optional cache_read and cache_write. Each item in pricing["tiers"] applies when a request uses more than above_tokens input tokens. Turn costs use the same rates. pricing is None when no price is known.

billing is metered for an API key, subscription for an account sign-in and local for a provider without credentials. limits is the limits report of the provider; all models of one provider share it.

Model( provider: str, name: str, label: str, pricing: dict[str, typing.Any] | None, billing: str, is_reasoning: bool, reasoning_options: tuple[str, ...], is_tool_call: bool, is_image_input: bool, modalities: dict[str, list[str]], is_available: bool, is_default: bool, limits: dict[str, typing.Any], pricing_source: str | None = None, family: str | None = None, context: int | None = None, input_limit: int | None = None, output_limit: int | None = None, knowledge_cutoff: str | None = None, release_date: str | None = None)
provider: str
name: str
label: str
pricing: dict[str, typing.Any] | None
billing: str
is_reasoning: bool
reasoning_options: tuple[str, ...]
is_tool_call: bool
is_image_input: bool
modalities: dict[str, list[str]]
is_available: bool
is_default: bool
limits: dict[str, typing.Any]
pricing_source: str | None
family: str | None
context: int | None
input_limit: int | None
output_limit: int | None
knowledge_cutoff: str | None
release_date: str | None
@classmethod
def from_wire(cls, value: dict[str, typing.Any]) -> Model:
@classmethod
def many( cls, value: list[dict[str, typing.Any]]) -> tuple[Model, ...]:
class ProtocolError(blockether.vis.engine.TransportError):

Peer is incompatible or returned malformed protocol data.

Query = collections.abc.Mapping[str, str | int | float | bool | None]
@dataclass(frozen=True, slots=True)
class Response:

Response bytes with their status and headers, before content decoding.

Returned by binary and transcript APIs. Use content directly for files, inspect headers for their media type, or call json for JSON transcripts. json raises ProtocolError for malformed or non-JSON content.

Response(status: int, content: bytes, headers: dict[str, str])
status: int
content: bytes
headers: dict[str, str]
def json(self) -> Any:

Decode content as JSON, or raise ProtocolError if it is invalid.

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

A conversation handle bound to one execution layer and session ID.

Get it from Agent.session, ExecutionLayer.create_session or ExecutionLayer.session. Constructing a handle does not read the session. The handle borrows the layer. Keep the layer open while you use the handle.

Use send for a new turn and turns for history. read fetches the current session state. events follows progress. input_views and answer let your application handle pending human input. transcript, artifacts, upload and download_attachment cover content exchange. council binds messaging and managed-agent controls to this conversation.

Requests return dictionaries in contract form, unless a method documents a typed handle or Response. HTTP errors raise GatewayError, including missing or inaccessible session IDs. No method automatically retries mutations.

Session(client: ExecutionLayer, id: str)
client: ExecutionLayer
id: str
def council(self, *, group_id: str | None = None):

Bind communication to the current run without blocking session-only controls.

Disabled or unavailable Council groups are reported when you use communication or group_id. A successful binding keeps the activation that it had when it was acquired.

def read(self):

Fetch the current session record as a dictionary.

Pending application-owned tool calls are serviced on the calling thread before fetching state. This is a fresh read, not a cached snapshot.

def update(self, **fields):

Patch session fields and return the engine's JSON response.

Field names and validation follow the gateway session-update contract. The engine rejects unknown or invalid fields.

def delete(self):

Ask the engine to delete this conversation and return its response.

This is a destructive session operation, not a connection close. It does not undo changes already made to project files. Do not use this handle for further conversation requests after successful deletion.

def turns(self, **query):

Read turn history, or the queued subset with status='queued'.

def artifacts(self, **query):

Return the gateway's artifact inventory without downloading its bytes.

def transcript( self, *, format: str = 'json', **query) -> Response:

Read a JSON, Markdown or HTML transcript, preserving bytes and headers.

def upload(self, content: bytes, *, filename: str, media_type: str):

Upload bytes. Pass the returned upload_id in send(attachments=[...]).

def download_attachment( self, iteration_id: str, index: int) -> Response:

Download one persisted iteration attachment without decoding binary data.

def goal( self, objective: str, *, iteration_budget: int | None = None, **options) -> Turn:

Submit an explicit goal through /goal and return its execution Turn.

No goal is inferred from ordinary messages. Read its state in read()['goal']. To pause, resume or cancel it, use send('/goal --pause'), --resume or --cancel.

The optional budget counts loop iterations, including prose and empty replies. The last iteration can run its tools, and then the limit blocks the next model request. Token usage is recorded only as a statistic.

def send( self, request: str, *, idempotency_key: str | None = None, **options) -> Turn:

Submit a user message or slash command and return its Turn handle.

Arguments:
  • request: Message text sent to this conversation.
  • idempotency_key: Stable key for retrying the same submission after uncertain IO. If you omit it, each call creates a new key.
  • **options: Turn submission fields such as provider, model and attachments. Upload attachment bytes first with upload, and pass the returned upload IDs in attachments.

A return does not mean that the model has finished. Use Turn.wait or Turn.read. The returned cursor marks the event position before this submission. Pass it to events to follow later progress. Model requests can cost money and change project files. Submission failures propagate, and the SDK never automatically retries this mutation.

def run( self, request: str, *, timeout: float = 300, response_model: type[~ResponseModel] | None = None, max_corrections: int = 2, **options) -> Union[dict, ~ResponseModel]:

Submit a request and wait for its result, as Agent.run does.

Without response_model, return the turn dictionary in contract form. With a Pydantic model, validate the final answer and return the instance. An invalid answer starts up to max_corrections correction turns. Agent.run describes the answer rules, the errors and the timeout.

def model(self) -> SessionModel:

Read the model of the next turn: the pin, else the default pair.

def set_model( self, provider: str | None, model: str | None) -> SessionModel:

Pin the provider and model of the next turns, or clear the pin with None.

Get the names from ExecutionLayer.models. The gateway refuses a provider that it does not serve. A running or queued turn keeps its model. A model request can cost money; Model.pricing gives the rates.

def fork( self, through_turn_id: str | None = None) -> Session:

Copy this conversation into a new independent session and return it.

The fork holds every turn through through_turn_id, or through the last turn when you omit it. The source session does not change.

def export(self) -> bytes:

Return a snapshot of this session and its managed children as bytes.

Save the bytes as a .vissnap file, and restore them with ExecutionLayer.restore_session. The snapshot holds turns, attachments and the Council entries inside the session tree. It does not hold provider credentials or workspace files. The session_snapshot_limit setting limits its size.

def input_views(self) -> list[blockether.vis.views.InputView]:

Read pending input requests as typed blockether.vis.views.InputView records.

Use their view IDs with answer to submit values. An empty list means no input requests were returned in this snapshot.

def live_views( self, placement: str | None = None, archived: bool = False) -> list[blockether.vis.views.LiveView]:

Read current live views as typed blockether.vis.views.LiveView records.

This fetches a snapshot and does not subscribe to future updates. Use events to follow view events. A placement (inline, panel or attachment) keeps only those views. archived=True lists only the archived apps. Malformed view data raises ProtocolError.

def live_view(self, view_id: str) -> blockether.vis.views.LiveView:

Read one live view or app by id, also a closed or restored app.

This is a snapshot, like live_views. Malformed view data raises ProtocolError.

def view_action(self, view_id: str, action: str, **values):

Send an operator action to a view and return the engine's JSON response.

values supplies the action-specific fields of the operator action schema. The request is validated before submission. For the common case of submitting input values, use answer. A refused action of a closed, disconnected or changed app raises GatewayError with status 409 and the refusal reason as code, for example disconnected.

def attach(self, app, *, owner, on_event=None):

Open or adopt an app in this session as owner program owner.

app is a vis.ui.app(...) declaration or an app attachment row. The engine records the owner as sdk/<owner>; owner uses 1 to 64 characters of a-z, 0-9, ., _ and -. A declaration with a key adopts the open app of the same owner and key, so a restarted program gets its app back. Answers the same LiveAttachment handle that extensions use.

on_event(event) runs on a daemon thread that long-polls the app events, until the app closes. A callback error is logged and polling goes on. To serve events on the calling thread, use LiveAttachment.run instead.

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

Read the TUI key bindings that the extensions of this session declare.

Each binding has key, extension and command with label and group, run with label and group, or action. The list is in extension order, then declaration order.

def run_keybinding( self, extension: str, key: str, group: str | None = None) -> dict[str, typing.Any]:

Run the function of one run binding, as if the person pressed its key.

Give the extension, key and group that keybindings() lists. The run adds no turn. The answer is {"status": "ok" | "error", "title", "body"}; a function that fails answers status error.

def answer(self, view_id, values):

Submit input values to a view, returning the engine's response.

Pass a view ID from input_views and values matching that input request. This is shorthand for view_action(view_id, "submit", values=values).

def events(self, **options) -> Events:

Follow this conversation's progress as an iterable of Event records.

Options are cursor (nonnegative replay position, default 0), reconnects (default 3) and retry_delay (seconds, default 0.2). Use a context manager to close the stream when you stop iterating. The stream spans the conversation, not just one turn. To track a specific request, filter on turn_id. Closing the stream does not cancel work.

Gateway sessions normally use SSE. Local execution and application-owned callbacks use finite event-page polling so Python callbacks run on the calling thread. Both expose the same event interface.

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

The model of the next turn of a session.

provider and model are the pin when is_pinned is true, else the default pair. default_provider and default_model are the pair that the turns use without a pin; a project configuration can change it.

SessionModel( provider: str | None, model: str | None, is_pinned: bool, default_provider: str | None = None, default_model: str | None = None)
provider: str | None
model: str | None
is_pinned: bool
default_provider: str | None
default_model: str | None
@classmethod
def from_wire(cls, value: Any) -> SessionModel:
class Speech:

Manage voices, speak text and transcribe audio with the gateway engines.

Without engine, a call uses the engine that the gateway selected. Voice changes use the cloning engine pocket-tts-local by default. Long work runs as a gateway job; these methods wait for it and return the result.

Speech(layer: ExecutionLayer)
def voices( self, engine: str | None = None) -> list[Voice]:

List the voices of engine, imported voices included.

def add_voice( self, recording: str | pathlib._local.Path | bytes, name: str, *, language: str | None = None, transcript: str | None = None, engine: str = 'pocket-tts-local') -> Voice:

Import a recording of one speaker as a voice and return it.

Use 3 to 30 seconds of clean speech. WAV works everywhere; the gateway converts other containers when it can. transcript is what the clip says and improves the clone. A voice with the same id is replaced, so this also replaces the audio of an existing voice. Raises GatewayError 400 for an unusable recording and 409 for an engine that cannot learn voices.

def update_voice( self, voice_id: str, *, name: str | None = <object object>, language: str | None = <object object>, transcript: str | None = <object object>, engine: str = 'pocket-tts-local') -> Voice:

Change the label, language or clip transcript of an imported voice.

An omitted argument keeps its value; None or "" removes it. The id stays the same. Raises GatewayError 404 when the engine has no such imported voice.

def remove_voice(self, voice_id: str, *, engine: str = 'pocket-tts-local') -> bool:

Delete an imported voice. Return False when the engine has no such voice.

def voice_sample(self, voice_id: str, *, engine: str | None = None) -> bytes:

Return a short audio sample of a voice. Raises GatewayError 404 without one.

def model(self, engine: str | None = None, *, prepare: bool = False) -> dict:

Return the readiness of a speaking engine; prepare=True starts its download.

def speak( self, text: str, *, voice: str | None = None, engine: str | None = None, seed: int | None = None, timeout: float = 600) -> SpeechAudio:

Speak text and return the audio, usually 24 kHz mono WAV.

seed selects another take of the same text on engines that sample noise, such as vibevoice-local and qwen3-tts-local. The gateway speaks at most 20000 characters at once. Raises GatewayError 425 while the engine model is not ready and VisTimeout when the job takes longer than timeout seconds.

def transcribe( self, audio: str | pathlib._local.Path | bytes, *, engine: str | None = None, timeout: float = 600) -> Transcript:

Transcribe a 16-bit PCM WAV recording and return its text.

Engines include parakeet-local and vibevoice-asr-local. Raises GatewayError 400 for audio that is not WAV and VisTimeout after timeout seconds.

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

Audio that an engine spoke, as the bytes of one file.

SpeechAudio( content: bytes, media_type: str, engine: str | None, voice: str | None, seed: int | None)
content: bytes
media_type: str
engine: str | None
voice: str | None
seed: int | None
def save(self, path: str | pathlib._local.Path) -> pathlib._local.Path:

Write the audio to path, create missing parents and return the path.

class StructuredOutputError(builtins.ValueError):

No valid structured result: a turn did not complete or answers stayed invalid.

errors lists the problems in the last turn as dictionaries. Each has a JSONPath-like path ($ is the whole value), a message and a source. The source is turn, answer, json, schema or pydantic. attempts has one dictionary per turn. It holds the turn record, the final prose answer (None when absent) and errors. turn is the last record.

The message lists the first problems. Messages include short scalar values from the answer, but never whole objects or arrays. If the response model sets Pydantic's hide_input_in_errors, messages include no answer values.

StructuredOutputError(reason: str, attempts: list[dict])
attempts
turn
errors
@dataclass(frozen=True, slots=True)
class Transcript:

The text that a transcription engine heard in a recording.

Transcript( text: str, engine: str | None, segments: tuple[dict[str, typing.Any], ...] = ())
text: str
engine: str | None
segments: tuple[dict[str, typing.Any], ...]
class TransportError(builtins.RuntimeError):

Connection failed or the client has been closed.

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

Handle for one submitted request, independent of other session turns.

Returned by Agent.send or Session.send. id identifies this request. cursor is the session event position captured before submission. The handle borrows its session and execution layer. Keep that layer open while you call read, wait or cancel.

Turn( session: Session, id: str, cursor: int = 0)
session: Session
id: str
cursor: int
def read(self, *, timeout=None):

Fetch this turn's current record, including its status.

timeout optionally overrides this request's transport deadline in seconds. Application-owned callbacks are serviced before reading state.

def cancel(self):

Request cancellation of this turn and return the engine's response.

Use read or wait to inspect the resulting status. Cancellation does not revert file edits or other side effects that already happened.

def wait(self, *, timeout: float = 300):

Poll this turn until it finishes or suspends, returning its record.

Arguments:
  • timeout: Positive finite overall wait deadline in seconds.
Returns:

A dictionary in contract form. Its status is completed, failed, cancelled, suspended or error. A failure status is data, not a raised model-work exception. Inspect it before you use the result.

Raises:
  • VisTimeout: The deadline expired. This method does not request cancellation. You can wait again or explicitly call cancel.

Transport and HTTP errors also propagate. For a local engine, a timeout of the underlying pipe operation closes that process. This keeps replies in order. It is different from only ending the poll loop.

class VisTimeout(blockether.vis.engine.TransportError):

Deadline expired. The remote operation can still be running.

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

One voice that a speaking engine can use.

Voice( id: str, label: str, engine: str, language: str | None = None, is_imported: bool = False, clip_text: str | None = None, license: str | None = None)
id: str
label: str
engine: str
language: str | None
is_imported: bool
clip_text: str | None
license: str | None
def ensure_installed() -> str:

Return the vis-agent command that matches this SDK, and download it when missing.

An installed vis-agent is used only when it runs the same release as the SDK. Otherwise the SDK uses, or downloads, its own copy in $VIS_HOME/sdk/<tag>. A development build of the SDK uses any installed command, or the latest stable release.

Raises:
  • InstallError: The download is turned off with VIS_NO_AUTO_INSTALL, or install() fails.
def final_answer(turn: dict) -> str | None:

Return the final answer text of the turn record turn, or None.

A turn record has no answer field. The answer is the Markdown of the last prose block in turn["content"]. A turn without that block, or with only blank text in it, gives None.

def install(*, force: bool = False) -> str:

Download the Vis engine of this SDK now and return its vis-agent path.

Use it to prepare a computer or a container image before the first run. It downloads into $VIS_HOME/sdk/<tag> even when another Vis is installed, and reuses a complete download unless force is true. A development build of the SDK installs the latest stable release. VIS_NO_AUTO_INSTALL does not apply.

Raises:
  • InstallError: The platform has no engine, the release has no bundle, or the download fails its checks.