Official Python SDK for the QuantumBPM platform - DMN evaluation, BPMN process orchestration, and external job workers.
pip install quantumbpm-sdkPython 3.10+. Async-first; built on asyncio.
| Module | Purpose |
|---|---|
quantumbpm.QuantumBPM |
Top-level client exposing .dmn, .bpmn, plus new_worker(...) |
quantumbpm.auth |
TokenProvider, ClientCredentialsTokenProvider, StaticTokenProvider |
quantumbpm.dmn |
DMN evaluation: stored definitions, ad-hoc XML, batch |
quantumbpm.bpmn |
BPMN resources, instances, messaging, user tasks, processes |
quantumbpm.workers |
External job worker runtime - long-poll, lock heartbeat, dispatch |
quantumbpm.variables |
Vars wrapper with typed accessors and FEEL-context conversion |
quantumbpm.api[_client] |
OpenAPI-generated client. Reachable via client.raw, never hand-edited |
import asyncio
import os
from quantumbpm import ClientCredentialsTokenProvider, QuantumBPM, Vars
async def main():
provider = ClientCredentialsTokenProvider(
token_url="https://idp.example.com/realms/acme/protocol/openid-connect/token",
client_id="billing-svc",
client_secret=os.environ["QUANTUMBPM_CLIENT_SECRET"],
)
async with QuantumBPM(
base_url="https://quantumbpm.example.com",
project_id="00000000-0000-0000-0000-000000000000",
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) fetches a bearer token on entry, so an authentication problem shows up right away. Every API call also asks the token provider for a token first, so calling methods without the context manager works too, and a long-running worker keeps sending a valid token.
The SDK awaits TokenProvider.get_token() before every API call and sends the result as the bearer token. Two implementations ship out of the box.
ClientCredentialsTokenProvider fetches access tokens from your OIDC identity provider (Keycloak, Microsoft Entra ID, Okta, Auth0 and others) with the client credentials grant. Register a confidential client for your service at the identity provider and pass its token endpoint and credentials:
import os
from quantumbpm import ClientCredentialsTokenProvider
provider = ClientCredentialsTokenProvider(
token_url="https://idp.example.com/realms/acme/protocol/openid-connect/token",
client_id="billing-svc",
client_secret=os.environ["QUANTUMBPM_CLIENT_SECRET"],
scopes=["openid"], # optional
audience="quantumbpm", # optional, some providers such as Auth0 require it
ssl_ca_cert="/path/to/ca.crt", # optional, for an identity provider behind a private CA
)The exact token endpoint depends on the identity provider. The provider's OIDC discovery document (/.well-known/openid-configuration) lists it as token_endpoint.
scopes are sent space-joined as the scope parameter and audience as the audience parameter. Each is left out of the request when empty. The client must be set up at the identity provider so its tokens carry the audience and roles your QuantumBPM deployment expects.
The provider caches the token in memory and fetches a new one once less than 60 seconds of its lifetime remain, so the lookup before each call is cheap. Concurrent callers wait for one shared request. A non-2xx response from the token endpoint raises quantumbpm.exceptions.ApiException with the HTTP status and response body.
For a token acquired out of band, for example by another library or in tests:
from quantumbpm import StaticTokenProvider
provider = StaticTokenProvider("eyJhbGciOi...")Implement the Protocol:
from quantumbpm import TokenProvider
class MyProvider:
async def get_token(self) -> str:
return await fetch_my_token()
provider: TokenProvider = MyProvider()get_token runs before every API call, so cache the token when fetching one is expensive.
The client.dmn sub-client offers four async methods.
result = await client.dmn.evaluate(
"loan-eligibility",
Vars().set("requestedAmt", 5000).set("creditScore", 720),
)Returns dict[str, EvaluationResult] keyed by decision name. Each result has value, hit_rules, error, and type.
Pin a version, restrict the evaluated decisions, or attach decision services:
result = await client.dmn.evaluate(
"loan-eligibility",
vars,
version=3,
decisions=["eligibility", "rate"],
)For "evaluate while editing" flows that don't store the XML:
result = await client.dmn.evaluate_design(
dmn_xml,
vars,
additional_xmls=[imported_xml1, imported_xml2],
decisions=["eligibility"],
)rows = [
Vars().set("requestedAmt", 1000),
Vars().set("requestedAmt", 5000),
Vars().set("requestedAmt", 25000),
]
batch = await client.dmn.evaluate_design_batch(dmn_xml, rows)client.bpmn covers the full BPMN runtime surface.
draft = await client.bpmn.create_resource("loan-process", bpmn_xml)
await client.bpmn.deploy_resource(draft.id)
# Re-fetch to get the process versions the deploy created.
deployed = await client.bpmn.get_resource(draft.id)
process_def = deployed.processes[0]
# Starts the latest version. Pass version=N to pick one.
instance_id = await client.bpmn.start_instance(
process_def.process_id,
Vars().set("applicantID", "u-123").set("requestedAmt", 25000),
)state = await client.bpmn.get_instance(instance_id)
print(state.status, state.active_scopes)
vars = await client.bpmn.get_instance_variables(instance_id)
children = await client.bpmn.get_instance_children(instance_id)
# History pages run oldest first. Without options a page holds the newest
# entries, and an entry read by its seq carries its inputs and outputs.
page = await client.bpmn.get_instance_history(instance_id, from_seq=1, limit=100)
entry = await client.bpmn.get_instance_history_entry(instance_id, page.items[0].seq)
version = await client.bpmn.get_process_version("loan-process", 2)await client.bpmn.publish_message(
"loan-approved",
Vars().set("approvedAmt", 24000),
correlation_keys={"applicantID": "u-123"},
ttl="PT5M",
)
await client.bpmn.publish_signal("system-maintenance")page = await client.bpmn.list_user_tasks(
assignee="alice@example.com",
status="CREATED",
)
# Lists of instances and user tasks page by cursor: pass the page's
# next_cursor back for the next one. It is None on the last page.
if page.pagination.next_cursor:
page = await client.bpmn.list_user_tasks(
assignee="alice@example.com",
status="CREATED",
cursor=page.pagination.next_cursor,
)
# The caller's tasks soonest due first, and the project's overdue ones.
mine = await client.bpmn.list_user_tasks_for_caller(sort="dueDate")
overdue = await client.bpmn.list_user_tasks(status="CREATED", due_before=datetime.now(timezone.utc))
await client.bpmn.complete_user_task(execution_key, Vars().set("approved", True))
# Or fail with a BPMN error code (matches boundary error events):
await client.bpmn.throw_user_task_error(execution_key, "REVIEW_REJECTED")Workers handle service tasks asynchronously. Register a handler per task type with a decorator, then call run. The runtime owns long-polling, lock heartbeats, dispatch, and outcome mapping.
import asyncio
from quantumbpm import QuantumBPM, Vars, BpmnError
from quantumbpm.workers import Job
async def main():
async with QuantumBPM(...) as client:
worker = client.new_worker(client_id="billing-svc")
stop = asyncio.Event()
@worker.handler("send-email")
async def handle(job: Job) -> Vars:
recipient = job.vars.lookup("recipient")
subject = job.vars.lookup("subject")
await emailer.send(recipient, subject)
return Vars().set("messageID", "msg-123") # → Complete
await worker.run(stop) # blocks until stop is set
asyncio.run(main())run(stop) resolves once the asyncio.Event is set and running handlers have finished, or the shutdown timeout has passed. See Heartbeats, lost locks, and shutdown.
@worker.handler(
"send-email",
max_jobs=10, # up to 10 in flight per task type
poll_timeout="5s", # long-poll wait per request, at most 5s
lock_duration="2m", # exclusive lock per job
)
async def handle(job: Job) -> Vars:
...Concurrency is per task type. Different task types run independently.
poll_timeout and lock_duration take Go duration strings such as 30s, 1m30s, 12.5s or 500ms. handler and handle raise ValueError for a value that does not parse. One poll request waits at most 5 seconds on the server, and a larger poll_timeout is capped to that. When a poll returns no job, the worker polls again right away.
The server clamps every lock to between 10 seconds and 5 minutes by default (ENGINE_WORKER_LOCK_MIN and ENGINE_WORKER_LOCK_MAX). Each polled job carries job.raw.lock_expires_at, the expiry of the lock the server granted. While the handler runs, the runtime heartbeats at half of the granted lock, or at half of lock_duration when that is shorter. A lock_duration of 11m gets a 5 minute lock and a heartbeat every 2.5 minutes.
A polled job is locked to this worker. Each heartbeat extends the lock by lock_duration, clamped the same way, and the server caps one attempt at 30 minutes by default (ENGINE_JOB_MAX_HOLD). When the lock runs out, the server charges the job a retry and offers it again, or raises an incident when no retries are left. So abandoning a running job always costs a retry.
When a heartbeat answers that this worker no longer holds the job (HTTP 409 or 404), the runtime cancels the handler with asyncio.CancelledError and reports nothing. A complete or error report rejected with 409 is discarded by the server and logged as a warning. Let CancelledError propagate out of handlers.
Setting the stop event shuts the worker down gracefully:
- No new poll starts. A poll already in flight cannot be cancelled. It returns within 5 seconds, and jobs it returns before the shutdown timeout still run.
- Running handlers keep heartbeating, and their results are reported.
- Handlers still running
shutdown_timeoutseconds (default20.0) afterstopis set are cancelled and not reported. Their locks run out on the server. Thenrunreturns.
Wire stop to SIGTERM so a stop signal drains the worker:
import asyncio
import signal
from quantumbpm import QuantumBPM, Vars
from quantumbpm.workers import Job
async def main():
async with QuantumBPM(...) as client:
worker = client.new_worker(client_id="billing-svc", shutdown_timeout=20.0)
@worker.handler("send-email")
async def handle(job: Job) -> Vars:
...
stop = asyncio.Event()
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGTERM, stop.set)
loop.add_signal_handler(signal.SIGINT, stop.set)
await worker.run(stop)
asyncio.run(main())On Kubernetes, set terminationGracePeriodSeconds above the shutdown timeout, or the pod is killed mid-drain. A poll in flight at stop can also hold process exit for up to 5 seconds, so leave room for that too.
Raise a BpmnError to fail the job with a code that boundary error events on the originating service task can catch:
@worker.handler("charge-card")
async def handle(job: Job) -> Vars:
try:
tx_id = await charge(job.vars.to_dict())
return Vars().set("transactionID", tx_id)
except InsufficientFundsError:
raise BpmnError("INSUFFICIENT_FUNDS", Vars().set("availableBalance", 12.0))Any other exception is reported as WORKER_ERROR, which the server treats as a retryable failure that decrements the job's retry budget. The exception message goes into the error variable, cut to max_error_message_bytes bytes of UTF-8 (a new_worker option, default 2048).
Type-annotate the Job parameter with a Pydantic model - the runtime validates the job's input variables before invoking the handler. The decoded value lands in job.typed.
from pydantic import BaseModel
from quantumbpm.workers import Job
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().set("messageID", "msg-123")Decode failures become WORKER_ERROR automatically, with the validation message in the variables.
Vars is a thin wrapper around dict[str, Any] shared by DMN, BPMN, and workers.
v = Vars().set("amount", 100).set("name", "Alice")
v = Vars.from_dict({"amount": 100, "name": "Alice"})amount = v.get("amount", float)
flag = v.get("approved", bool)
class Loan(BaseModel):
requestedAmt: float
approved: bool
loan = v.as_type(Loan)Vars.get(name, type_) and Vars.as_type(type_) accept Pydantic models, dataclasses, and primitives - Pydantic's TypeAdapter does the validation.
The client.raw property exposes the underlying generated ApiClient for endpoints not yet wrapped (instance migration, modification, ad-hoc triggers, batch job complete/error, etc.):
from quantumbpm.api.bpmn_api import BpmnApi
await client.authenticate()
api = BpmnApi(client.raw)
result = api.migrate_bpmn_instance(client.project_id, instance_id, body)Calls through client.raw send the token of the last SDK call. await client.authenticate() puts a current token on it first.
MIT License - see LICENSE for details.