shiftlabs

Type hints and a testing kit for shiftlabs workflows written in Python, and shiftlabs.client for Python apps that call the shiftlabs API.

uv add --dev shiftlabs   # or: pip install shiftlabs

from shiftlabs import Context, workflow


async def charge(invoice_id: str) -> dict[str, str]:
    ...


@workflow(payload={"type": "object", "properties": {"id": {"type": "string"}}, "required": ["id"]})
async def refund_check(ctx: Context) -> dict[str, str]:
    charged = await ctx.step.run("charge", charge, ctx.payload["id"], retries={"limit": 3, "delay": "10 seconds"})
    ctx.log.info("charged", {"id": charged["id"]})
    return charged

For workflows, the package is for your editor, type checker and tests. You don't need it at run time: the platform adds the same shiftlabs module to every Python workflow when you deploy, at the version the platform runs, and never adds shiftlabs.testing or shiftlabs.client. An app that calls the API with shiftlabs.client installs it as a dependency.

Run npx shiftlabs types to write shiftlabs_types.pyi with your workspace's secret, var, agent and workflow names. After that, ctx.secrets.NAME and ctx.vars.NAME only accept those names, and so do the client's agent and workflow arguments.

Testing

shiftlabs.testing.run_workflow runs a workflow inside a test, the way the platform runs it, with no deploy. Steps run at once: retries don't wait, FatalError fails a step at its first attempt, and sleeps return at once. Waits are answered by step name from events, approvals, asks, calls and starts, next to payload, secrets, vars and state. A missing event returns None, a missing approval is denied, and a missing ask or call fails its step. The payload is checked against the workflow's schema, as a webhook's is. ctx.secrets.NAME reads a stand-in, as on the platform: give a secret as {"value": …, "readable": True} to reveal it, or {"value": …, "hmac": True} to sign with it.

from unittest.mock import AsyncMock, patch

from shiftlabs.testing import run_workflow

import refund_check


def test_refund_check_charges_once() -> None:
    with patch.object(refund_check, "charge", AsyncMock(return_value={"id": "ch_123"})) as charge:
        result = run_workflow(refund_check.refund_check, payload={"id": "ch_123"})
    assert result.status == "completed", result.error
    assert result.output == {"id": "ch_123"}
    charge.assert_awaited_once_with("ch_123")

It returns a RunResult with status, output, error, steps, logs, state and started. Then it runs the workflow a second time on the stored step results, as a run that resumes after a sleep does, and raises AssertionError("code outside a step changed on replay: …") if that calls other steps, such as a step named from time.time(). replay=False skips that check. It mocks no network: use unittest.mock or your HTTP client's mocks. run_workflow runs the workflow in an event loop of its own, so sync and async tests can call it.

Client

shiftlabs.client calls the shiftlabs API from a Python app: chat with agents as your users and stream their answers, ask for JSON answers, run workflows, read runs, send events, answer approvals and mint chat tokens. Shiftlabs waits for each call; AsyncShiftlabs has the same members for asyncio. Both use only the standard library.

import os

from shiftlabs.client import Shiftlabs

sl = Shiftlabs(api_key=os.environ["SHIFTLABS_API_KEY"])
for event in sl.agents.chat("billing", person="user_ann").send("Where is my refund?"):
    if event["type"] == "text":
        print(event["delta"], end="")
started = sl.workflows.run("stripe-refund", {"invoiceId": "in_123"})
print(sl.runs.get(started.run_id)["run"]["status"])

Failures raise shiftlabs.client.ShiftlabsError, whose code, hint and to_dict() match the CLI's --json errors. See the client docs.

Docs: docs.shiftlabs.org.