Splox Python SDK

Official Python SDK for the Splox API — create and observe harness runs, resolve human-in-the-loop interactions, browse the MCP catalog, and manage harnesses programmatically.

Primary surfaces:

• client.runs / client.interactions — execute and observe harnesses.
• client.harnesses — create a harness from its files: the programs that declare its agents.
• splox.evals — run, grade, and compare evaluation suites.
• client.llm_endpoints / client.models / client.tool_servers — discovery.

Installation

Python 3.11 or newer is required.

pip install splox

Quick Start

from splox import SploxClient

client = SploxClient(api_key="your-api-key")

# A harness is its files. Both the model and the credential are needed at run
# time, so ask the API which ones this account may use.
endpoint = next(e for e in client.llm_endpoints.list() if e.is_default)
model = client.llm_endpoints.models(endpoint.id)[0]

# A harness is programs: programs/splox/main.py declares the agents in python
# and says with handle() which of them answers a chat.
program = f"""
from splox import agent

assistant = agent(
    "Assistant",
    system_prompt="Answer clearly and concisely.",
    model={model.id!r},
    text_llm_endpoint_id={endpoint.id!r},
)


def handle(msg):
    return assistant
"""

harness = client.harnesses.create(
    "Quick Start Assistant",
    files={"programs/splox/main.py": program},
)

run = client.runs.create(
    "Summarize the latest sales report",
    harness_id=harness.id,
)
print(run.id, run.status)  # run_01... queued

# Block until the run reaches a terminal state.
run = run.wait()
print(run.status)  # succeeded | failed | cancelled

# Read the transcript and outputs.
for message in run.messages().data:
    print(f"[{message.role}] {message.text}")
for output in run.outputs().data:
    print(output.type, output.value)

# Usage (includes descendant runs; amount is a decimal string).
usage = run.usage()
print(usage.input_tokens, usage.output_tokens, usage.amount, usage.currency)

Creating runs

run = client.runs.create(
    input,                               # str, MessageInput, or content parts
    harness_id="h_...",               # required; runs the tip of main
    metadata={"ticket": "T-123"},     # optional client metadata
    chat_id="chat_...",       # optional multi-turn continuation
    idempotency_key="my-key",         # optional; generated when omitted
)

input accepts:

"What is 2+2?"                                            # text shorthand
{"role": "user", "content": [{"type": "text", "text": "hi"}]}  # full MessageInput
[{"type": "text", "text": "hi"}, {"type": "json", "value": {"x": 1}}]  # parts

Every POST /v2/runs carries an Idempotency-Key (auto UUIDv4). Retrying with the same key and body returns the same run; the SDK only ever retries POSTs that carry an idempotency key.

Streaming events (SSE)

# Live event stream: auto-reconnects with Last-Event-ID, dedupes by event id,
# and ends after the terminal run.status_changed event.
for event in run.events():          # stream=True is the default
    print(event.sequence, event.type, event.data)

# Durable JSON pages instead of a stream:
page = run.events(stream=False, limit=100)
for event in page.data:
    print(event.sequence, event.type)

Listing and cancelling

page = client.runs.list(status=["running", "waiting"], limit=20)
for run in page.data:
    print(run.id, run.status)
if page.page.has_more:
    page = client.runs.list(status=["running", "waiting"], cursor=page.page.next_cursor)

run = client.runs.cancel(run_id)    # idempotent; terminal runs are returned unchanged
tree = client.runs.tree(run_id)     # run + descendants snapshot

Human-in-the-loop interactions

# Pending questions raised by a run:
for interaction in run.pending_interactions():
    print(interaction.type, interaction.prompt, interaction.payload)

# Respond (variant must match the interaction type):
client.interactions.respond(interaction.id, type="approval", approved=True)
client.interactions.respond(interaction.id, type="text", text="blue")
client.interactions.respond(interaction.id, type="choice", option_ids=["opt_a"])
client.interactions.respond(interaction.id, type="confirmation", confirmed=True)
# ...or pass a prebuilt body:
client.interactions.respond(interaction.id, response={"type": "approval", "approved": False})

# Inbox-style listing:
page = client.interactions.list(status="pending", limit=50)

Public IDs

v2 IDs are <prefix>_<26-char Crockford base32> (run_, h_, int_, ...). The SDK accepts raw UUIDs for harness ids and encodes them client-side:

from splox import encode_id, decode_id

encode_id("h", "019f455e-a84c-7d4c-87b0-c951d38bc224")   # -> "h_01KX2NXA2C..."
decode_id("h_01KX2NXA2CFN68FC69A79RQGH4")                # -> UUID(...)

Async Support

Every resource has an async twin with the same shape:

import asyncio
from splox import AsyncSploxClient

async def main():
    async with AsyncSploxClient(api_key="your-api-key") as client:
        run = await client.runs.create(
            "Hello!", harness_id="h_01JAZ6Y5M3Q8F7N2R4T6V9W0XC"
        )

        async for event in run.events():           # SSE with auto-reconnect
            print(event.type, event.data)

        run = await run.wait()
        page = await run.messages()
        print([m.text for m in page.data])

asyncio.run(main())

Error Handling

All non-2xx v2 responses are RFC 9457 problem+json and map onto a typed hierarchy; code carries the stable machine-readable error code and trace_id correlates with server logs.

from splox.exceptions import (
    SploxAPIError,          # base for HTTP errors (.status_code/.code/.trace_id/.problem)
    SploxBadRequestError,   # 400
    SploxAuthError,         # 401
    SploxForbiddenError,    # 403
    SploxNotFoundError,     # 404
    SploxConflictError,     # 409  (idempotency_key_conflict, interaction_not_pending, ...)
    SploxGoneError,         # 410  (event_cursor_expired, ...)
    SploxValidationError,   # 422  (.errors = [{name, reason, ...}] with JSON Pointers)
    SploxRateLimitError,    # 429  (.retry_after)
    SploxServerError,       # 5xx
    SploxTimeoutError,      # run.wait()/result() timeouts (also a TimeoutError)
    SploxConnectionError,   # network failures
    SploxStreamError,       # SSE stream gave up reconnecting
)

try:
    run = client.runs.create("hi", harness_id=harness_id)
except SploxValidationError as e:
    for item in e.errors:
        print(item["name"], item["reason"])
except SploxRateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}")
except SploxAPIError as e:
    print(f"API error {e.status_code} ({e.code}): {e.message}")

Retries: GET requests are retried up to 3 times with exponential backoff on connection errors, 429 and 5xx. POSTs are retried only when they carry an Idempotency-Key (always true for runs.create and interactions.respond), reusing the same key so replays are safe.

Discovery & building an agent

Three resource surfaces cover discovery: client.llm_endpoints (GET /v2/llm-endpoints, .models(id)) and client.tool_servers (GET /v2/tool-servers, .tools(id)). One endpoint carries is_default=True. Both the model and the credential are required at run time: an agent that names no model fails its turn with text_llm_model is required, and one that names neither text_llm_provider nor text_llm_endpoint_id fails because nothing says whose credential it runs on.

# 1. LLM endpoints: pick one (or rely on the default).
endpoints = client.llm_endpoints.list()
default_ep = next(e for e in endpoints if e.is_default)
model = client.llm_endpoints.models(default_ep.id)[0]  # Model, all operations
# model.input_schema is the JSON Schema of the model's own
# generation parameters; model.id goes into "text_llm_model".

# 2. Tool servers: system servers (kind="system") + your own (kind="user").
servers = client.tool_servers.list()
compute = next(s for s in servers if s.id == "system:compute")
tool_names = [t.name for t in client.tool_servers.tools(compute.id)]

# 3. Write the harness: one program, declaring one agent with the tools it may
#    call, and a handle() that hands every message to it.
program = f"""
from splox import agent

researcher = agent(
    "Researcher",
    system_prompt="You are a research assistant.",
    model={model.id!r},
    text_llm_endpoint_id={default_ep.id!r},
    tools=[{compute.id!r}],
)


def handle(msg):
    return researcher
"""

wf = client.harnesses.create("researcher", files={"programs/splox/main.py": program})

run = client.runs.create(
    "How many CPUs does this sandbox have?",
    harness_id=wf.id,
).wait()

Read a version back with client.harness_versions.get(harness_id, commit); its files are the version — every path in the commit against that file's contents, the same bytes the runtime parses.

splox.evals — grade a harness in code

splox.evals is synchronous and grades with ordinary Python—there is no eval DSL. In persisted mode, run_cases() creates a server-side evaluation against one harness_id, launches real runs at the tip of that harness's main branch, records scorer verdicts, and returns the familiar [case][attempt] grid with authoritative metrics on grid.evaluation.

from splox import evals

inputs = ["Run uname -sm", "Print only the current working directory"]
expected = [
    {"required_tool": "compute_exec"},
    {"required_tool": "compute_exec"},
]


def grade(result, case):
    passed = (
        result.status == "succeeded"
        and case.expected["required_tool"] in result.tools_used
    )
    return evals.Score(
        verdict="passed" if passed else "failed",
        score="1.00" if passed else "0.00",
        feedback="required compute tool observed" if passed else "compute tool missing",
    )


grid = evals.run_cases(
    client,
    inputs,
    expected=expected,
    k=3,
    concurrency=3,
    budget_cap=0.25,  # soft cap; already-running attempts can overshoot it
    suite_id="compute-contract-v1",
    harness_id=wf.id,                 # the harness under test
    scorer=evals.Scorer(id="compute-contract", version="v1", score_fn=grade),
)

print(grid.evaluation.status)
print(grid.evaluation.metrics.pass_at_1)

# Standalone LLM grading remains explicit and billed.
verdict = evals.judge(
    client,
    grid[0][0].output,
    "The answer reports an operating system and CPU architecture, with no extra prose.",
    threshold=4,
)

Use a stable suite_id to retry/resume the same immutable suite without re-running recorded attempts. Change the suite id or version when its definition changes. budget_cap is a soft launch cap, not a hard billing limit. Local compare() remains available for arbitrary paired boolean matrices and p-values.

Harness versions API

A harness (h_...) is {id, name, versions} — no description, no timestamps, because the prose lives in the files and the age is the commit, and nothing makes the name unique either. A version is {number, commit}: the commit is the tree, the number is the label refs/versions/N puts on it for people. A run executes the tip of main.

page = client.harnesses.list(cursor=None, limit=100)
harness = client.harnesses.get(wf.id)              # id, name, version registry
harness.name                                       # "researcher"
harness.versions[0].commit                         # the commit that IS the tree
exact = client.harness_versions.get(harness.id, harness.versions[0].commit)
exact.files                                        # the version's files, path -> contents
client.harnesses.update(harness.id, name="Renamed")  # a rename; nothing else
# client.harnesses.delete(harness.id)

AsyncSploxClient.harnesses and .harness_versions expose async versions of these resources. evals is intentionally synchronous.

MCP (Model Context Protocol) — unchanged

The MCP module is fully supported and unchanged in this release: catalog, connections, tool execution, OAuth and connection links work exactly as in 0.0.x.

Catalog

# Search the MCP catalog
catalog = client.mcp.list_catalog(search="github", per_page=10)
for server in catalog.mcp_servers:
    print(f"{server.name} — {server.url}")

# Get featured servers
featured = client.mcp.list_catalog(featured=True)

# Get a single catalog item
item = client.mcp.get_catalog_item("mcp-server-id")
print(item.name, item.auth_type)

Connections & tools

conns = client.mcp.list_connections()
owner_servers = client.mcp.list_connections(scope="owner_user")

tools = client.mcp.get_server_tools("mcp-server-id")

result = client.mcp.execute_tool(
    mcp_server_id="mcp-server-id",
    tool_slug="list_servers",
    args={"query": "x"},
)
print(result.result.content, result.result.structured_content, result.result.is_error)

client.mcp.delete_connection("connection-id")

Connection Token & Link

from splox import generate_connection_token, generate_connection_link

token = generate_connection_token(
    mcp_server_id="mcp-server-id",
    owner_user_id="owner-user-id",
    end_user_id="end-user-id",
    credentials_encryption_key="your-credentials-encryption-key",
)

link = generate_connection_link(
    base_url="https://app.splox.io",
    mcp_server_id="mcp-server-id",
    owner_user_id="owner-user-id",
    end_user_id="end-user-id",
    credentials_encryption_key="your-credentials-encryption-key",
)
# → https://app.splox.io/tools/connect?token=eyJhbG...

Webhooks

from splox import SploxClient

client = SploxClient()  # No API key needed for webhooks

result = client.events.send(
    webhook_id="your-webhook-id",
    payload={"order_id": "12345", "status": "paid"},
)
print(result.event_id)

Custom Base URL

client = SploxClient(
    api_key="your-api-key",
    base_url="https://your-self-hosted-instance.com/api/v1",
)

v1 endpoints use the base URL as-is; v2 endpoints (/v2/...) are resolved against the server origin (the base URL with its /api/v1 suffix stripped).

API Reference

SploxClient / AsyncSploxClient

client.runs (v2)

Run handles expose the same operations instance-bound: run.wait(), run.cancel(), run.events(), run.messages(), run.outputs(), run.tree(), run.usage(), run.pending_interactions(), run.refresh().

client.interactions (v2)

client.harnesses / client.harness_versions

client.chats, client.memory, client.billing, client.mcp

Unchanged from 0.0.x — see the sections above.

License

MIT