Python SDK¶

pip install agenttier gives you sync and async clients with typed Pydantic models, auto-detected auth, streaming HTTP, and a typed exception hierarchy. The same install also ships the agenttier CLI on your PATH.

Source: python-sdk/. PyPI: pypi.org/project/agenttier.

Install¶

pip install agenttier

Supported Python: 3.10, 3.11, 3.12, 3.13. Runtime deps are httpx and pydantic — nothing else.

Hello sandbox¶

from agenttier import AgentTierClient

with AgentTierClient(api_url="https://agenttier.company.com") as client:
    sandbox = client.create_sandbox(template="general-coding", name="hello")
    sandbox.wait_until_running()

    result = sandbox.exec("uname -a")
    print(result.stdout.strip())
    print("exit:", result.exit_code)

    sandbox.terminate()

Authentication¶

The SDK auto-detects credentials in priority order:

AGENTTIER_API_KEY env var → sent as X-API-Key.
AGENTTIER_TOKEN env var → sent as Authorization: Bearer <token> (OIDC JWT).
In-cluster ServiceAccount token at /var/run/secrets/kubernetes.io/serviceaccount/token.
Unauthenticated — accepted only in the Router's dev mode.

Or pass an explicit provider:

from agenttier import AgentTierClient, APIKeyAuth, BearerTokenAuth

# API key
client = AgentTierClient(
    api_url="https://agenttier.company.com",
    auth=APIKeyAuth("sk_live_abc123"),
)

# OIDC JWT you fetched yourself
client = AgentTierClient(
    api_url="https://agenttier.company.com",
    auth=BearerTokenAuth(jwt_string),
)

Async¶

import asyncio
from agenttier import AsyncAgentTierClient

async def main():
    async with AsyncAgentTierClient(api_url="https://agenttier.company.com") as client:
        sandbox = await client.create_sandbox(
            template="general-coding", name="async-demo",
        )
        await sandbox.wait_until_running()
        result = await sandbox.exec("python -c 'print(2+2)'")
        print(result.stdout)
        await sandbox.terminate()

asyncio.run(main())

Both clients share the same exception hierarchy and model types.

Agent mode¶

mode: agent sandboxes are driven through sandbox.agent.configure(...) and sandbox.agent.invoke(...) instead of the terminal. The SDK ships sync and async surfaces; both speak the Router's Server-Sent Events wire format under the hood and return typed Pydantic models.

from agenttier import AgentTierClient

with AgentTierClient(api_url="https://agenttier.company.com") as client:
    sb = client.create_sandbox(template="langgraph-agent", name="my-agent")
    sb.wait_until_running()

    # 1. Upload code + run install + persist the entrypoint
    sb.agent.configure(
        files=[
            ("/workspace/agent.py", "./agent.py"),  # local path
            {"path": "/workspace/requirements.txt", "content": "langgraph\nhttpx\n"},
        ],
        install_command=["pip", "install", "-r", "/workspace/requirements.txt"],
        entrypoint=["python", "/workspace/agent.py"],
        on_log=lambda stream, line: print(f"[install:{stream}] {line}"),
    )

    # 2. Invoke and aggregate the result
    result = sb.agent.invoke({"prompt": "summarize this PR"})
    print(result.stdout)
    print("exit:", result.exit_code, "duration:", result.duration_ms, "ms")

    # 3. Or stream events live for a richer UI
    for event in sb.agent.invoke_stream({"prompt": "..."}):
        if event.event == "log":
            print(event.data["data"])
        elif event.event == "exit":
            print("done:", event.data["exitCode"])

configure accepts:

files: list of (path, source) tuples or {"path": ..., "content": ...} / {"path": ..., "contentBase64": ...} dicts. Tuple sources can be local paths (str / Path) or bytes. Binary input auto-base64s.
install_command: argv list run once. Idempotent — re-running with the same files + command is a no-op.
entrypoint: argv list persisted onto the sandbox so invoke() knows what to run.
on_log=callback(stream, line): optional live install logs.

invoke accepts a payload (dict / str / bytes), an optional prompt (appended as --prompt= to argv and fed to stdin when payload is empty), and an invoke_timeout Go duration string (e.g. "5m") to lower the server-side cap below the template's default.

invoke_cancel(invoke_id) terminates an in-flight invoke. Pair it with the invoke_id from the first event of invoke_stream. Best-effort: returns silently on success, raises NotFoundError if the invoke already completed.

Async users get the same shapes via await sandbox.agent.configure(...), await sandbox.agent.invoke(...), and async for event in sandbox.agent.invoke_stream(...).

See agent-memory.md for the three patterns AgentTier supports for agent memory.

API reference¶

Clients¶

AgentTierClient(api_url, auth=None, timeout=30.0, verify=True)
AsyncAgentTierClient(api_url, auth=None, timeout=30.0, verify=True)

Use them as context managers (with / async with) so the underlying HTTPX client closes cleanly. Methods:

Method	Returns
`create_sandbox(template, name, namespace="default", timeout=None, idle_timeout=None, storage_size=None)`	`Sandbox`
`list_sandboxes(namespace=None, status=None)`	`list[SandboxSummary]`
`get_sandbox(sandbox_id)`	`Sandbox`
`list_templates()`	`list[Template]`
`get_template(name)`	`Template`
`current_user()`	`CurrentUser` (includes `is_admin` bit)

Sandbox handle¶

sandbox = client.create_sandbox(template="general-coding", name="demo")

# State
sandbox.status()                    # SandboxSummary
sandbox.phase                       # SandboxPhase enum
sandbox.wait_until_running(timeout=120.0)  # SandboxSummary

# Lifecycle
sandbox.stop()
sandbox.resume()
sandbox.terminate()   # alias: .delete()

# Execution
result = sandbox.exec("ls -la /workspace", timeout=30)
# CommandResult(stdout=..., stderr=..., exit_code=...)

# Port forwarding
fp = sandbox.forward_port(8080)
# ForwardedPort(port=8080, protocol='http', internal_url=..., preview_url=...)
sandbox.list_ports()
sandbox.remove_port(8080)

The async counterpart AsyncSandbox has the same methods with await and the same return types.

Models¶

Models are Pydantic v2 objects. Field names are snake_case in Python but accept the Router's camelCase JSON transparently.

SandboxSummary — sandbox_id, name, namespace, status, phase, pod_name, pvc_name, template_ref, created_at, last_activity_at, created_by, message.
SandboxPhase — enum (CREATING, RUNNING, STOPPED, ERROR, DELETING, UNKNOWN).
CommandResult — stdout, stderr, exit_code.
ForwardedPort — port, protocol, preview_url, internal_url.
Template — name, description, image, resource_version, spec (free-form dict).
CurrentUser — sub, email, name, groups, is_admin.
CreatedBy — email, display_name.
AuditEvent — timestamp, event_type, sandbox_id, sandbox_name, namespace, user_email, details.
UsageAnalytics — fleet-wide rollups.

Error handling¶

Every SDK error inherits from AgentTierError. Catch that to handle anything.

from agenttier import (
    AgentTierClient,
    AgentTierError,
    AuthenticationError,
    AuthorizationError,
    PolicyViolationError,
    NotFoundError,
    ConflictError,
    SandboxTimeoutError,
    SandboxErrorState,
    APIError,
)

try:
    sandbox = client.create_sandbox(template="general-coding", name="demo")
    sandbox.wait_until_running(timeout=60)
except PolicyViolationError as e:
    for v in e.violations:
        print(f"Rejected by policy [{v['code']}]: {v['message']}")
except AuthenticationError:
    print("Bad credentials. Check AGENTTIER_API_KEY or AGENTTIER_TOKEN.")
except AuthorizationError:
    print("Authenticated but not authorized.")
except NotFoundError:
    print("Sandbox, template, or resource not found.")
except ConflictError:
    print("Operation is invalid for the current state.")
except SandboxTimeoutError:
    print("Sandbox didn't reach Running in time.")
except SandboxErrorState as e:
    print(f"Sandbox went to Error phase: {e}")
except APIError as e:
    print(f"HTTP {e.status_code}: {e}")
except AgentTierError as e:
    print(f"SDK error: {e}")

PolicyViolationError.violations is a list of {"code": ..., "message": ...} dicts. The codes (user_quota_exceeded, template_not_allowed, cpu_limit_exceeded, etc.) are stable — safe to branch on in UI code. See Governance for the full list.

Real-world snippets¶

Run tests in a sandbox and download the result¶

with AgentTierClient(api_url=API) as client:
    sbx = client.create_sandbox(template="general-coding", name="ci-run-123")
    sbx.wait_until_running()

    # Clone and test
    r = sbx.exec(
        "git clone https://github.com/org/repo /workspace/r && "
        "cd /workspace/r && make test > /tmp/test.log 2>&1; echo $?",
        timeout=300,
    )
    print("tests finished, exit:", r.stdout.strip())

    # Stream the log back via exec (file transfer API lands in 0.2.x).
    log = sbx.exec("cat /tmp/test.log").stdout
    with open("test.log", "w") as f:
        f.write(log)

    sbx.terminate()

Orchestrate a handful of agents in parallel¶

import asyncio
from agenttier import AsyncAgentTierClient

API = "https://agenttier.company.com"
TASKS = ["fix-auth-bug", "add-feature-x", "upgrade-deps"]

async def run_agent(client: AsyncAgentTierClient, name: str) -> str:
    sbx = await client.create_sandbox(template="claude-code-bedrock", name=name)
    try:
        await sbx.wait_until_running()
        r = await sbx.exec(
            f"claude -p 'Task: {name}. Work in /workspace, commit when done.'",
            timeout=1800,
        )
        return r.stdout
    finally:
        await sbx.terminate()

async def main():
    async with AsyncAgentTierClient(api_url=API) as client:
        results = await asyncio.gather(*[run_agent(client, t) for t in TASKS])
    for task, output in zip(TASKS, results):
        print(f"--- {task} ---\n{output[:500]}…")

asyncio.run(main())

Preview a running web server through a forwarded port¶

with AgentTierClient(api_url=API) as client:
    sbx = client.create_sandbox(template="general-coding", name="preview")
    sbx.wait_until_running()

    # Start a server inside the sandbox (fire-and-forget is just a timeout=1).
    sbx.exec("nohup python3 -m http.server 8080 > /tmp/srv.log 2>&1 &", timeout=5)

    port = sbx.forward_port(8080)
    print("Internal URL:", port.internal_url)
    print("Preview URL:", port.preview_url or "(set networking.previewDomain to enable public URL)")

Command-line install¶

Quick one-off without a venv:

python3 -m pip install --user agenttier
python3 -c "from agenttier import __version__; print(__version__)"

For a pinned install in your own project, prefer an explicit range:

# pyproject.toml
[project]
dependencies = ["agenttier>=0.2.0,<0.3"]

Upgrading the SDK¶

pip install --upgrade agenttier

The SDK follows the same version as the rest of the platform, but it is versioned independently on PyPI. You can pair an older SDK with a newer server as long as both support the endpoints you call; the SDK tolerates new server fields gracefully thanks to Pydantic's extra="ignore" setting on every model.

License¶

Apache-2.0.