maritime (Python SDK)

Official Python SDK for Maritime: provision and drive AI agents on Maritime's serverless infrastructure, straight from your own backend.

Build an app where every one of your users gets their own agent: one call on sign-up spins up an isolated agent on Maritime's fleet; another sends it a message. You never touch a container.

pip install maritime

Python 3.9+. Zero dependencies (pure standard library).

Quick start

from maritime import Maritime

client = Maritime()  # reads MARITIME_API_KEY

# When YOUR user signs up, give them their own agent. provision() is idempotent
# on external_id, so it's safe to call on every login.
agent = client.agents.provision(
    external_id=f"customer_{user.id}",   # your id for this agent
    name=f"assistant-{user.id}",
    template="openclaw",
)

# Talk to it (sleeping agents auto-wake):
reply = client.agents.chat(agent["id"], "Summarize my unread email.")["response"]
print(reply)

Run code in a sandbox

A sandbox is a Linux machine you create, run commands in, and kill. It bills per second only while it runs. After 2 minutes idle, it pauses and keeps its memory, processes, and disk.

To create sandboxes, use an API key with the sandboxes scope on an account with a sandbox plan. Without a plan, create raises plan_required, and the error's checkout_url starts the $20 plan.

from maritime import Maritime

client = Maritime()  # reads MARITIME_API_KEY

with client.sandboxes.create() as sbx:
    result = sbx.run("python3 -c 'print(2+2)'")
    print(result.stdout, result.exit_code)  # "4\n" 0
# the sandbox is killed here

This example costs about $0.0001. A small sandbox costs $0.022 per hour, billed per second, and the block runs for a few seconds.

If the block raises and the kill also fails, with raises your error and attaches the kill failure as err.kill_error.

Stream output, run in the background, and serve a port:

import sys

sbx.run("npm test", timeout_seconds=600, on_stdout=sys.stdout.write)

server = sbx.run("python3 -m http.server 8000", background=True)
print(sbx.url(8000))  # https://8000-sbx-....sb.mtm.sh/?token=...
server.kill()

Other calls on a sandbox: files.read, files.write, files.list, files.remove, files.watch, commands.get, commands.kill, commands.history, terminal, run_code, pause, update, save, info, refresh, and rotate_secret. Account calls: client.snapshots, client.templates, client.keys (with sandbox= for per-sandbox keys), client.sandbox_webhooks, client.secrets, and client.usage. Sandboxes bill through your account's usage plan, managed on the billing page.

Every sandbox failure is a MaritimeSandboxError. Branch on err.code, and show err.hint to people. The SDK retries reads and create, update, run, pause, kill, and save up to 3 attempts with the same idempotency key, so a retried run never starts a second process.

To check a sandbox webhook, pass the raw body and the Maritime-Signature header:

if client.sandbox_webhooks.verify_signature(raw_body, request.headers.get("Maritime-Signature"), secret):
    event = client.sandbox_webhooks.parse_event(raw_body)

Sandbox templates

A template is a root filesystem that you build once, from a public image or a Dockerfile. Every sandbox created from it starts with the same files and the same fixed size. Registering a template needs a paid plan. Template storage is free.

from maritime import Maritime, MaritimeTemplateBuildError

client = Maritime()

client.templates.create(
    name="py-ml",
    dockerfile="FROM python:3.12\nRUN pip install numpy pandas\n",
    size="medium",
)

try:
    template = client.templates.wait("py-ml")
except MaritimeTemplateBuildError as err:
    print(err.error)  # why the build failed
    raise

with client.sandboxes.create(from_="py-ml") as sbx:
    print(sbx.run("python3 -c 'import pandas'").exit_code)  # 0

Pass image="python:3.12" instead of dockerfile to use a public image. Set the size with size, or with vcpu, memory_mib, and disk_gib. A value you leave out comes from small. start_cmd and ready_cmd are optional.

create returns the template with status set to building. wait reads the template every 5 seconds until it is ready, and returns it. If the build fails, wait raises MaritimeTemplateBuildError. If the template is still building after timeout_seconds (40 minutes by default), wait raises MaritimeError. The build continues.

Other calls:

template = client.templates.get("py-ml")
page = client.templates.list(limit=20)  # newest first; stock templates come last
next_page = client.templates.list(cursor=page.next_cursor)
client.templates.delete("py-ml")  # sandboxes created from it keep running

Errors are MaritimeSandboxError:

err.code	Cause
plan_required	The account has no paid plan. err.checkout_url opens the billing page.
concurrency_limit	The account has 100 templates, or 20 templates that are building.
snapshot_exists	A stock template, a template, or a saved snapshot already has the name.
snapshot_not_found	No template has the name.
scope_missing	You tried to delete a stock template.

Authentication

Mint an API key (mk_...) from the dashboard (Settings → API keys) or the CLI (maritime keys create), then set MARITIME_API_KEY, or pass it explicitly:

client = Maritime(api_key="mk_...")

Keys carry scopes: hand a narrower key to a subsystem that only needs part of the surface:

Scope	Grants
provision	create agents
deploy	start/stop/restart/sleep/chat
secrets	read/write env vars
manage	everything, incl. delete + key management (wildcard)

worker = client.keys.create("chat-worker", scopes=["deploy"])
# worker["raw_key"] is shown once, so store it now.

Keys can also be project-scoped: the key to put in a partner-facing or per-tenant service. A scoped key only works on the front door + usage endpoints for that one project and is rejected everywhere else (other projects 404), so a leak means "message your own fleet", not your wallet:

runtime = client.keys.create("poke-runtime", scopes=["deploy", "manage"],
                             project_id=agent["projectId"])

Agents

# Create (kicks off deploy). Prefer template: a bare framework yields a broken image.
agent = client.agents.create(
    "support-bot",
    template="openclaw",
    external_id="customer_42",
    instructions="You are a friendly support agent for Acme Inc.",
    env=[{"key": "ACME_API_KEY", "value": "...", "secret": True}],
    idle_ttl_seconds=3600,   # 0 = always-on
)

# Idempotent get-or-create by external_id (recommended for per-user provisioning).
# The server does the get-or-create: a repeat call returns the same agent.
agent = client.agents.provision(external_id="customer_42", name="support-bot")

client.agents.get(agent["id"])
client.agents.list(external_id="customer_42")   # filter by your id
client.agents.list()                             # all of your agents

# Chat (synchronous; auto-wakes a sleeping agent)
result = client.agents.chat(agent["id"], "Hello")
print(result["response"])   # None + result["error"] if delivery failed

# Lifecycle
client.agents.start(agent["id"])
client.agents.sleep(agent["id"])     # cheapest resting state (serverless snapshot)
client.agents.restart(agent["id"])
client.agents.delete(agent["id"])    # tears down container + volume + network

# Env vars (secrets encrypted at rest; reach a running container after reload_env)
client.agents.set_env(agent["id"], "STRIPE_KEY", "sk_live_...", secret=True)
client.agents.list_env(agent["id"])
client.agents.reload_env(agent["id"])

# Logs
client.agents.logs(agent["id"], limit=100, level="error")

# Real-world identity (Identity add-on agents): phone, email, tunnel host.
# Channels never provisioned come back None; a fresh phone number reports
# smsStatus "pending" during carrier warm-up.
identity = client.agents.identity(agent["id"])
identity["phoneNumber"], identity["emailAddress"]

# Skills: list, search ClawHub, install, upload a custom skill, export.
# Install/upload auto-resolve missing binaries inside the agent's own VM
# and report the final state; "needs_input" means the skill just needs
# the listed env values (an API key, say) from your user.
overview = client.agents.skills.list(agent["id"])
result = client.agents.skills.upload(agent["id"], open("daily-brief.zip", "rb").read())
if result["status"] == "needs_input":
    client.agents.skills.set_env(agent["id"], {"SOME_API_KEY": value})
tarball = client.agents.skills.export(agent["id"], result["name"])
client.agents.skills.remove(agent["id"], result["name"])

# Files: browse the agent's persistent volume, move bytes both ways, and
# run one-shot commands. File operations wake a sleeping agent; transfers
# are capped at 100 MB per file.
listing = client.agents.files.list(agent["id"])          # {path, root, entries}
root = listing["root"]                                   # /data for most frameworks
client.agents.files.upload(
    agent["id"], open("report.csv", "rb").read(), "report.csv",
    dest_dir=f"{root}/inbox",     # omit dest_dir to deliver as a chat attachment
)
client.agents.files.write(agent["id"], f"{root}/notes.md", "# notes")
data = client.agents.files.download(agent["id"], f"{root}/notes.md")
client.agents.files.move(agent["id"], f"{root}/notes.md", f"{root}/docs/notes.md")
client.agents.files.delete(agent["id"], f"{root}/docs/notes.md")
result = client.agents.exec(agent["id"], ["ls", "-la", root])
result["exitCode"], result["stdout"]

Agent URLs: open your agent's web app

Every agent with an exposed_port has its own hostname, https://<agent-id>.mtm.sh/, in agent["webUrl"]. It is private by default: a visitor needs a link your backend minted.

# Your backend, when a signed-in user opens their workspace:
link = client.agents.share_links.create(
    agent["id"],
    end_user_external_id="customer_42",   # dies with the end user's suspension
    ttl_seconds=86400,                    # default 3600; None for a link that never expires
)
redirect(link["url"])                     # the first visit swaps the secret for a cookie

# Housekeeping
client.agents.share_links.list(agent["id"])
client.agents.share_links.revoke(agent["id"], link["id"])
client.agents.share_links.revoke_for_end_user(agent["id"], "customer_42")

# Or let anyone with the URL in, and let a visit wake the agent:
client.agents.set_visibility(agent["id"], "public")

The app behind the URL sees its own cookies, its own Host, and WebSockets work. Search engines are told not to index agent URLs.

Run commands

Attached execution is the default. Omitting the execution timeout selects 60 seconds. Stdout and stderr are separate, and each returned stream is limited to 262,144 characters plus a truncation marker. On agents with detached support, every accepted command has an execution ID. Results are temporary and can be lost on restart or agent replacement.

client.agents.exec() accepts detached=True and a positive timeout in seconds. Omitted timeout and None select 60 seconds when attached or 3,600 seconds when detached. If your installed SDK rejects detached, submit through the REST API.

execution = client.agents.exec(
	agent["id"], ["sh", "-lc", "npm run build"],
	detached=True, timeout=600,
)
execution_id = execution["executionId"]
print(execution_id, execution["status"])

The Python and TypeScript SDKs do not provide named helpers for result retrieval or cancellation. Use GET /api/agents/{agent_id}/exec/{execution_id} to poll and DELETE on the same path to cancel. TypeScript's agents.exec() currently supports attached execution only, so use REST for detached submission as well. The API reference includes all three requests and their response shapes.

The SDK client's request timeout is separate from the command's execution timeout. For a long attached call, set the client timeout above the execution timeout with room for agent wake-up. Python's client timeout is in seconds; TypeScript's is in milliseconds. Detached submission avoids waiting for completion within that request.

When detached execution is supported, disconnecting does not cancel commands in either mode. Repeating submission creates another execution, so do not resubmit solely because a response was lost. Some older agents support attached execution only, with a 120-second timeout cap, combined output in stdout, and no execution ID or status. Detached submission returns HTTP 501 with unsupported: outdated vm. Please use a new agent. Use a new agent for detached execution and retained results.

Swarms: one command in many sandboxes

A swarm runs one shell command in N short-lived sandboxes at once and ends them when they finish. Each sandbox sees MARITIME_SWARM_INDEX (0 to N-1), MARITIME_SWARM_COUNT and MARITIME_SWARM_ID. Pass inputs= instead of count= to give each sandbox its own args, env and files. Sandboxes are billed per second at the sandbox rates, and starting a run needs a key with the provision scope.

run = client.swarms.start(
    'echo "case $MARITIME_SWARM_INDEX" && test $((MARITIME_SWARM_INDEX % 3)) -ne 0',
    count=10,
    timeout_seconds=300,
)

done = client.swarms.wait(run["id"], timeout=15 * 60)
print(done["status"], done["counts"]["succeeded"], "succeeded,", done["counts"]["failed"], "failed")
for sandbox in done["sandboxes"]:
    print(sandbox["index"], sandbox["exitCode"], sandbox["stdoutTail"].strip())

start() sends an Idempotency-Key (pass idempotency_key= to choose it), so a retried start returns the same run. wait() polls every 3 seconds until the run is completed or cancelled. list(limit=), get(id) and cancel(id) cover the rest, and results(id) returns every sandbox (exit code, output tails, the result_path file) as a list of dicts. Each sandbox carries a sandboxId (sbx-...) that client.sandboxes.get() can read for 7 days after its job ends. A run's size is {"name", "vcpus", "memoryGb"}, where name is the smallest sandbox size that fits (default small: 2 vCPU, 4 GB). template defaults to base.

Sandboxes with a fixed lifetime

client.ephemeral.create_ephemeral() starts a sandbox with a fixed lifetime. Maritime ends it, disk included, when the lifetime ends (5 to 1440 minutes, default 60). Its id (sbx-...) works with every client.sandboxes call, so run commands and use files through a Sandbox client. Every call needs a key with the sandboxes scope. A small sandbox (2 vCPU, 4 GB) costs $0.022 per hour, billed per second.

box = client.ephemeral.create_ephemeral(name="ci-run", lifetime_minutes=30, size="small")
sbx = client.sandboxes.connect(box["id"])
print(sbx.run("python3 --version").stdout, "ends at", box["expiresAt"])

create_ephemeral() takes the client.sandboxes.create() fields except the idle timer ones (name, from_ or source, size, screen, network, end_user, for_end_user, env, metadata, fail_fast), plus lifetime_minutes. The older mem_mb, vcpus, disk_gb and external_id still work: the sizing fields pick the smallest size that fits.

Every call returns the Sandbox object as a dict with expiresAt added. list_ephemeral() returns every live sandbox, including ones made with client.sandboxes.create(); expiresAt is None for a sandbox that never ends by itself. extend_ephemeral(sandbox_id, lifetime_minutes) restarts the lifetime from now. kill_ephemeral(sandbox_id) ends the sandbox at once and is safe to repeat.

Computers on demand: Linux and Windows desktops, and Android phones

client.computers.create() takes os ("linux", "windows" or "android", default linux) and arch ("x86_64" or "arm64", default x86_64). arm64 is for Android apps that ship only ARM code, and it runs only where your account has ARM hosts; otherwise the API returns 400 unsupported_arch. The default x86_64 Android image has no ARM translator, so apps with ARM-only native libraries do not run on it. Both apply to a new computer only: an existing computer keeps its own. list(), get(id), wake(id), sleep(id) and delete(id) work as in the TypeScript SDK.

windows = client.computers.create(external_user_id=user_id, os="windows")

phone = client.computers.create(name="arm-apps", os="android", arch="arm64")
print(phone["os"], phone["arch"])

Projects: one agent per end-user (the front door)

Route your end-users' messages through a project: each external_user_id gets a sticky binding to its own agent (spawned from your template, warm-pool backed). Full guide: docs/FRONT_DOOR.md.

# One-time: fleet policy on the agent's auto-created project
client.projects.update(agent["projectId"],
                       new_chat_policy="spawn",  # dedicated agent per end-user
                       warm_pool_size=3)         # new users bind in ~1s

# Per message: get-or-create + wake + deliver
msg = client.projects.message(agent["projectId"],
                              external_user_id=f"user_{uid}",
                              message="What does my dashboard say?",
                              idempotency_key=request_id)
if msg["status"] == "replied":
    print(msg["reply"])
# "provisioning" | "queued" -> reply arrives via the message.reply webhook,
# or poll: client.projects.get_message(project_id, msg["messageId"])

# Your customer list
client.projects.users(project_id)
client.projects.unbind_user(project_id, "user_42")

Subscribe to replies (and verify deliveries) with the webhook helper:

from maritime import verify_webhook_signature

hook = client.webhooks.create("https://api.acme.co/maritime/webhook",
                              events=["message.reply", "message.failed"])
# in your handler -- pass the RAW body:
ok = verify_webhook_signature(hook["secret"], raw_body, sig_header)

Billing: rebill your users

Every account created after July 2026 is on a plan: a flat subscription that includes a number of machines, where a machine is one micro-VM. An agent is a machine and a Computer is a machine, so one plan covers both and nothing is metered per action or per session. The usage report below works on any account. set_auto_recharge applies only to the grandfathered wallet model, which no new account uses.

# Per-agent / per-end-user cost over a date range (max 92 days), from the
# real money ledger. Rows carry externalUserId, so invoicing is a one-liner.
report = client.billing.usage(from_="2026-07-01", to="2026-08-01")
for row in report["agents"]:
    invoice(row["externalUserId"], row["totalCostCents"])

# Auto-recharge (wallet accounts only): below threshold_cents, Maritime
# charges your saved card for amount_cents (max 3x/24h, 6h backoff after a
# failure). Kills the "wallet hits zero and everything sleeps" cliff.
client.billing.set_auto_recharge(enabled=True, threshold_cents=2000, amount_cents=10000)

# Per-end-user compute caps: stamped on every NEW instance the project
# spawns; over-cap instances auto-sleep instead of running up a bill.
client.projects.update(project_id, instance_compute_minutes_limit=300)

Scheduled wakes (agent-side)

These helpers run INSIDE a Maritime agent (a BYO image speaking the contract). A Maritime micro-VM sleeps at ~zero cost between messages, so a plain timer thread never fires while it's snapshotted; instead, publish your schedule and Maritime wakes the VM at the right moments.

from maritime import observe_scheduler, push_schedules

# One line: every add/remove on your scheduler re-publishes the snapshot.
observe_scheduler(
    my_scheduler,
    get_snapshot=lambda s: [
        {"id": j.id, "nextRunAt": j.next_run.isoformat()} for j in s.jobs()
    ],
)

# Or push explicitly (send the FULL list; [] clears all synced wakes):
push_schedules([
    {"id": "digest", "cron": "0 9 * * 1-5", "tz": "America/New_York",
     "prompt": "Send the digest"},
    {"id": "followup", "nextRunAt": "2026-08-01T14:00:00Z"},
])

Entries use nextRunAt (the next occurrence as YOUR scheduler computed it) or a 5-field cron + tz. An entry with prompt gets it delivered to your POST /chat (source "scheduled") right after the wake. Credentials come from the env every Maritime agent already has; off-Maritime both helpers are silent no-ops, so they're safe in local dev. Alternative with no SDK at all: serve GET /schedules returning the same array and Maritime polls it.

Errors

Every failure is a subclass of MaritimeError: catch the base to catch them all, or narrow by type:

from maritime import (
    MaritimeAuthError,            # 401, or a 403 for an under-scoped key
    MaritimePaymentRequiredError, # 402: plan limit reached or wallet needs funding
    MaritimeNotFoundError,        # 404: no such agent (or not yours)
    MaritimeConflictError,        # 409: name already taken
    MaritimeRateLimitError,       # 429
    MaritimeAPIError,             # any other non-2xx (.status, .detail, .code, .retry_after_seconds)
    MaritimeConnectionError,      # never reached Maritime (network/timeout)
)

try:
    client.agents.create("dupe", template="openclaw")
except MaritimeConflictError:
    ...  # an agent with that name already exists
except MaritimeAPIError as err:
    print(err.status, err.code, err.detail, err.request_id)

Refusals carry a stable code beside the sentence: branch on err.code, not on err.detail. For example a 503 capacity_unavailable (retry after err.retry_after_seconds), a 403 machine_unavailable or swarm_too_large (a plan answer, not a key problem, so a plain MaritimeAPIError), and a 409 idempotency_conflict or confirm_required.

Configuration

Maritime(
    api_key="mk_...",                    # or MARITIME_API_KEY
    base_url="https://api.maritime.sh",  # or MARITIME_API_URL
    timeout=60.0,                        # per-request seconds
    max_retries=2,                       # network + 5xx/429 (GET/DELETE and 429/503 only)
    default_headers={"x-team": "acme"},
)

GET and DELETE retry on any transient failure. A POST or PUT retries only on 429, 503, and connection errors raised before the request was sent. It never retries after a timeout or a 5xx, because the server may already have applied the write. agents.chat sends an Idempotency-Key, so a retried chat returns the first turn's reply instead of running the turn again.

License

MIT