Metadata-Version: 2.4
Name: shiftlabs
Version: 0.3.0
Summary: Type hints for shiftlabs Python workflows, the module the platform injects at deploy, shiftlabs.testing to run them in tests, and shiftlabs.client for apps that call the shiftlabs API.
Author: shiftlabs
Author-email: shiftlabs <packages@shiftlabs.org>
License-Expression: MIT
License-File: LICENSE
Classifier: Programming Language :: Python :: 3
Classifier: Typing :: Typed
Requires-Python: >=3.11
Project-URL: Homepage, https://shiftlabs.org
Project-URL: Documentation, https://docs.shiftlabs.org
Description-Content-Type: text/markdown

# shiftlabs

Type hints and a testing kit for [shiftlabs](https://shiftlabs.org) workflows written in Python, and `shiftlabs.client`
for Python apps that call the shiftlabs API.

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

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

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

```python
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](https://docs.shiftlabs.org/client#python).

Docs: [docs.shiftlabs.org](https://docs.shiftlabs.org).
