Python SDK reference

The shiftlabs module is what a Python workflow imports. Every deploy includes it, and the shiftlabs package on PyPI gives your editor its type hints and adds the testing kit. This page is generated from its Python source.

Duration

Milliseconds, or a string such as "30 seconds" with units from second to year, singular or plural.

pythontype alias

Duration: TypeAlias = int | float | str

FatalError

Raised in a step, fails the step and the run at once, with no more attempts. Raised outside a step, it fails the run as any error does.

pythonclass

class FatalError(Exception): ...

Retries

step.run(..., retries=...) and step.exec(..., retries=...): how many times to retry a failing step (a FatalError in step.run never is), the duration between attempts, and the backoff (exponential by default).

pythonclass

class Retries(TypedDict):
    limit: int
    delay: Duration
    backoff: NotRequired[Literal["constant", "linear", "exponential"]]

PayloadSchema

@workflow(payload=...): a pydantic model or a JSON Schema dict.

pythontype alias

PayloadSchema: TypeAlias = type[_PayloadModel] | dict[str, Any]

Concurrency

@workflow(concurrency=...): at most limit runs of the workflow at once, from 1 to the workspace's concurrent runs. With key, a dotted path into the payload such as "customer.id", runs whose payloads hold the same value there share the limit, and a run without the field uses the key "". Runs past the limit wait in arrival order with the status queued, and start as runs with the same key end.

pythonclass

class Concurrency(TypedDict):
    limit: int
    key: NotRequired[str]

WorkflowDefinition

What @workflow returns. A workflow module defines exactly one; shiftlabs.testing.run_workflow takes it.

pythonclass

class WorkflowDefinition:
    fn: Callable[[Context], Awaitable[Any]]
    payload: PayloadSchema | None
    concurrency: Concurrency | None

workflow

Marks the module's entry: @workflow async def refund_check(ctx): …. @workflow(payload=Invoice) takes the payload's schema, a pydantic model or a JSON Schema dict: webhooks and API runs whose payload fails it get 400. @workflow(concurrency={"limit": 1, "key": "customer_id"}) caps the runs at once; runs past it wait.

pythonfunction

@overload
def workflow(
    fn: Callable[[Context], Awaitable[Any]],
    /,
) -> WorkflowDefinition: ...
@overload
def workflow(
    *,
    payload: PayloadSchema | None = None,
    concurrency: Concurrency | None = None,
) -> Callable[[Callable[[Context], Awaitable[Any]]], WorkflowDefinition]: ...

ApprovalDecision

What step.wait_for_approval returns. by: the approver's email or API key id, None when nobody answered in time; note: what the approver wrote.

pythonclass

class ApprovalDecision(TypedDict):
    approved: bool
    by: str | None
    note: str | None

SecretRef

{"secret": NAME} in step.exec's env or files: a secret marked readable, whose value the runner gets for that command only.

pythonclass

class SecretRef(TypedDict):
    secret: str

ExecResult

What step.exec returns, and ExecError's result: the exit code (124 when the command timed out), the last 256 KiB of stdout and stderr with secret values replaced by "[redacted]", each stream's whole size in bytes, truncated when stdout or stderr holds only the end of its stream, timed_out, how long it ran in milliseconds, the name of the runner that ran it, and which try this was, 1 for the first.

pythonclass

class ExecResult(TypedDict):
    exit_code: int
    stdout: str
    stderr: str
    stdout_bytes: int
    stderr_bytes: int
    truncated: bool
    timed_out: bool
    duration_ms: int
    runner: str
    attempt: int

ExecError

Raised by step.exec when the command exits with a code outside ok_exit_codes, or times out. result holds its exit code and output.

pythonclass

class ExecError(Exception):
    result: ExecResult

Step

Code outside step.run() runs again every time the run resumes (after a sleep or a retry); only step results are durable, so side effects belong inside steps.

async def run(name: str, fn: Callable[[*Ts], Awaitable[T]], *args: *Ts, retries: Retries | None = None, timeout: Duration | None = None) -> T async def run(name: str, fn: Callable[[*Ts], T], *args: *Ts, retries: Retries | None = None, timeout: Duration | None = None) -> T
Runs fn(*args) once and stores its JSON-serializable result; a resumed run gets the stored result.
async def sleep(name: str, duration: Duration) -> None
Durable sleep. The run resumes after duration, on the same version.
async def sleep_until(name: str, until: datetime.datetime | int | float) -> None
Durable sleep until a datetime or epoch seconds.
async def ask(name: str, *, agent: str, message: str, schema: None = None) -> str async def ask(name: str, *, agent: str, message: str, schema: dict[str, Any]) -> Any
Asks agent and returns its final answer: the text, or with schema (a JSON Schema dict) the JSON that matches it. A run has one conversation per step, which the workflow owns; a resumed run gets the stored answer. An answer that doesn't match gets one more turn; a second miss fails the step, with no retries.
async def start(name: str, *, workflow: str, payload: object = None) -> str
Starts a run of another workflow in this workspace and returns its run id, without waiting. The id comes from this run and name, so a resumed run never starts a second one. It counts toward your limits like any run.
async def call(name: str, *, workflow: str, payload: object = None) -> Any
Starts a run like start(), waits for it to end and returns its result; a failed run fails this step with its error. The wait counts toward the run's lifetime, not its steps, and the run sleeps while it waits.
async def exec(name: str, *, run: str | None = None, command: list[str] | None = None, env: Mapping[str, str | SecretRef] | None = None, files: Mapping[str, str | SecretRef] | None = None, stdin: str | None = None, timeout: Duration | None = None, runner: str | None = None, ok_exit_codes: list[int] | Literal["any"] | None = None, retries: Retries | None = None) -> ExecResult
Runs a command on a runner in your own network and returns its result: run, a bash script, or command, a program and its arguments. env and files (by relative path) take text, or {"secret": NAME} for a secret marked readable, and stdin goes to the command's standard input. timeout defaults to "10 minutes", runner to the pool "default" and ok_exit_codes to [0] ("any" accepts every exit code); retries tries again after a failure. Another exit code raises ExecError with the result, and a command no runner ran or finished raises an error that says why. A resumed run gets the stored result.
async def wait_for_event(name: str, *, timeout: Duration | None = None) -> Any
Waits for the event name (sent to event_url(name) or POST /v1/runs/:id/events/:name) and returns its data, the parsed JSON body or the text, or None once timeout (default "24 hours") passes. An event sent before the run waits is kept. Waiting counts against the run's lifetime, not its steps.
def event_url(name: str) -> str
A URL that sends this run the event name when POSTed to, with no API key: it reaches this run and event only.
def idempotency_key(name: str) -> str
sl_ and 32 hex characters from this run's id and name: the same on every retry, replay and resume of the run, different in every other run. Pass it as Idempotency-Key to Stripe and similar APIs. Not a step.
async def wait_for_approval(name: str, *, message: str | None = None, timeout: Duration | None = None, approvers: list[str] | None = None, approvals: int | None = None) -> ApprovalDecision
Puts a request in the workspace's Approvals inbox, notifies its approvers, and waits for one to approve or deny. Unanswered by timeout (default "72 hours") it counts as denied. Waiting counts against the run's lifetime, not its steps. Names can't end like refund #2. approvers: up to 20 of "role:admin", "role:manager" or member emails; only they see and answer it. Default: every admin. approvals: how many approvers must approve, 1 to 5 (default 1). The workspace's setting in Settings → Approvals is the least a request needs; asking for fewer has no effect. Under 2 or more, each approval comes from a different member; an API key can't approve.

Secrets

ctx.secrets.NAME: a stand-in for each secret with hosts, which the gateway swaps for the value on requests to those hosts. An unknown name raises KeyError.

pythonclass

class Secrets: ...

Vars

ctx.vars.NAME: the workspace value, overridden per workflow. A name that is not set reads as None.

pythonclass

class Vars: ...

Log

ctx.log.info(message, fields), and debug, warn, error: one log line each, fields as JSON after the message.

def debug(message: object, fields: Mapping[str, object] | None = None) -> None
A line at level debug.
def info(message: object, fields: Mapping[str, object] | None = None) -> None
A line at level log.
def warn(message: object, fields: Mapping[str, object] | None = None) -> None
A line at level warn.
def error(message: object, fields: Mapping[str, object] | None = None) -> None
A line at level error.

State

ctx.state: JSON values kept per workflow, shared by all its runs and versions; the last write wins. Every method works only inside step.run() and raises outside one. A key is 1 to 512 bytes. Writes past the plan's limits (keys per workflow, value size, total) raise.

async def get(key: str) -> Any
The value stored at key, or None if there is none.
async def set(key: str, value: Any) -> None
Stores value as JSON at key, replacing what was there.
async def delete(key: str) -> None
Removes key. Removing a key that isn't there does nothing.
async def list(prefix: str | None = None, after: str | None = None, limit: int | None = None) -> list[str]
The workflow's keys in order: those starting with prefix, after after, at most limit (1 to 1,000).

Context

The run context, the one argument of the @workflow function.

payload: Any
The webhook body: parsed JSON when sent as application/json, the raw text otherwise.
run_id: str
This run's id.
step: Step
Durable steps and sleeps.
vars: Vars
vars.NAME: the workspace value, overridden per workflow.
secrets: Secrets
secrets.NAME: a stand-in for each secret with hosts.
reveal: Callable[[str], str]
reveal(name): the value of a secret marked readable. Every load is audited and the value is scrubbed from logs.
sign: Callable[[str, str, Literal["hex", "base64"]], Awaitable[str]]
await sign(name, data, encoding="hex"): HMAC-SHA256 of data with a secret marked hmac. Returns only the signature.
log: Log
log.info(message, fields), and debug, warn, error: one log line each, shown with the run.
state: State
await state.get(key), set, delete, list: JSON values kept per workflow across runs, only inside step.run().

Testing kit

shiftlabs.testing runs a workflow inside your test, the way the platform runs it, with no deploy. Only the PyPI package has it: deploys never include it. See Test a workflow.

StepType

The step method a RunStep was.

pythontype alias

StepType: TypeAlias = Literal[
    "run",
    "sleep",
    "sleep_until",
    "wait_for_event",
    "wait_for_approval",
    "ask",
    "start",
    "call",
    "exec",
]

ApprovalAnswer

An answer in approvals given as a dict. by defaults to "test" and note to None.

pythonclass

class ApprovalAnswer(TypedDict):
    approved: bool
    by: NotRequired[str | None]
    note: NotRequired[str | None]

SecretValue

A secret in secrets with the flags PUT /v1/secrets/:NAME takes. Both are False unless given, as there: readable lets reveal(NAME) return the value, and hmac lets sign(NAME, …) sign with it.

pythonclass

class SecretValue(TypedDict):
    value: str
    readable: NotRequired[bool]
    hmac: NotRequired[bool]

ExecAnswer

A command's outcome in execs. stdout and stderr default to "". A command the runner stopped at its timeout exits with 124, so with timed_out True exit_code is 124 or left out.

pythonclass

class ExecAnswer(TypedDict):
    exit_code: NotRequired[int]
    stdout: NotRequired[str]
    stderr: NotRequired[str]
    timed_out: NotRequired[bool]

ExecRequest

What a function in execs gets: the step's name and its options as the platform checked them, with their defaults, None for run or command, and timeout and the retries' delay in milliseconds.

pythonclass

class ExecRequest(TypedDict):
    name: str
    run: str | None
    command: list[str] | None
    env: dict[str, str | shiftlabs.SecretRef]
    files: dict[str, str | shiftlabs.SecretRef]
    stdin: str | None
    timeout: int
    runner: str
    ok_exit_codes: list[int] | Literal["any"]
    retries: shiftlabs.Retries | None

RunStep

One step the run took. attempts: the attempts step.run or step.exec made, 1 when the first succeeded, and 1 for every other step. output: what it returned: a result, the event's data, the decision, the answer, a run id. error: why it failed, if it did.

pythonclass

@dataclasses.dataclass(frozen=True)
class RunStep:
    type: StepType
    name: str
    attempts: int = 1
    output: Any = None
    error: str | None = None

LogLine

A log line, as the run's logs show it: info writes level "log".

pythonclass

@dataclasses.dataclass(frozen=True)
class LogLine:
    level: Literal["debug", "log", "warn", "error"]
    message: str

StartedRun

A run that step.start or step.call started. name: the step's name.

pythonclass

@dataclasses.dataclass(frozen=True)
class StartedRun:
    name: str
    workflow: str
    payload: Any
    run_id: str

RunResult

What run_workflow returns: output when the workflow returned, error when it raised. steps: every step, in the order the workflow called them. state: the workflow's state when the run ended.

pythonclass

@dataclasses.dataclass(frozen=True)
class RunResult:
    status: Literal["completed", "failed"]
    output: Any
    error: Exception | None
    steps: list[RunStep]
    logs: list[LogLine]
    state: dict[str, Any]
    started: list[StartedRun]

run_workflow

Runs the @workflow function in this process as the platform runs it, in an event loop of its own, and returns what it did. Steps run inline: retries happen at once, FatalError fails a step with no more attempts, and sleeps return at once. Waits are answered by step name: events gives a wait's data (none returns None, as a timeout does), approvals True, False or {"approved", "by", "note"} (none is denied), asks the agent's answer and calls the other run's result (none fails the step), starts a run id. An exception in asks, calls or starts fails that step with its message. execs gives each step.exec's outcome, {"exit_code", "stdout", "stderr", "timed_out"}, or a function of the ExecRequest and the attempt that returns one; an exception fails the attempt as a command no runner ran does, retries apply, and a step with none fails. secrets gives each secret's value, or {"value", "readable", "hmac"}: ctx.secrets.NAME reads a stand-in shaped like the platform's, reveal needs a readable secret and sign an hmac one. payload ({} when None) must pass the workflow's payload schema, or this raises ValueError. Then the workflow runs again on its stored step results, as a resumed run does, and this raises AssertionError "code outside a step changed on replay" if it calls other steps; replay=False skips that.

pythonfunction

def run_workflow(
    definition: shiftlabs.WorkflowDefinition,
    *,
    payload: object = None,
    secrets: Mapping[str, str | SecretValue] | None = None,
    vars: Mapping[str, str] | None = None,
    state: Mapping[str, object] | None = None,
    events: Mapping[str, object] | None = None,
    approvals: Mapping[str, bool | ApprovalAnswer] | None = None,
    asks: Mapping[str, object] | None = None,
    calls: Mapping[str, object] | None = None,
    starts: Mapping[str, str | Exception] | None = None,
    execs: Mapping[
        str,
        ExecAnswer | Exception | Callable[
            [ExecRequest, int],
            ExecAnswer | Exception | Awaitable[ExecAnswer | Exception],
        ],
    ] | None = None,
    replay: bool = True,
) -> RunResult: ...