Skip to main content

Python

quantumbpm-sdk — async-first, built on asyncio. Requires Python 3.10+.

Installation​

pip install quantumbpm-sdk

Quick start​

import asyncio
from quantumbpm import QuantumBPM, Vars, ZitadelTokenProvider

async def main():
    provider = ZitadelTokenProvider(
        "./service-account.json",
        "https://auth.quantumbpm.com",
        "your-zitadel-project-id",
    )

    async with QuantumBPM(
        base_url="https://api.quantumbpm.com",
        project_id="YOUR_PROJECT_ID",
        token_provider=provider,
    ) as client:
        result = await client.dmn.evaluate(
            "loan-eligibility",
            Vars().set("requestedAmt", 1000).set("creditScore", 720),
        )
        print(result)

asyncio.run(main())

The async context manager (async with) acquires a fresh bearer token on entry; calling methods directly outside the context also works — the SDK refreshes tokens on demand.

BPMN​

draft = await client.bpmn.create_resource("loan-process", bpmn_xml)
await client.bpmn.deploy_resource(draft.id)

deployed = await client.bpmn.get_resource(draft.id)
process_def = deployed.processes[0]

workflow_id = await client.bpmn.start_instance(
    process_def.id,
    Vars().set("applicantID", "u-123"),
)

External job workers​

Decorator-based registration. The runtime uses asyncio task groups per task type for concurrency.

from quantumbpm import Vars, BpmnError
from quantumbpm.workers import Job

worker = client.new_worker(client_id="billing-svc")
stop = asyncio.Event()

@worker.handler("send-email", max_jobs=10, lock_duration="1m")
async def handle(job: Job) -> Vars:
    recipient = job.vars.lookup("recipient")
    await emailer.send(recipient)
    return Vars().set("messageID", "msg-123")  # → Complete

await worker.run(stop)  # resolves when stop.set() is called

Raise a BpmnError to fail with a code boundary error events can catch:

raise BpmnError("INSUFFICIENT_FUNDS", Vars().set("balance", 12.0))

Type-annotate Job[T] with a Pydantic model for opt-in typed dispatch — the runtime decodes job variables into T before invoking the handler:

from pydantic import BaseModel

class EmailJob(BaseModel):
    recipient: str
    subject: str

@worker.handler("send-email")
async def handle(job: Job[EmailJob]) -> Vars:
    await emailer.send(job.typed.recipient, job.typed.subject)
    return Vars()

Full reference​

See the SDK README for the complete API surface and the Vars typed-access helpers.