Skip to content

Latest commit

 

History

History
89 lines (69 loc) · 3.5 KB

File metadata and controls

89 lines (69 loc) · 3.5 KB

Boardwalk Python SDK

The Python SDK for Boardwalk agent workflows. A workflow is a plain async function; capabilities are imports:

from boardwalk import agent, phase, secrets

async def run(input, context):
    key = await secrets.get("STRIPE_API_KEY")
    phase("analyze")
    note = await agent(f"Why did payment {input['id']} fail?")
    return {"action": "retry", "note": note}

The entry is a module-level run (async or sync), found by name — positional params, Lambda-style: input is param 0, context is param 1, and declaring fewer params is fine. The return value is the run's output. Typed I/O is opt-in: annotate input with a Pydantic model / dataclass / TypedDict and the return type, and the platform derives the contract — rich types (datetime, bytes, set, Decimal, big int) cross the wire in canonical JSON encodings and arrive revived.

Status: pre-alpha. Built as part of the workflow format redesign; the platform side is still landing.

The surface

from boardwalk import (
    agent,        # run an LLM leaf: await agent(prompt, model=..., schema=..., tools=[...])
    workflows,    # workflows.call / workflows.run / workflows.schedule
    sleep,        # await sleep(1500) / sleep(until=...)
    human_input,  # pause for a person; text / choice / multiselect
    secrets,      # await secrets.get("NAME")
    artifacts,    # await artifacts.write(name, content_type, body)
    computer,     # computer.open_browser() -> BrowserSession / open_desktop() -> DesktopSession
    shell,        # await shell(cmd) -> ShellResult (never raises on non-zero exit)
    parallel,     # run thunks concurrently; failures isolate to None
    phase,        # phase("analyze") — a run-timeline marker
    auth,         # await auth.id_token(audience) / auth.api_token()
    usage,        # await usage.get() — live budget state
    Context,      # run(input, context)'s second parameter (frozen dataclass)
)

Unit testing

install_test_host makes run(input, context) a plain function call over stubs — no socket, no engine:

from boardwalk import install_test_host

host = install_test_host(
    agent=lambda prompt, opts: "LGTM",
    secrets={"STRIPE_API_KEY": "sk_test_1"},
)
out = await run({"id": "pay_1"}, host.context())

How it works

Every capability is a thin client of the Boardwalk host protocol — newline-delimited JSON-RPC 2.0 over a local Unix socket (BOARDWALK_HOST_SOCK) served by the runner inside the run's machine. The SDK holds no model credentials and no platform secrets; the runner brokers everything. Inline agent() tools run in your program process — only their declarations cross the wire, and the engine calls the handlers back over the protocol. The runner invokes your program via python -m boardwalk._loader <entry>.

Development

make install   # uv sync
make check     # ruff lint + format check + mypy --strict + pytest with coverage

Stdlib-only at runtime (pydantic is optional, used for input validation only when your run is annotated with a model and pydantic is importable).

Sibling repos

  • sdk-typescript@boardwalk-labs/workflow, the TypeScript SDK
  • runner — the worker runtime that serves the host protocol
  • cli — the boardwalk CLI
  • examples — reference workflows

License

MIT