Integrations: durable engines¶
Flows on Temporal and DBOS: Run flows on Temporal or DBOS. Each module imports its SDK lazily: install the extra first.
agent_adapter.integrations.temporal¶
Temporal: a Flow as a workflow, one superstep per activity; approvals from your store.
A reviewer decides in the approval store; an update or signal only wakes the workflow to
re-check it. The workflow holds no I/O: it calls two activities by name and waits. The
activities run each superstep through the DurableFlows runner, so the executor's
approvals, hooks, budgets, cache, ledger and events apply, and its checkpoint store lets a
retried activity continue without re-running a finished tool call.
ADVANCE = 'agent_adapter.flow.advance'
module-attribute
¶
NOTIFY = 'notify'
module-attribute
¶
PENDING = 'pending'
module-attribute
¶
RESUME = 'agent_adapter.flow.resume'
module-attribute
¶
WORKFLOW = 'agent_adapter.flow'
module-attribute
¶
FlowWorkflow
¶
The flow workflow: run(request) drives one flow run to its end (no I/O here).
check(notice)
¶
The update's validator: refuse a notice that is not for the pending approval.
notify(notice)
async
¶
The update: wake the run to re-check the approval store.
notify_signal(notice)
¶
The signal: like the update, but a refused notice is dropped and logged.
pending()
¶
The query: the approval the run waits for, or None.
run(request)
async
¶
Advance superstep by superstep; at a pause, wait for notices until the store decides.
flow_activities(flows)
¶
The two activities a worker registers: advance a run, and resume it from the store.
Raises:
| Type | Description |
|---|---|
MissingDependency
|
temporalio is not installed. |
ToolUsageError
|
|
flow_worker(client, flows, *, task_queue, **options)
¶
A Temporal Worker for the flow workflow and its activities; options go to Worker.
Raises:
| Type | Description |
|---|---|
MissingDependency
|
temporalio is not installed. |
ToolUsageError
|
|
flow_workflow()
cached
¶
A FlowWorkflow subclass with Temporal's decorators applied (built once).
Unsandboxed: the sandbox re-imports workflow modules, and this one imports temporalio lazily. The workflow code does no I/O; activities do the work.
Raises:
| Type | Description |
|---|---|
MissingDependency
|
temporalio is not installed. |
notify(handle, approval_id, *, wait=True)
async
¶
Wake a paused run after you decided approval_id in the approval store.
The notice carries no verdict: the run re-checks the store and goes on only with a
decided record there, once. wait=True is an update the workflow refuses unless it
waits for that approval; wait=False is a signal.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
|
WorkflowUpdateFailedError
|
the workflow refused it ( |
pending(handle)
async
¶
The approval the run waits for (no arguments), or None.
result(handle)
async
¶
Wait for the run to end and return its report (done or denied).
start_flow(client, flow, state, *, run_id, task_queue, approval_timeout_s=None, activity_timeout_s=600.0, max_attempts=0, identity=None)
async
¶
Start a flow run as a workflow whose id is the run id; max_attempts=0 retries forever.
identity (user, thread, principal) is sealed into the run; every step acts for it.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
a bad flow name, state, run id, queue, timeout or identity. |
Also exported here, documented under the canonical path: ApprovalNotice, DurableFlows, FlowStepReport, PendingApproval, RunIdentity.
agent_adapter.integrations.dbos¶
DBOS: a Flow as a @DBOS.workflow, one superstep per @DBOS.step; approvals from your store.
A reviewer decides in the approval store; a DBOS message only wakes the workflow to re-check it.
Each step runs a superstep through the DurableFlows runner, so the executor's approvals,
hooks, budgets, cache, ledger and events apply. DBOS records each finished step; a step cut
short by a crash runs again, and the runner's checkpoint store replays its finished tool
calls instead of calling them twice.
ADVANCE = 'agent_adapter.flow.advance'
module-attribute
¶
NOTIFY_TOPIC = 'agent_adapter.notify'
module-attribute
¶
PENDING_EVENT = 'agent_adapter.pending'
module-attribute
¶
RESUME = 'agent_adapter.flow.resume'
module-attribute
¶
WORKFLOW = 'agent_adapter.flow'
module-attribute
¶
ApprovalNotice
¶
Bases: BaseModel
A wake-up for a paused run: re-check the approval store for this approval's decision.
It carries no verdict. Only a decided record in your approval store resumes the run.
DbosFlows
¶
The DBOS workflow and steps for flows; build it once per process, before DBOS.launch().
Whoever can send messages to the workflow can approve: keep DBOS's system database (and its admin access) private.
__init__(flows, *, max_attempts=5, interval_s=1.0)
¶
Register the workflow and its two steps with DBOS.
Raises:
MissingDependency: dbos is not installed.
ToolUsageError: flows is not a DurableFlows, or a bad retry setting.
notify(run_id, approval_id)
async
¶
Wake a paused run after you decided approval_id in the approval store.
The message carries no verdict: the run re-checks the store and goes on only with a decided record there, once.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
an empty run id or approval id. |
pending(run_id, *, timeout_s=0.0)
async
¶
The approval the run waits for (no arguments), or None.
result(handle)
async
staticmethod
¶
Wait for the run to end and return its report (done or denied).
start(flow, state, *, run_id, approval_timeout_s=None, identity=None)
async
¶
Start a flow run as a workflow whose id is the run id; starting it again is a no-op.
identity (user, thread, principal) is sealed into the run; every step acts for it.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
a bad flow name, state, run id, timeout or identity. |
DurableFlows
¶
Runs registered flows one superstep per call, for a durable engine's activities or steps.
Every call goes through runner, so its executor's approvals, hooks, budgets, cache,
ledger and events still apply. The runner's checkpoint store is the source of truth: a
call that is retried after a crash continues the run, and finished tool calls replay.
names
property
¶
The registered flow names.
__init__(runner, flows)
¶
Wrap a checkpointed runner and the flows it may run.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
|
advance(flow, run_id, state=None, identity=None)
async
¶
Start the run (with state) or continue it by one superstep; idempotent per position.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
unknown flow, a new run without |
CheckpointConflict
|
another worker holds the run (retry). |
RecoveryNeedsDecision
|
a call may have run before a crash and is not idempotent. |
FlowError
|
a node, route or merge failed (the run is marked failed). |
flow(name)
¶
pending_request(flow, run_id, identity=None)
async
¶
The full pending request (arguments included) of a paused run, for a reviewer.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
unknown flow, a run id held by another flow, or another identity. |
resume(flow, run_id, approval_id, *, timed_out=False, identity=None)
async
¶
Re-check the approval store; a decided record resumes the paused superstep once.
A record still pending (or no record: a forged notice) leaves the run paused. With
timed_out, a still-pending record is first decided as denied in the store. A retry
after the decision was used continues the stored run instead.
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
unknown flow, the approval's port has no store, or as |
CheckpointNotFound
|
nothing is stored for |
ApprovalAlreadyDecided
|
the store record was already used. |
(CheckpointConflict, RecoveryNeedsDecision, FlowError)
|
as |
FlowStepReport
¶
Bases: BaseModel
Where a durable run stands after one step; state is set once it is done or denied.
PendingApproval
¶
Bases: BaseModel
What a paused run waits for; the tool arguments stay out of engine history.
RunIdentity
¶
Bases: BaseModel
Who a durable run acts for; sealed into its checkpoints, so every step must match.
of(raw)
classmethod
¶
An identity from engine input (None is the empty identity).
Raises:
| Type | Description |
|---|---|
ToolUsageError
|
not an identity. |
StepRetry
¶
Bases: Exception
A step failed on something passing (a store or network blip): DBOS runs it again.