splox 0.3.6

Build, run, and evaluate Splox agent workflows from Python

Splox Python SDK

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

Primary surfaces:

• client.runs / client.interactions — execute and observe workflows.
• splox.builder — declare and upsert workflow graphs in Python.
• 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, builder

client = SploxClient(api_key="your-api-key")
workflow = builder.upsert(
    client,
    "Quick Start Assistant",
    builder.agent("assistant", "Answer clearly and concisely."),
)

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

# Block until the run finishes (raises SploxTimeoutError on timeout).
run = run.wait(timeout=300)
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(
    workflow_id,                      # "wf_..." or raw UUID
    input,                         # str shorthand, or a full MessageInput dict,
                                   # or a list of content parts
    metadata={"ticket": "T-123"},  # optional client metadata
    conversation_id="conv_...",    # optional: continue a conversation (multi-turn)
    workflow_version=3,               # optional: pin a workflow version
    idempotency_key="my-key",      # optional: auto-generated UUIDv4 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_, wf_, int_, ...). The SDK accepts raw UUIDs for workflow ids and encodes them client-side:

from splox import encode_id, decode_id

encode_id("wf", "019f455e-a84c-7d4c-87b0-c951d38bc224")  # -> "wf_01KX2NXA2C..."
decode_id("wf_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("wf_01JAZ6Y5M3Q8F7N2R4T6V9W0XC", "Hello!")

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

        run = await run.wait(timeout=300)
        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(workflow_id, "hi")
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 — the endpoint new agent nodes get automatically; model choice is equally optional (omit text_llm_model in node data → server default).

# 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 agent node's
# "additional_llm_config"; 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. Build a workflow: an agent wired to compute tools via a tool edge.
wf = client.workflows.create("researcher")
client.workflows.set_graph(
    wf.id,
    nodes=[
        {"id": "agent", "type": "agent", "data": {
            "system_prompt": "You are a research assistant.",
            # optional — omit for server defaults:
            "text_llm_endpoint_id": default_ep.id,
            "text_llm_model": model.id,
        }},
        {"id": "tools", "type": "tool", "data": {
            "mcp_server_id": compute.id,  # "system:compute"
            "allowed_tools": ["compute_exec", "compute_read_file"],
        }},
    ],
    edges=[{"source": "agent", "target": "tools"}],  # tool edge (inferred)
)

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

splox.builder — declare a graph in Python

splox.builder turns nested agent() / tool() specs into one synchronous upsert() call (create or replace by exact workflow name). It accepts ordinary lists, so generated code stays simple. Two pure helpers configure the agent node's context knobs, mirroring the editor's "Rule-based Context" and "Summarization" panels:

from splox import SploxClient, builder

client = SploxClient()  # SPLOX_API_KEY and SPLOX_BASE_URL are read from env
root = builder.agent(
    "assistant",
    "You help the user.",
    tools=[builder.tool("system:compute", allow=["compute_exec"])],
    # Token-based memory: summarize trimmed history instead of dropping it.
    context_memory=builder.summarization(
        "Summarize the conversation so far, keeping open questions.",
        max_tokens=70000, trim_target_percent=30,
    ),
    # Rule-based context: each rule fires only when its condition matches
    # (keyword/regex are instant; intent asks the model).
    context_providers=[
        builder.context_rule(
            text="Today is 2025-01-01.", keywords=["date", "today"]
        ),
        builder.context_rule(
            tool_server="system:compute", tool_name="compute_exec",
            tool_args={"command": "uname -a"},
            intent="the user asks about the machine",
        ),
    ],
)
res = builder.upsert(client, "Assistant", root)

# Safe graph round-trip: response-only fields are removed automatically.
graph = builder.pull(client, res.workflow_id)
builder.upsert(client, "Assistant Copy", graph=graph)

splox.evals — grade a workflow in code

splox.evals is synchronous and grades with ordinary Python—there is no eval DSL. run_cases() returns a [case][attempt] matrix of RunResult objects; judge() grades one output with a persistent, billed judge workflow; and compare() compares paired boolean matrices with pass@k, regressions, and an exact paired p-value.

from splox import evals

inputs = ["Run uname -sm", "Print only the current working directory"]
runs = evals.run_cases(
    client,
    res.workflow_id,
    inputs,
    k=3,
    concurrency=3,
    budget_cap=0.25,  # soft cap; already-running attempts can overshoot it
)

# Deterministic grading is just Python.
passed = [
    [run.status == "succeeded" and "compute_exec" in run.tools_used for run in case]
    for case in runs
]

# LLM grading is explicit and billed. The judge workflow persists in the account.
verdict = evals.judge(
    client,
    runs[0][0].output,
    "The answer reports an operating system and CPU architecture, with no extra prose.",
    threshold=4,
)

# Candidate results must have the same [case][attempt] shape and case order.
candidate_passed = passed  # replace with grades from a candidate workflow
report = evals.compare(passed, candidate_passed)
print(report)

Every run_cases() call creates fresh runs. Set suite_id="my-eval-v1" only when retrying/resuming the same suite; reusing it intentionally reuses the same idempotency keys. budget_cap is a soft launch cap, not a hard billing limit.

Workflow graph API

The builder is the concise path; the underlying v2 resource is also public:

page = client.workflows.list(cursor=None, limit=100)
workflow = client.workflows.get(res.workflow_id)
graph = client.workflows.get_graph(workflow.id)
update = client.workflows.set_graph(
    workflow.id,
    nodes=[{"id": "agent", "type": "agent", "data": {"system_prompt": "Help."}}],
    edges=[],
    publish=True,
)
# client.workflows.delete(workflow.id)

AsyncSploxClient.workflows exposes async versions of this low-level resource. builder and evals are 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

Parameter	Type	Default	Description
api_key	str | None	SPLOX_API_KEY env	API authentication token
base_url	str	SPLOX_BASE_URL env, then https://splox.io/api/v1	API base URL
timeout	float	300.0	Request timeout in seconds

client.runs (v2)

Method	Description
create(workflow_id, input, *, metadata=, conversation_id=, workflow_version=, idempotency_key=)	Create a run; returns a bound Run handle
get(run_id)	Get a run
list(*, status=, workflow_id=, created_after=, created_before=, cursor=, limit=)	Cursor-paged listing (newest first)
cancel(run_id)	Idempotent cancellation
wait(run_id, *, timeout=, poll_interval=)	Poll until terminal
stream_events(run_id, *, cursor=)	SSE stream with auto-reconnect + dedupe
list_events(run_id, *, cursor=, limit=)	Durable JSON event pages
messages / outputs / tree / usage / pending_interactions	Run reads

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)

Method	Description
list(*, status=, run_id=, cursor=, limit=)	Cursor-paged listing (newest first)
get(interaction_id)	Get an interaction
respond(interaction_id, *, type=, ..., response=, idempotency_key=)	Resolve a pending interaction

client.workflows (v2 graph API)

Method	Description
create(name, description=None, *, idempotency_key=None)	Create an empty workflow
list(*, cursor=None, limit=None) / get(id)	List/get workflows
get_graph(id)	Read the active graph
set_graph(id, nodes, edges, *, publish=True)	Create a graph version
delete(id)	Delete a workflow

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

Unchanged from 0.0.x — see the sections above.

License

MIT