# Python (/api-reference/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 [#installation]

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

```bash
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](/api-reference/authentication) for key types and permissions.

### Server compatibility [#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 [#quick-start]

Use a context manager to close the HTTP session automatically:

```python
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 [#async-workflow-execution]

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

```python
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](/api-reference/workflow-runs/getWorkflowRunV2) and the [resume endpoint](/api-reference/workflow-runs/resumeWorkflowRunV2) 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 [#error-handling]

`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.

| Code | Meaning |
| - | - |
| `UNAUTHORIZED` | Missing, invalid, or expired credential |
| `BAD_REQUEST` | V2 request validation failed |
| `TIMEOUT` | The client stopped waiting |
| `RATE_LIMIT_EXCEEDED` | SDK wrapper for an HTTP `429` |
| `USAGE_LIMIT_EXCEEDED` | Usage policy rejected the request or run |
| `INVALID_EXECUTION_TIMEOUT` | An unsupported or invalid async timeout option |
| `EXECUTION_ERROR` | Transport, parsing, or execution-request failure |
| `STATUS_ERROR`, `USAGE_ERROR` | A status or usage helper failed |

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](/api-reference/getting-started#error-handling).

`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.

```python
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.

<Callout type="warn">
  A client timeout or lost response does not prove the workflow never started. Avoid blindly repeating an execution that may have effects. For caller-chosen run IDs and reconciliation, use the raw v2 API described in [Getting started](/api-reference/getting-started#sync-vs-async-execution).
</Callout>

## Method reference [#api-reference]

### Constructor [#constructor]

```text
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]

```text
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`.

| Parameter | Behavior and default |
| - | - |
| `workflow_id` | Deployed workflow ID |
| `input` | Optional value; a dictionary becomes the v2 `input` object, while a primitive or list becomes `{"input": value}` inside it; omitted/`None` becomes `{}` |
| `timeout` | Client HTTP timeout in seconds; default `30.0` |
| `async_execution` | Default omitted, so server execution is synchronous; `True` returns an async receipt |
| `execution_timeout_seconds` | Async-only integer cap, `1`–`604800`; cannot extend account policy |
| `selected_outputs` | Optional list of normalized block-name/path selectors such as `agent.content`; incompatible with async submission |
| `stream` | Defaults omitted/false; the current SDK parser cannot consume the SSE response when `True`; use raw HTTP below |
| `max_retries` | Retry count after the initial attempt; default `3` |
| `initial_delay`, `max_delay` | Calculated retry delay in seconds; defaults `1.0`, `30.0` |
| `backoff_multiplier` | Delay multiplier; default `2.0` |

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 [#read-workflow-deployment-and-runs]

```text
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:

```python
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 [#read-limits-and-update-the-client]

```text
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 [#data-classes]

### WorkflowExecutionResult [#workflowexecutionresult]

| Field | Python type/default | Current behavior |
| - | - | - |
| `success` | `bool`, required | `True` only for completed or paused |
| `status` | `Optional[str] = None` | Server status: completed, failed, paused, or cancelled |
| `output` | `Optional[Any] = None` | Workflow output, including any returned partial result |
| `error` | `Optional[str] = None` | Failure message, when present |
| `metadata` | `Optional[Dict[str, Any]] = None` | SDK-built `duration`, `runId`, `startTime`, `endTime`; duration is milliseconds |
| `total_duration` | `Optional[float] = None` | Duration in milliseconds |
| `logs`, `trace_spans` | `Optional[list] = None` | Compatibility fields; not populated by the v2 adapter |

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](/workflows/deployment/api#oversized-outputs).

### AsyncExecutionResult [#asyncexecutionresult]

| Field | Python type/default |
| - | - |
| `success` | `bool`, required; current accepted receipts set `True` |
| `run_id` | `str`, required |
| `status_url` | `str`, required |
| `message` | `str = ""`; current adapter sets `"Workflow execution queued"` |
| `async_execution` | `bool = True` |

### WorkflowStatus [#workflowstatus]

| Field | Python type/default |
| - | - |
| `is_deployed` | `bool`, required |
| `deployed_at` | `Optional[str] = None` |
| `needs_redeployment` | `bool = False` |

### Run dictionary [#run-dictionary]

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

| Field | Shape |
| - | - |
| `runId`, `workflowId` | String identifiers |
| `status` | `queued`, `pending`, `running`, `redacting`, `paused`, `completed`, `failed`, or `cancelled` |
| `trigger` | String or `None` |
| `startedAt`, `endedAt` | ISO timestamps or `None`; queued runs report enqueue time as startedAt |
| `durationMs` | Number or `None` |
| `paused` | Pause details or `None`; use `pauseKind` and `resumeAt` to identify the resume condition |
| `cost` | `{"total": number}` in credits, or `None` when unavailable/withheld |
| `error` | Structured `code`/`message` failure, or `None` |
| `output` | Final output when requested for a completed run, otherwise `None` |
| `blockOutputs` | Selected block-ID/path outputs, or `None` |
| `files` | Produced-file metadata when output is requested/available, otherwise `None` |

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](/api-reference/workflow-runs/getWorkflowRunV2) for the full contract, functional-output permission conflict, and file limits.

### RateLimitInfo [#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:

```python
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 [#usagelimits]

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

```text
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.

```python
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](/api-reference/billing/getBillingStatus). Logs APIs also report USD; v2 Billing and run-resource amounts use credits.

## File inputs [#file-inputs]

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

```python
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 [#batch-workflow-execution]

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

```python
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 [#streaming-workflow-execution]

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:

```python
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](/workflows/deployment/api#streaming) and the [reference client](/workflows/deployment/agent-events#reference-client) for framing and event handling.

## Upgrade from 0.1.x [#upgrade-from-01x]

`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 [#related-docs]

* [API authentication](/api-reference/authentication)
* [TypeScript SDK](/api-reference/typescript)
* [API getting started](/api-reference/getting-started)
* [Get workflow run](/api-reference/workflow-runs/getWorkflowRunV2)
* [Python package source and Apache-2.0 license](https://github.com/simstudioai/sim/tree/main/packages/python-sdk)