splox 0.2.6

Official Python SDK for the Splox API — run workflows, manage chats, and monitor execution

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.

v0.1.0 is a breaking release. The run-execution surface moved to the v2 API (client.runs / client.interactions). Workflow CRUD reads, chats, memory, billing and the MCP module are unchanged. See CHANGELOG.md for the full breaking-change list.

Installation

pip install splox

Quick Start

from splox import SploxClient

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

# Create a run: workflow id ("wf_...") or a raw workflow UUID both work.
run = client.runs.create(
    "wf_01JAZ6Y5M3Q8F7N2R4T6V9W0XC",
    "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

client.capabilities() (GET /v2/capabilities) returns everything needed to assemble a runnable workflow graph: available LLM endpoints (one is marked is_default — the endpoint new agent nodes get automatically), the default model, and the MCP tool servers usable in tool nodes (system system:* servers include their tool list).

caps = client.capabilities()

default_ep = next(e for e in caps.llm_endpoints if e.is_default)
compute = next(s for s in caps.tool_servers if s.id == "system:compute")
print(caps.default_model, default_ep.client, [t.name for t in compute.tools])

# Pick a model from the endpoint's light list, then fetch its parameter schema:
model = default_ep.models[0]                       # LLMModelSummary(id, name)
schema = next(m for m in client.models(default_ep.id) if m.id == model.id).input_schema
# schema is the JSON Schema of the agent node's "additional_llm_config":
#   {"text_llm_model": model.id, "additional_llm_config": {"temperature": 0.2, ...}}

# 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 — omitted fields fall back to the defaults above:
            "text_llm_endpoint_id": default_ep.id,
            "text_llm_model": caps.default_model,
        }},
        {"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()

Workflow provisioning (v1, unchanged)

Workflow CRUD reads remain available for provisioning:

workflows = client.workflows.list(search="Test")
full = client.workflows.get(workflow_id)
version = client.workflows.get_latest_version(workflow_id)
entry_nodes = client.workflows.get_entry_nodes(version.id)
versions = client.workflows.list_versions(workflow_id)
# plus workflow secrets management: list_secrets / set_env_secret / ...

Chats (client.chats), memory (client.memory), billing (client.billing) and webhooks (client.events) are also unchanged.

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 (v1 provisioning reads)

Method	Description
list(...) / get(id)	List/get workflows
get_latest_version(id) / list_versions(id)	Version reads
get_entry_nodes(version_id)	Entry nodes for a version
list_secrets / set_env_secret / set_file_secret / delete_secret / ...	Secrets management

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

Unchanged from 0.0.x — see the sections above.

License

MIT