splox 0.4.0

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(
    "Summarize the latest sales report",
    workflow_version_id=workflow.workflow_version_id,
)
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(
    input,                               # str, MessageInput, or content parts
    workflow_version_id="wfv_...",    # required immutable execution target
    metadata={"ticket": "T-123"},     # optional client metadata
    conversation_id="conv_...",       # 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_, 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(
            "Hello!", workflow_version_id="wfv_01JAZ6Y5M3Q8F7N2R4T6V9W0XC"
        )

        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("hi", workflow_version_id=workflow_version_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 — 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 atomically with immutable v1.
wf = client.workflows.create(
    "researcher",
    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(
    "How many CPUs does this sandbox have?",
    workflow_version_id=wf.versions[0].id,
).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_version_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. In persisted mode, run_cases() creates a server-side evaluation, launches real runs pinned by one immutable workflow_version_id, 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",
    workflow_version_id=res.workflow_version_id,  # the only execution pin
    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.

Workflow versions API

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

page = client.workflows.list(cursor=None, limit=100)
workflow = client.workflows.get(res.workflow_id)  # metadata + version registry
exact = client.workflow_versions.get(workflow.versions[0].id)  # exact graph
created = client.workflows.create_version(
    workflow.id,
    nodes=[{"id": "agent", "type": "agent", "data": {"system_prompt": "Help."}}],
    edges=[],
)
# client.workflows.delete(workflow.id)

AsyncSploxClient.workflows and .workflow_versions expose async versions of these resources. 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(input, *, workflow_version_id, metadata=, conversation_id=, idempotency_key=)	Create a run pinned to one immutable version
get(run_id)	Get a run with exact version provenance
list(*, status=, workflow_version_id=, created_after=, created_before=, cursor=, limit=)	Cursor-paged listing by exact version
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 / client.workflow_versions

Method	Description
workflows.create(name, description=None, *, nodes, edges=)	Atomically create metadata + immutable v1 graph
workflows.list(...) / workflows.get(wf_id)	List metadata / get metadata plus version registry
workflows.update(wf_id, name=..., description=...)	Update mutable metadata only
workflows.create_version(wf_id, nodes, edges=)	Create a new immutable snapshot
workflow_versions.get(wfv_id)	Read one exact immutable graph
workflows.delete(wf_id)	Delete a workflow lineage

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

Unchanged from 0.0.x — see the sections above.

License

MIT