Skip to content

Latest commit

 

History

17 Commits

Folders and files

NameName
Last commit message
Last commit date
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

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

About

The Python SDK for Boardwalk agent workflows: async def run(input, context) + imported capabilities

Resources

Stars

1 star

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages