Python

Use SimStudioClient to execute deployed workflows from a Python application. It supports synchronous requests, server-queued asynchronous runs, rate-limit retries, and file-like inputs. Its HTTP methods are blocking; asynchronous execution does not make it an asyncio client.

Install and configure

The package requires Python 3.8+ and requests>=2.25.0:

pip install simstudio-sdk

Create a key in Account settings → Sim API keys, or use a workspace key if your administrator requires one. Keep it on your server as SIM_API_KEY. Set WORKFLOW_ID to a deployed workflow you can access. See Authentication for key types and permissions.

Server compatibility

SDK 0.2.x executes and polls through /api/v2. Your deployment must serve POST /api/v2/workflows/{workflowId}/execute; the SDK does not fall back when that route is unavailable. Upgrade an older self-hosted server, or intentionally retain simstudio-sdk<0.2 for its legacy execution endpoints.

The default origin is https://www.sim.ai. Set base_url to your self-hosted deployment's origin. Versions before 0.2.1 defaulted to https://sim.ai; set the www origin explicitly when using an older SDK against Sim Cloud to avoid redirects.

Run a workflow

Use a context manager to close the HTTP session automatically:

import os
from simstudio import SimStudioClient, SimStudioError

with SimStudioClient(
    api_key=os.environ["SIM_API_KEY"],
    base_url=os.getenv("SIM_BASE_URL", "https://www.sim.ai"),
) as client:
    try:
        result = client.execute_workflow_sync(
            os.environ["WORKFLOW_ID"],
            input={"message": "Process this data"},
            timeout=30.0,
        )
        if result.status == "completed":
            print(result.output)
            print("Duration in ms:", result.total_duration)
        elif result.status == "paused":
            print("The workflow is waiting to resume.")
        else:
            print(result.status, result.error)
    except SimStudioError as error:
        print(error.code, error.status, str(error))
        raise

Replace message with an input field defined by your deployed trigger. The SDK runs the active deployment; it does not select manual draft execution.

A failed synchronous workflow returns normally with status="failed", success=False, and error set. A cancelled run also returns success=False, often without an error. success=True means completed or paused; a pause is unfinished work. HTTP, authentication, and client timeout errors raise SimStudioError. This differs from the TypeScript SDK, which throws for a failed synchronous run.

Start an asynchronous run

Queue a deployed run and poll its run_id. The initial success=True confirms acceptance, not completion:

import os
import time
from simstudio import AsyncExecutionResult, SimStudioClient

waiting = {"queued", "pending", "running", "redacting"}
workflow_id = os.environ["WORKFLOW_ID"]

with SimStudioClient(api_key=os.environ["SIM_API_KEY"]) as client:
    result = client.execute_workflow(
        workflow_id,
        input={"message": "Process this dataset"},
        async_execution=True,
        execution_timeout_seconds=3600,
    )
    if not isinstance(result, AsyncExecutionResult):
        raise RuntimeError("Expected an asynchronous run receipt")

    deadline = time.monotonic() + 300
    while True:
        run = client.get_workflow_run(workflow_id, result.run_id, include_output=True)
        if run["status"] not in waiting:
            break
        if time.monotonic() >= deadline:
            raise TimeoutError(f"Stop waiting locally; run {result.run_id} may still be active")
        time.sleep(2)

    if run["status"] == "completed":
        print(run["output"])
    elif run["status"] == "paused":
        print("Resume condition:", run["paused"])
    else:
        print(run["status"], run["error"])

The five-minute polling deadline is an application choice; it does not cancel the run. A time pause can resume automatically, while a human pause requires input. Resume can create another run attempt. Use Get workflow run and the resume endpoint for the current state and resume behavior.

Async submission cannot include stream=True, selected_outputs, or other output-shaping options. execution_timeout_seconds accepts an integer from 1 to 604800, requires async execution, and only shortens the account policy. The HTTP timeout remains separate.

Handle errors and rate limits

SimStudioError(message, code=None, status=None) exposes code and the HTTP status when available. A workflow failure returned in WorkflowExecutionResult must be handled separately from exceptions.

Other server codes can pass through. V2 uses BAD_REQUEST for invalid JSON/request input, rather than the legacy INVALID_JSON branch. See API errors.

execute_with_retry retries only rate-limit exceptions. Its defaults allow three retries after the initial attempt, with exponential delay and ±25% jitter. A supplied Retry-After takes precedence over the calculated delay; jitter applies to that wait too.

import os
from simstudio import SimStudioClient

with SimStudioClient(api_key=os.environ["SIM_API_KEY"]) as client:
    result = client.execute_with_retry(
        os.environ["WORKFLOW_ID"],
        input={"message": "Process this"},
        timeout=30.0,
        max_retries=3,
        initial_delay=1.0,
        max_delay=30.0,
        backoff_multiplier=2.0,
    )
    print(result)

initial_delay and max_delay are seconds. RateLimitInfo.retry_after is milliseconds. max_delay caps the calculated backoff, not a server-supplied wait.

Method reference

Constructor

SimStudioClient(api_key: str, base_url: str = "https://www.sim.ai")

api_key is required. base_url is an origin, without an /api suffix; trailing slashes are removed. The client sends the key in X-API-Key.

Execute

execute_workflow(workflow_id: str, input=None, *, timeout=30.0,
    stream=None, selected_outputs=None, async_execution=None,
    execution_timeout_seconds=None) -> WorkflowExecutionResult | AsyncExecutionResult

execute_workflow_sync(workflow_id: str, input=None, *, timeout=30.0,
    stream=None, selected_outputs=None) -> WorkflowExecutionResult

execute_with_retry(workflow_id: str, input=None, *, timeout=30.0,
    stream=None, selected_outputs=None, async_execution=None,
    execution_timeout_seconds=None, max_retries=3, initial_delay=1.0,
    max_delay=30.0, backoff_multiplier=2.0)
    -> WorkflowExecutionResult | AsyncExecutionResult

All options after * are keyword-only. execute_workflow_sync forces async_execution=False; the retry method accepts the same execution options as execute_workflow.

File-like objects nested in input are converted to base64. The current synchronous adapter does not expose v2 blockOutputs, even when selected_outputs is sent. Use a run read or raw HTTP when you need that map.

Read workflow deployment and runs

get_workflow_status(workflow_id: str) -> WorkflowStatus
validate_workflow(workflow_id: str) -> bool
get_workflow_run(workflow_id: str, run_id: str, *,
    include_output=None, selected_outputs=None) -> Dict[str, Any]
get_job_status(job_id: str) -> Dict[str, Any]

get_workflow_status reads the legacy /api/workflows/{id}/status route. validate_workflow returns only its is_deployed flag; it returns False if the status request raises SimStudioError. It does not validate credentials, input, quota, or deployment readiness. Use the status method directly to distinguish a failed lookup from an undeployed workflow:

status = client.get_workflow_status("workflow-id")
print(status.is_deployed, status.needs_redeployment)

get_workflow_run reads and unwraps the v2 run resource. include_output=True includes final output for a completed run; default omitted/false. Run-read selected_outputs uses block IDs, as blockId or blockId.path, rather than block names. Selected block outputs are independent of include_output.

get_job_status is for IDs created by legacy 0.1.x async execution at /api/jobs/{jobId}. Use get_workflow_run for runs started by 0.2.x.

Read limits and update the client

get_rate_limit_info() -> RateLimitInfo | None
get_usage_limits() -> UsageLimits
set_api_key(api_key: str) -> None
set_base_url(base_url: str) -> None
close() -> None

get_rate_limit_info reads cached headers from the most recent response that supplied rate information. It does not make a request. get_usage_limits calls the legacy account helper; see its shape below. Setters change subsequent requests. close closes the HTTP session; a context manager calls it automatically.

Return objects

WorkflowExecutionResult

Nested oversized values may be LargeValueRef objects with __simLargeValueRef, version, id, kind, size, optional key, executionId, and preview. Their keys are opaque server pointers, not download URLs. See large outputs.

AsyncExecutionResult

WorkflowStatus

Run dictionary

get_workflow_run returns the server's data dictionary without changing camelCase keys:

Each file includes id, name, byte size, MIME type, downloadPath, and nullable base64. The SDK does not expose file-inlining options; download at the returned path with an authenticated raw HTTP request. Permission policies can withhold cost or recorded output fields. See Get workflow run for the full contract, functional-output permission conflict, and file limits.

RateLimitInfo

Fields are limit: int, remaining: int, reset: int, and retry_after: Optional[int] = None. V2 ISO reset headers become epoch milliseconds. Integer reset headers from older endpoints retain their original epoch-seconds value. For a v2 response:

from datetime import datetime, timezone

rate = client.get_rate_limit_info()
if rate:
    print(rate.remaining)
    print(datetime.fromtimestamp(rate.reset / 1000, tz=timezone.utc))

Use a known response surface when converting reset; use resetAt ISO strings from the usage helper when you need a format that is unambiguous across endpoints.

UsageLimits

The dataclass contains success: bool, rate_limit: Dict[str, Any], and usage: Dict[str, Any]:

rate_limit:
  sync / async:
    isLimited: bool
    requestsPerMinute: number
    maxBurst: number
    remaining: number
    resetAt: ISO timestamp
  authType: "api" or "manual"
usage:
  currentPeriodCost: number (USD)
  limit: number (USD)
  plan: str

requestsPerMinute is the refill rate; maxBurst is the bucket capacity. There is no rate_limit["sync"]["limit"] field. The Python dataclass does not expose the legacy route's storage result.

limits = client.get_usage_limits()
print("Sync remaining:", limits.rate_limit["sync"]["remaining"])
print("Resets at:", limits.rate_limit["sync"]["resetAt"])
print("Period cost in USD:", limits.usage["currentPeriodCost"])

This helper resolves account/key-owner usage and can differ from the payer for a selected workspace. For workspace credit and storage state, use Get billing status. Logs APIs also report USD; v2 Billing and run-resource amounts use credits.

File inputs

Include a binary file-like object under the input field your deployed trigger expects:

import os
from simstudio import SimStudioClient

with SimStudioClient(api_key=os.environ["SIM_API_KEY"]) as client:
    with open("document.pdf", "rb") as document:
        result = client.execute_workflow_sync(
            os.environ["WORKFLOW_ID"],
            input={"documents": [document], "query": "Summarize this document"},
        )
    print(result.output)

The SDK reads and base64-encodes the bytes, retaining the file name and using its content_type attribute when available. Otherwise MIME type defaults to application/octet-stream. Validate upload size and content on your backend; the execute request has a 10 MiB JSON-body limit, including base64 overhead.

Execute several workflows

The client has no batch-execution method. A sequential loop makes one request per workflow and handles each result independently:

for workflow_id, workflow_input in [
    ("workflow-1", {"message": "Analyze this"}),
    ("workflow-2", {"message": "Summarize this"}),
]:
    try:
        result = client.execute_workflow_sync(workflow_id, workflow_input)
        print(workflow_id, result.status, result.output, result.error)
    except SimStudioError as error:
        print(workflow_id, error.code, str(error))

Use this inside the configured client context. A deployment-status precheck is optional and cannot guarantee the later run will succeed.

Stream through HTTP

The current 0.2.x SDK sends stream=True but then parses the response as JSON. It does not expose an SSE iterator; a valid SSE response fails that parser. Use raw HTTP for streaming:

import os
import requests

workflow_id = os.environ["WORKFLOW_ID"]
with requests.post(
    f"https://www.sim.ai/api/v2/workflows/{workflow_id}/execute",
    headers={
        "X-API-Key": os.environ["SIM_API_KEY"],
        "X-Sim-Stream-Protocol": "agent-events-v1",
    },
    json={"input": {"message": "Count to five"}, "stream": True,
          "selectedOutputs": ["agent.content"]},
    stream=True,
    timeout=(10, 300),
) as response:
    response.raise_for_status()
    for line in response.iter_lines():
        if line:
            print(line.decode("utf-8"))

This prints raw SSE. Applications must parse complete events: success ends with {event:"final",data:...}, failure ends with {event:"error",...}, then [DONE]. Preserve structured output and reset events; do not filter only for chunk or the obsolete done event. The protocol header enables live answer tokens and requires handling chunk_reset; reasoning/tool events require their separate flags.

For Flask or another web backend, authenticate the caller and authorize the workflow before opening the upstream request. Forward all SSE bytes with text/event-stream, check upstream HTTP status before starting the downstream response, and close the upstream response if the caller disconnects. See API streaming and the reference client for framing and event handling.

Upgrade from 0.1.x

0.2.0 changed execution requests to /api/v2, nests named input under input, and sends async/timeout options in the body instead of X-Execution-Mode and X-Execution-Timeout-Seconds headers.

Replace AsyncExecutionResult.job_id with run_id; its old execution_id field is removed. Poll with get_workflow_run(workflow_id, run_id), not get_job_status. Synchronous metadata is now SDK-built: use metadata["runId"] instead of the former metadata["executionId"]. Read status to distinguish failed and cancelled returns. Logs and trace spans are no longer populated by this adapter.

Workflow status, legacy jobs, and account usage helpers still use their original routes. This SDK does not expose every v2 operation, manual execution, resume/cancel, caller-selected run headers, or NDJSON/agent-event streaming controls.

Related docs

• API authentication
• TypeScript SDK
• API getting started
• Get workflow run
• Python package source and Apache-2.0 license