Vis

Python SDK basics

Use Agent() to embed Vis in your program and run tasks in your current project. To use a gateway that runs separately, give Agent() a GatewayClient. Add your application's functions with extensions=[...]. They keep access to your Python objects in both modes. Both modes provide run(), send() and one conversation for follow-up requests.

For classes, methods, signatures and type annotations, browse the generated Python SDK API reference. The reference follows main, so it can show APIs that the newest PyPI release does not have. Use this guide for installation and task examples.

When to use

For a program in another language, read HTTP API.

Install the SDK

You need Python 3.12 or newer on Linux or macOS:

python3 -m venv .venv
. .venv/bin/activate
python -m pip install --upgrade vis-agent

You do not have to install Vis first. When your program starts a private agent, the SDK uses the Vis engine of its own release:

  • If the installed vis-agent runs that release, the SDK uses it.
  • Otherwise the SDK downloads that release once into ~/.vis/sdk and uses that copy.

The download has only the engine and its Python runtime, not the terminal, web or desktop app. The SDK checks it against the release checksums. It does not change your shell profile or your installed Vis. Only the first run needs network access. To turn off the download, set VIS_NO_AUTO_INSTALL=1.

To download the engine ahead of time, for example in a container image, call install():

from blockether.vis.engine import ensure_installed, install

print(install())  # downloads now and prints the path of the vis-agent command
print(ensure_installed())  # the command that a private agent uses, downloaded if missing

install() always uses the copy in ~/.vis/sdk, also when you have Vis installed. Pass force=True to download it again. Both functions raise InstallError when the download fails.

Before the first task, configure a provider and model. A gateway client needs only the Python package, because the provider runs on the gateway machine.

Requests can incur model charges and use the engine's tools and files. Choose an appropriate account and access policy before you run a task. Keep credentials and engine state outside the project you ask the agent to inspect.

Let your program own a private agent

Save this as local_task.py and run python local_task.py from your project:

# local_task.py
import json

from blockether.vis.engine import Agent


def main():
    with Agent(project=".") as agent:
        result = agent.run("Summarize this project without changing files.")
        print(f"Session: {agent.session.id}")
        print(f"Status: {result['status']}")
        print(json.dumps(result["content"], indent=2))


if __name__ == "__main__":
    main()

. means the current directory when you construct the agent. Agent() is identical. Entering the context starts a private engine with no HTTP listener. Leaving it stops that process and discards its temporary session database. It does not undo file edits or isolate your credentials and configuration.

You should see a session ID, Status: completed and the answer's content blocks. By default, run() returns a turn record, not a text string. A failed, cancelled or suspended turn also returns a record: always check status.

The record has no answer field. The answer is in content, a list of blocks. To get the answer as text, call final_answer(result) from blockether.vis.engine. It returns the Markdown of the last prose block, or None when the turn has no answer.

To continue the conversation, call agent.run() again on the same Agent. Each call is a new turn in the same session. The agent sees your earlier messages, its answers and its tool results:

# conversation.py
from blockether.vis.engine import Agent, final_answer


def main():
    with Agent(project=".") as agent:
        for message in [
            "List the files in this folder. Do not change any file.",
            "Which of those files is the largest?",
        ]:
            result = agent.run(message)
            if result["status"] != "completed":
                raise SystemExit(f"Turn ended with status {result['status']}")
            print(final_answer(result))


if __name__ == "__main__":
    main()

The second answer names one of the files from the first answer. To test such a conversation on many tasks and get a score, read Write a task with several turns.

To keep the agent away from some tools, pass tools. For example, Agent(tools=["-py.repl_start", "-py.repl_stop"]) turns off those two tools of the py extension. Choose single tools describes the names and rules. To change the list later, call agent.session.update(tools=[...]).

Get a validated result

When your program needs fields rather than a prose answer, define a Pydantic model and pass it to Agent.run(). Pydantic is installed with vis-agent. There is no separate extra to enable:

# structured_quote.py
from pydantic import BaseModel, Field

from blockether.vis.engine import Agent


class Quote(BaseModel):
    cents: int = Field(ge=0)
    express: bool


def quote_delivery(agent: Agent) -> Quote:
    return agent.run(
        "For a 1000 gram parcel, quote 1200 cents with express false. "
        "Do not inspect or change files.",
        response_model=Quote,
    )


def main():
    with Agent(project=".") as agent:
        quote = quote_delivery(agent)
        print(quote.cents, quote.express)


if __name__ == "__main__":
    main()

Run python structured_quote.py in a project with a configured provider. A valid result prints 1200 False.

With response_model=Quote, run() does three things. It includes the model's validation JSON Schema in the request. It checks the final prose block against that schema. Then it uses Pydantic to run your Python validators and return a Quote. Without response_model, run() returns the original turn record. Both modes work with a private local engine or a borrowed GatewayClient.

The schema is a request to the model, not provider-enforced JSON mode: Vis uses plain-text completions. The result must be one complete JSON value matching your model (an object for Quote), without a Markdown fence or commentary.

Handle invalid answers

If the answer is not valid JSON, run() asks the agent to correct it in the same conversation. It does the same when the answer does not match the schema or fails one of your validators. The request lists each problem with its location, where $ is the whole value. It also repeats the schema. For example, the answer {"cents": -5, "express": "no"} gets this feedback:

- $.cents: does not satisfy minimum 0 (input: -5)
- $.express: expected boolean, got string (input: "no")

By default, run() sends up to two corrections. To change the limit, set max_corrections. max_corrections=0 accepts only the first answer. Each correction is another model call, which can cost money. The timeout covers the first answer and all corrections.

The agent is asked to keep work it has already done. Corrections do not undo earlier tool calls and file edits. Closing a private agent discards its session database, not those file edits.

run() raises StructuredOutputError in these cases:

  • A turn fails, is cancelled or suspends.
  • The answer is still invalid after the last correction.
  • The timeout leaves no time for another correction.

Its message lists up to 10 problems, and its attributes show what went wrong:

Attribute Contents
errors Problems in the last answer, each with a path, a message and a source
attempts One entry per turn, with its turn record, final prose answer and errors
turn The last turn record

The source names the check that found the problem:

  • turn: a turn that did not complete.
  • answer: a turn without a final prose answer.
  • json: text that is not one JSON value.
  • schema: a schema mismatch.
  • pydantic: your model's validators.

Messages quote short values from the answer, such as numbers and short strings, but never whole objects or arrays. If your model sets Pydantic's hide_input_in_errors, messages quote no values.

Use stdio from the Python SDK

The local Agent example above starts vis-agent stdio for you. In this mode, Python owns a private Vis process and talks to it over standard input and output. It does not start an HTTP gateway, and it is not an MCP server or a terminal chat.

To choose the executable yourself or share one local process between agents, use LocalEngine:

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"])

LocalEngine gives the process a temporary session database and a log file for its standard error. When the context closes, it stops the process and removes both files. If the engine fails, the error shows the exit status and the end of the log.

With the default executable="vis-agent", the SDK uses the installed command or downloads it. To use another Vis build, pass the absolute path of its vis-agent launcher. Pass the launcher, not the bare native binary: the launcher also loads the bundled Python runtime. Starting or stopping this process does not affect a running gateway.

Connect to a gateway and run a task

Use a gateway when conversations must survive your script or be shared with the Vis app. You need only the Python SDK on the client machine, not a local engine. Start a gateway, then set these variables in your private terminal for a local, default-state installation:

export VIS_GATEWAY_URL=http://127.0.0.1:7890
export VIS_GATEWAY_TOKEN="$(cat "$HOME/.vis/gateway.token")"
export VIS_PROJECT_ROOT="$PWD"

Do not print or commit the token. For a remote gateway, use its HTTPS origin. Get its token from the operator through a secure channel. VIS_PROJECT_ROOT must be an absolute path on the gateway machine, not a path on your laptop. Remote Agent rejects . and does not guess a server directory.

Save this independent example as gateway_task.py and run python gateway_task.py:

# gateway_task.py
import json
import os

from blockether.vis.engine import Agent, GatewayClient


def main():
    with GatewayClient(
        os.environ["VIS_GATEWAY_URL"],
        token=os.environ["VIS_GATEWAY_TOKEN"],
    ) as execution_layer:
        with Agent(
            project=os.environ["VIS_PROJECT_ROOT"],
            execution_layer=execution_layer,
        ) as agent:
            result = agent.run("Summarize this project without changing files.")
            print(f"Session: {agent.session.id}")
            print(f"Status: {result['status']}")
            print(json.dumps(result["content"], indent=2))


if __name__ == "__main__":
    main()

The script reads the environment variables. Neither object discovers a gateway or loads a local token. GatewayClient accepts an HTTP(S) origin with no path prefix, query, fragment or URL credentials. TLS verification stays enabled and redirects are refused. Transport options belong to the execution layer, not to Agent.

An Agent borrows the execution layer you pass in. Closing it detaches its application extensions. Closing the outer GatewayClient releases the client lease and closes its streams. Neither action stops the gateway or deletes the saved conversation. A running turn may continue, but it cannot call application functions after their Agent closes.

A GatewayClient continues to work when the gateway restarts. A short outage does not end its client lease. When the gateway no longer holds the lease, the client takes a new lease. Application extensions cannot move to a new lease, so a client with an Agent that has extensions raises TransportError for each request. Then create a new client and a new Agent.

The session uses the app channel, so it is visible in the app on that gateway. Keep the printed session ID: to resume it later, open a GatewayClient(url, token=...) context and use client.session(session_id). Each new Agent creates a new session. It does not implicitly resume an old one.

When the with block starts, GatewayClient calls get_capabilities() and checks the protocol. It then sends the token with each request. The client also has one generated method for each gateway route. The method name is the HTTP method and the words of the route path, for example get_session_usage(). Each generated method returns the parsed JSON answer.

To follow the progress of a turn, read Follow progress.

Give the agent your functions

Pass an Extension to your Agent to let it use business rules, query your data or call your services. You do not need a .vis/extensions/ file or a global registration call. The extension belongs to that Agent's conversation, not to other agents sharing its execution layer.

Your functions run in your application process, on the SDK calling thread. They can capture existing objects such as a database client or an in-memory list. This also works with a remote gateway: function code, closures and local objects are not uploaded. Install their dependencies in your application's Python environment.

These functions run with your application's permissions, outside the model's sandbox. Review what they expose. Arguments and returned data cross to the engine and can become model context. Do not return credentials or unrelated private data.

Register a function

Save this complete application as delivery_task.py. Its closure records quoted weights in a list owned by the application. No code is installed on the gateway:

# delivery_task.py
import blockether.vis.extension as vis
from blockether.vis.engine import Agent


def quote_activity(*, phase, result, **_):
    if phase != "success":
        return None
    return vis.ActivityPresentation("Quote delivery", f"{result} cents")


def make_delivery_extension(quoted_weights: list[int]) -> vis.Extension:
    def delivery_quote(weight_grams: int, *, express: bool = False) -> int:
        """Return a delivery price in cents and remember the quoted weight.

        Charge 500 cents plus 100 per started kilogram. express defaults to
        False; True adds 500 cents. Raise ValueError for zero or negative weight.
        Append valid weights to application memory; no network or file changes.
        """
        if weight_grams <= 0:
            raise ValueError("weight_grams must be positive")
        quoted_weights.append(weight_grams)
        kilograms = (weight_grams + 999) // 1000
        return 500 + 100 * kilograms + (500 if express else 0)

    return vis.Extension(
        name="delivery",
        description="Delivery prices from this application.",
        alias="delivery",
        prompt="Use delivery_quote for delivery prices.",
        symbols=[
            vis.Symbol(
                delivery_quote,
                activity=vis.Activity(
                    label="Quote delivery",
                    show_start=False,
                    render=quote_activity,
                ),
            )
        ],
    )


def quote_delivery(agent: Agent) -> dict:
    result = agent.run(
        "Use delivery_quote to quote express delivery for a 1200-gram parcel. "
        "Report the returned price in cents."
    )
    print("Status:", result["status"])
    print(agent.session.transcript(format="markdown").content.decode())
    return result


def main():
    quoted_weights = []
    extension = make_delivery_extension(quoted_weights)
    with Agent(extensions=[extension]) as agent:
        quote_delivery(agent)
    print("Quoted weights:", quoted_weights)


if __name__ == "__main__":
    main()

vis.Symbol exposes the function's name, annotations and docstring to the agent. Here the callable is delivery_quote, not delivery.delivery_quote, because alias identifies the extension, not a function-name prefix. prompt explains when to use the function. symbols makes it callable.

Every exported function needs an Activity presentation. This quick calculation shows its price when it completes and adds no running indicator. Failures keep their error details.

Ask the agent to use it

Run python delivery_task.py from your project. The function returns 1200 cents, the Activity reads Quote delivery · 1200 cents, and your application prints Quoted weights: [1200] after one call. The agent can discover the contract through apropos and doc, call the function through python_execution, and use its result in the answer. This is a model request and can incur provider charges.

You can also add an extension after entering the Agent context, before its first request: call agent.register_extension(extension). Both forms use the same Extension declaration. For a remote agent, pass the same extension object to Agent(execution_layer=execution_layer, extensions=[extension], ...) inside the gateway context above. It still executes in your application.

Understand callback lifetime and supported declarations

Callbacks run while your program drives the SDK: run(), turn waiting, session or turn reads, and event iteration service pending calls. send() alone does not start a background thread that executes your application code. Keep driving the SDK and keep the application alive while the agent needs its functions.

Closing the Agent detaches its extensions. Disconnecting or cancelling releases the engine's wait, but it cannot stop a synchronous Python function that is already running in your process. A repeated delivery of the same pending call uses its saved result and does not run the function again. This is not an exactly-once guarantee across application restarts or new agent requests.

Client extensions support functions, bound methods and object namespaces declared with Symbol, an explicit Activity for every exported method, and a static prompt. For an object namespace, use Symbol(object, name="inventory") and annotate its methods with @vis.method(activity=...). See the extension API. Host-only activation, ctx, env, providers, op hooks, network filters, slash commands and callable prompts are rejected, not silently ignored. Use an engine-side extension for those features. Its registration entry point is vis.register_extension(...).

Arguments must be JSON data. Results can be JSON values, tuples or dataclass instances. Tuples become lists and dataclass fields become ordinary data, keeping None fields. Live object identity stays in your application, not in the returned value. Async functions are awaited on the SDK calling thread, which must not already be running an asyncio event loop.

Handle failures and choose a lifecycle

Situation Meaning and next step
ProtocolError SDK and gateway protocols disagree. Install compatible versions
GatewayError Inspect status and code for authentication, permissions or request errors
TransportError Read the engine's exit status and standard error in the message, if present. Then check the executable or gateway, network and TLS setup
VisTimeout from run() or turn.wait() The wait ended, but the turn can still be running. Inspect it or call turn.cancel()
StructuredOutputError from run() No valid structured result. Read its errors and attempts
A record whose status is not completed The task did not complete normally. Inspect its content and input requirements

A wait timeout is separate from the execution layer's transport timeout. A local pipe timeout stops its engine. Leaving a default local Agent context also stops unfinished work. An Agent with a borrowed layer leaves that layer running. A remote turn can outlive the client.

When you retry a submission, reuse the same explicit idempotency_key. A new key means a new request.

Catch gateway errors

import os

from blockether.vis.engine import GatewayClient, GatewayError, ProtocolError, TransportError

try:
    with GatewayClient(os.environ["VIS_GATEWAY_URL"], token=os.environ["VIS_GATEWAY_TOKEN"]) as client:
        client.get_capabilities()
except GatewayError as error:
    print("HTTP error:", error.status, error.code)
except ProtocolError:
    print("Update the SDK or the gateway to compatible versions.")
except TransportError:
    print("The gateway is not reachable.")

GatewayError has status and code, but not the message of the gateway. ProtocolError and VisTimeout are kinds of TransportError, so catch them first.

Choose a lifecycle

API Use it for Closing it
Agent(project=".") One local conversation, no gateway or HTTP listener Stops its engine and discards session history
Agent(project=..., execution_layer=layer) One conversation on a caller-owned local engine or gateway client Detaches its application extensions, but leaves the layer and saved conversation open
LocalEngine(executable=..., root=...) Several sessions in one owned stdio process Stops that process and discards its session database
GatewayClient(url, token=...) Persistent or shared sessions on a separately running gateway Closes streams and releases its client lease

Agent never picks a transport for you: pass a LocalEngine or a GatewayClient when the default private engine does not fit. Each layer and its session handles use one calling thread. Closing a layer never deletes a saved conversation. To delete one, call conversation.delete().

See also