openwop-client 2.x — Python SDK for OpenWOP

OpenWOP is the open-source protocol for multi-agent workflow orchestration: run AI agents as workflows you can watch, pause for human approval and replay, on any host that implements it.

openwop.dev · Quickstart

pip install "openwop-client>=2,<3"

from openwop_client import CreateRunRequest, OpenwopClient

client = OpenwopClient(base_url="https://api.example.com", api_key="hk_test_abc123")
resp = client.runs_create(CreateRunRequest(workflowId="my-workflow", inputs={"q": "hello"}))
for event in client.runs_events(resp.runId):
    print(event.sequence, event.type)

For implementers

This package is the reference Python client for the v2 major (spec/v2/, RFC 0168 §D): synchronous, zero runtime deps, one typed method per operation in spec/v2/path-manifest.json (55 operations), typed dataclasses, and pure-stdlib SSE iterators for the run and host event channels.

pip install "openwop-client>=2,<3"   # 2.10.0 — v2-only; the 1.x client stays on openwop-client<2

Spec: github.com/openwop/openwop · Corpus tag: see CORPUS_TAG · Mirrors: api/v2/openapi.yaml, schemas/v2/, spec/v2/errors.json · Sibling: the TypeScript client at sdk/typescript-v2/

The 1.x package (sdk/python/) is untouched and keeps publishing for v1 hosts. This is a v2-ONLY client: it never sends a /v1/… path. The import name stays openwop_client.

What is different from 1.x (RFC 0172 / 0171 / 0173)

Quickstart

from openwop_client import CreateRunRequest, OpenwopClient, RunConfigurable, WopError, is_terminal_run_status

client = OpenwopClient(base_url="https://api.example.com", api_key="hk_test_abc123")  # major=2 → OpenWOP-Version: 2.0

caps = client.discovery_capabilities()            # the closed v2 root
print(caps.preferredVersion, caps.protocolVersions, "webhooks" in caps.families)

resp = client.runs_create(
    CreateRunRequest(workflowId="my-workflow", inputs={"q": "hello"}, configurable=RunConfigurable(run={"runTimeoutMs": 60_000})),
    idempotency_key="idem-1",
    dedup=True,
)

for event in client.runs_events(resp.runId, stream_mode=("updates", "messages")):
    print(event.sequence, event.type)

cursor = None
while True:                                       # long-poll fallback
    page = client.runs_poll_events(resp.runId, after_sequence=cursor)
    cursor = page.lastSequence
    if page.isTerminal:
        break

comp = client.runs_compensation(resp.runId)       # None when `compensation` is unadvertised

try:
    client.runs_get("tenant/does-not-exist")
except WopError as err:
    if err.envelope and err.envelope.error == "not_found":
        ...

Webhook receivers:

from openwop_client import read_webhook_headers, verify_webhook_signature

read = read_webhook_headers(request.headers)
outcome = verify_webhook_signature(secret, read.signature, read.timestamp, raw_body, algorithm_header=read.algorithm) if read else None

Generated surface

src/openwop_client/_generated.py is produced by scripts/generate.py from the vendored spec/v2/errors.json and schemas/v2/capabilities.schema.json; python3 scripts/generate.py --check (run by scripts/sdks-check.sh) fails when it drifts.

Development

python3 scripts/generate.py --check
ruff check .
PYTHONPATH=src python3 -m unittest discover -s tests