| name | agently-triggerflow |
| description | Use when the user needs workflow orchestration such as branching, concurrency, approvals, waiting and resume, runtime stream, restart-safe execution, mixed sync/async function or module orchestration, event-driven fan-out, process-clarity refactors that make stages explicit, performance-oriented refactors that collapse split requests, or workflow definitions and chunk-level runtime metadata that must stay visible for debugging and visualization. The user does not need to say TriggerFlow explicitly. |
Agently TriggerFlow
Use this skill when the solution clearly needs orchestration semantics rather than one request family.
The user does not need to say TriggerFlow or Agently. Scenario language such as resumable approval flow, branching automation, output-fan-out refactor, mixed sync/async pipeline, process-clarity refactor, or draft-review-revise pipeline should still route here once orchestration is clearly the owner layer.
Native-First Rules
- prefer TriggerFlow for explicit multi-stage quality loops, branching, concurrency, waiting/resume, restart-safe execution, output-fan-out performance refactors, process-clarity refactors, or mixed sync/async orchestration
- treat TriggerFlow as Agently's first-class orchestration substrate: framework
features with multi-step loops, approval waits, retry, replan, verification,
or resume semantics should be TriggerFlow-backed rather than hidden inside a
local executor state machine
- prefer
agent.create_task(...) when the app needs one Agent-owned business
task loop where the model owns planning, verification, and replan; it returns
a task-strategy AgentExecution draft, not a separate public AgentTask
handle. Use agent.create_task_loop(...) only when the code should make the
task-loop strategy explicit; it still returns an AgentExecution draft. Use
TriggerFlow directly when the application owns explicit stages, branching,
pause/resume, or restart topology
- for AgentTaskLoop examples, consume the AgentExecution result/stream/meta
(
result = execution.get_result(), result.get_text(),
result.get_data(), result.get_meta(),
execution.get_async_generator(), execution.async_get_meta()) and surface task refs or
meta.stream_kind=="snapshot" items as compact intermediate state captures;
enable options={"agent_task": {"stream_progress": True}} only when the
example or host needs natural-language operator updates; omit
progress_model_key for template progress with no model requests, or set
progress_model_key to run a separate background model that summarizes only
existing snapshots/task metadata without adding main-loop fields or latency;
model-generated progress streams progress_delta items before the final
progress item; use progress_language or the global
agent_task.progress.language setting when the host needs a fixed language;
prove replan behavior from stream verification/replan events and snapshots
rather than hiding the proof in a local Python loop; mocked business systems
may supply defective facts or conflicting source data, but must not return
pass/fail verdicts or deterministic quality judgments
- default to async-first workflow handlers, execution entrypoints, and runtime stream consumers
- treat sync TriggerFlow APIs as wrappers for scripts or compatibility bridges, not as the default service interface
- derive complex workflow topology from the real business dependencies before
implementation: mark required serial edges, independent branches, provisional
structured progress that can support UI or cancelable/idempotent preparation,
and side-effect/capacity
constraints. Use
batch(...), for_each(...), or when(...) plus managed
emits for bounded concurrent work and graph-visible joins. Choosing an
all-serial topology without this analysis is a prohibited anti-pattern
- prefer explicit execution lifecycle control with
close() / async_close() for completion and cleanup
- use
flow.start(...) / flow.async_start(...) and flow-level runtime-stream
helpers only for a finite, self-closing run when the caller does not need an
execution handle. The boundary is lifecycle control, not script versus
service: a bounded request handler may use hidden sugar, while pause/resume,
external emits, save/load, intervention, inspection, cancellation, or
host-controlled close requires an explicit execution
- treat
create_execution(concurrency=N) and execution.set_concurrency(N) as an execution-wide handler dispatch budget, including nested dispatch from chunk continuations and data.async_emit(...); use operator-local batch(..., concurrency=...) or for_each(..., concurrency=...) only for local fan-out caps
- expose pressure controls at the layer that owns them: host admission and
in-flight execution/coroutine limits, TriggerFlow execution concurrency,
operator-local fan-out caps, model scheduler concurrency/rate limits, and
host worker or thread-pool sizes when blocking code is isolated. Do not
present worker or thread counts as a universal TriggerFlow setting
- use
execution.result when services, UIs, stream consumers, or intervention-aware workflows need multiple views of one execution outcome, such as state, compatibility final result, interventions, and metadata; use execution.close() / execution.async_close() for close snapshots
- use runtime intervention for optional guidance context added while an execution is already running: define explicit
.intervention_point(...) boundaries so execution creation can infer planned mode, or create the execution with intervention_mode="auto" for boundary policy insertion; chunks read inserted context with data.get_interventions(...) and explicitly audit usage with data.async_mark_intervention_consumed(...), relying on the chunk-name consumer default unless another consumer identity is clearer
- for human approvals, webhooks, or externally resumed waits, use
pause_for(..., resume_to="next" | "self" | {"event": ...}); treat it as a durable graph interrupt, not Python coroutine stack persistence; teach model-decided autonomous interrupts with model-owned pause_for(..., resume_to="self"), where the resumed chunk handles data.is_resume and the default max_resumes=1 prevents unbounded self-replay; teach prearranged approval gates with an explicit pause chunk plus when(...)
- for framework policy approvals, use the global PolicyApproval contract and
represent pending approvals as
pause_for(type="policy_approval", ...)
interrupts that resume through continue_with(...)
- for host-owned approval, webhook, queue, or UI-card transports, use the
ExecutionExchange provider seam instead of inventing a second wait/resume
channel: the provider publishes typed requests, while TriggerFlow owns the
interrupt ledger and all resumes still go through
continue_with(...)
- when a TriggerFlow exchange should be rendered by a host, project it through
agently.base.execution_exchange.project_pending_exchanges(execution) or
project_execution_exchanges(execution); do not make host code depend on raw
TriggerFlow interrupt internals as the UI contract
- do not put
pause_for(...) behind hidden execution sugar such as flow.start() or flow-level runtime stream helpers; create an explicit execution handle and consume get_pending_interrupts() / continue_with(...)
- close waiting executions explicitly:
close() / async_close() rejects pending interrupts by default, and pending_interrupts="cancel" must be chosen deliberately when abandoning waits
- when a sub-flow can pause, keep the external API rooted at the parent execution id plus projected root interrupt id; do not require callers to manage child execution ids
- use
emit_nowait(...) / async_emit_nowait(...) when a chunk must fan out without blocking the current handler, and rely on execution close to drain registered tasks
- after starting a finite workflow that uses execution-managed nowait fan-out,
call
await execution.async_close() to drain registered tasks and collect the
close snapshot. Do not add while True status polling loops such as
execution.result.get_state("status") unless a chunk explicitly owns and
updates that status contract
- rely on chunk-internal
data.emit(...), data.async_emit(...), data.emit_nowait(...), and data.async_emit_nowait(...) to inherit the current TriggerFlow runtime scope; do not assume unrelated external emits can be paired by when(..., mode="and") unless the host routes them through one scoped flow stage or carries explicit correlation in the payload
- use execution runtime state through
data.get_state(...) / data.set_state(...), async variants, and execution.result.get_state(...) as the per-execution data store and chunk-to-chunk handoff contract. Do not create parallel per-execution stores, storage helpers, translation helpers, or shadow dictionaries to shuttle workflow runtime data between chunks
- use Workspace, provider, or host storage only when data must outlive one execution, be shared across runs, or be externalized as a large artifact; keep compact refs, status, and audit facts in TriggerFlow execution state so runtime graphs and recovery snapshots stay coherent
- treat shared flow data /
flow_data as a risky cross-execution surface and avoid it unless the task explicitly needs shared state; do not use it as a substitute for execution state
- document its compatibility persistence exactly: the
execution.save()
snapshot includes a serialized copy of flow_data, and load() replaces the
current flow-shared value with that copy. This does not provide
execution-local isolation, CAS, merge behavior, or concurrency safety
- when discussing restart or distributed pause/resume, describe TriggerFlow as
providing foundations for host-managed recovery, not a complete production
distributed workflow engine.
execution.save() is a versioned top-level
execution snapshot with durable TriggerFlow progress, flow definition
fingerprint validation, interrupt/resume ledgers, resource requirements,
lease metadata, and managed execution resource requirements. TriggerFlow
core is process-stateless between save/load; live runtime_resources,
clients, callbacks, tasks, semaphores, stateful sessions, and Python
coroutine frames are not serialized. runtime_resources is only the
process-local mount point for live objects that the host has already created,
restored, and validated. If a resource carries state, the external system
that owns it must persist a state ref/version/lease or fence token and a
resolver/provider must validate it before load is ready. Declare future
resources with flow.declare_resource_requirement(...) or
execution.declare_resource_requirement(...), inspect or restore with
execution.inspect_load(...) / execution.async_load(...), pass a stable
resume_request_id to continue_with(...) for external callbacks, and
persist through a snapshot store that implements put_snapshot(...) while
the production store owns atomic claim, lease enforcement, conflict handling,
outbox ordering, and external side-effect idempotency. Fail-closed pending
resolvers or pending managed execution resources should be treated as
status="pending_resources" and ready=False until async_load(...)
resolves and validates live resources. Late callbacks delivered to an
expired execution-local lease fail fast before resume acceptance without
writing resume ledger entries; the reclaimed worker should load or claim
first and then use the same stable resume_request_id. For long-running
executions, use execution.set_compaction_policy(...) when snapshot writes
should automatically run a host-owned reducer and store large payloads behind
provider artifact refs; TriggerFlow records compaction facts, retained
anchors, and load read bounds, while Workspace or an enterprise provider owns
artifact storage and retention anchor persistence. For framework-level
validation, run examples/trigger_flow/durable_recovery.py when available; it
proves Workspace-backed snapshot load and duplicate callback idempotency
without pretending a production approval transport exists. Use
examples/trigger_flow/fastapi_sqlite_exchange_provider.py when service-level
provider replacement should be shown: it packages the flow as a module-level
TriggerFlow(...) object with top-level .to(...) / .when(...) wiring,
stores top-level execution snapshots in SQLite, and routes approval waits
through a SQLite ExecutionExchangeProvider exposed by FastAPI without
claiming production distributed guarantees. If a snapshot
fingerprint is missing or does not match the current flow definition,
inspect_load(...) reports invalid_snapshot and load(...) rejects the
snapshot
- for service packaging, prefer an importable module that owns the complete flow
definition: a module-level
TriggerFlow(...) object, module-level chunk
handlers, and module-level .to(...) / .when(...) wiring. Service modules
should import that flow object and start executions from it. Normal Python
imports execute the module body once per process for the same module name;
TriggerFlow duplicate-definition protection is only the second line of
defense when application code explicitly runs the same wiring again on the
same flow object. Create live dependencies in host-owned factories or
importable resolvers, and use runtime_resources only as the final
process-local attachment point for already-created live objects. Use a
build_flow(...) helper only when the application genuinely needs multiple
configured flow instances or test isolation, not as the default service shape
- route model-generated or app-submitted DAG data to
agently-dynamic-task; TaskDAG is the DAG foundation capability, Dynamic Task is the convenience facade over it, and TriggerFlow is the execution substrate rather than the facade API
- when discussing the new Blocks lifecycle, describe it as the framework
lowering bridge from bounded ExecutionPlan / PlanBlock instances or
validated TaskDAG nodes into TriggerFlow-backed ExecutionBlockGraph. Do not
tell users to generate live Python TriggerFlow chunks from planner output for
the current run; trusted ExecutionBlocks, handlers, Workspace resources, and
PolicyApproval waits must be registered or provided by the host. Blocks
wait blocks record waiting evidence, while resume state remains in the
TriggerFlow interrupt/resume ledger
- ordinary
TaskDAGExecutor.async_run(...) validates and compiles directly to
TriggerFlow. Blocks is an explicit opt-in carrier only through
compile_blocks(...) / async_run_blocks(...) when the caller needs an
ExecutionBlockGraph, Blocks evidence, or result adapters
- use
when(...) + emit_nowait(...) as the native signal-driven pattern for fan-out, loops, side branches, and dependency joins. Express looping behavior as a graph-visible back edge: a chunk emits the next iteration, retry, or revision signal and flow.when(...).to(...) routes it to the target chunk, or the flow uses an intentional .to(...) continuation back to an earlier named chunk. Definition idempotence must not be confused with runtime signal deduplication
- for a developer-owned Todo DAG or other dependency graph represented as
stable Python flow code, express multi-dependency joins with
flow.when(["task_a_done", "task_b_done"], mode="and").to(...) and have
upstream chunks call data.emit_nowait("task_a_done", payload). Do not
replace joins with sleeps, polling loops, local completed sets, or
pause_for(..., resume_to="self") unless the workflow is genuinely waiting
for external input across process time
- Do not compile model-generated or app-submitted DAG data directly into new TriggerFlow definitions.
Route that data through TaskDAG / DynamicTask validation and resolver
handlers. Direct
flow.to(...) / flow.when(...) wiring is reserved for
stable topology that the developer owns in trusted source code, not runtime
plan data
- when a TriggerFlow + Skills example relies on a trusted local Skill
bundle to provide declared helper capabilities, pass selector-level
auto_allow=True or use settings-backed
access_control_policy.auto_allow for trusted hosts; Skill metadata alone is
not a capability grant, and selector-level auto_allow must stay scoped to the
matching Skill
- keep runtime stream consumers safe by relying on execution close to stop the stream
- keep workflow stages visible instead of hiding nested request loops
- name chunks and stage boundaries so exported flow configs, Mermaid diagrams, and runtime graphs stay readable
- let TriggerFlow definition export and runtime metadata drive visualization instead of maintaining a second manual graph description
- combine with
agently-request when one workflow step needs model setup, prompt contracts, structured output, response reuse, session behavior, or retrieval
Python API Shape
When generating or editing Python code, use the actual Agently API shape:
from agently import Agent, Agently, TriggerFlow, TriggerFlowRuntimeData, Workspace
agent = Agent()
factory_agent = Agently.create_agent()
flow = TriggerFlow(name="workflow-name")
factory_flow = Agently.create_trigger_flow("factory-workflow")
shared_workspace = Workspace("./.agently/workflows/shared")
Agent() / Agently.create_agent() and TriggerFlow() /
Agently.create_trigger_flow(...) are both valid first-class creation styles.
Workspace(...) / Agently.create_workspace(...) are both valid first-class
Workspace creation styles. Use one style consistently in an example unless
showing the API equivalence.
when, emit_nowait, and pause_for are not top-level imports from
agently. They are methods on flow or runtime data objects:
- define graph branches with
flow.when(...).to(handler, name="...")
- fan out from inside a chunk with
data.emit_nowait(...)
- pause from inside a chunk with
await data.async_pause_for(...)
- create and close long-running/manual executions with
execution = flow.create_execution(auto_close=False),
await execution.async_start(input_value) using a positional start value, and
snapshot = await execution.async_close()
flow.create_execution() binds the current session/script default Workspace
by default and assigns the execution a scoped file root under
files/lineage/<root-kind>/<root-id>/.../execution/<execution-id>/files;
pass workspace=False to opt out, or
flow.create_execution(workspace=shared_workspace) when an application-owned
Workspace should be selected explicitly for Agents, service workers, or other
executions- do not call
execution.async_start(input_value=...); pass the start value
positionally
- do not assume
execution.async_start("start") emits a custom "start" event.
It starts the execution with "start" as the start input. Use flow.to(...)
for start-bound chunks, or explicitly call await execution.async_emit("start", payload)
after start when a custom event is intentional
async_close() / close() returns the close snapshot as a dict. Read it as
snapshot["key"] or inspect execution state through
execution.result.get_state("key"); do not write snapshot.state,
snapshot.pending_interrupts, or execution.result.state- create named flows with
TriggerFlow(name="...") or unnamed flows with
TriggerFlow(). Do not write TriggerFlow("name"); the first positional
argument is not the workflow name in the current API
Do not write from agently import when, emit_nowait, pause_for, and do not use
@flow.when(...) as a decorator.
Dynamic Task Boundary
Do not treat Dynamic Task as TriggerFlow syntax. If a model or application
submits a DAG as data, route to agently-dynamic-task and use
Agently.create_dynamic_task(...), TaskDAGValidator, and TaskDAGExecutor
there. TriggerFlow remains the substrate for stable workflow definitions that
the developer owns in code.
Anti-Patterns
- do not default a complex workflow to serial execution merely because a serial
loop is easier to write; keep serial edges only where dependencies, ordering,
side-effect safety, or external capacity require them
- do not invent a custom event bus or state machine before checking TriggerFlow
- do not implement a custom DAG scheduler in TriggerFlow when Dynamic Task can validate and execute submitted task graphs
- do not use untracked
asyncio.create_task(data.async_emit(...)) as the default nowait pattern when execution-managed emit_nowait(...) is available
- do not recommend
.end(), get_result(), or set_result() as the default lifecycle path for new TriggerFlow code
- do not add custom result containers around TriggerFlow executions when
execution.result can expose the needed state/final-result/intervention/meta view
- do not use runtime intervention as a required wait gate; use
pause_for(...) / continue_with(...) when the workflow must stop for external input
- do not treat
intervene(...) as emit(...), graph mutation, input rewriting, chunk cancellation, or replay of completed chunks
- do not use
get_runtime_data(...) / set_runtime_data(...) in new guidance when get_state(...) / set_state(...) communicates the same intent
- do not treat repeated silence after one deprecation warning as approval; Agently emits each deprecated API warning once per Python process
- do not treat
runtime.show_deprecation_warnings=False as a migration substitute; it is only a production noise-control setting
- do not use flow data for per-execution state
- do not write guidance that depends on code after
await data.async_pause_for(...) surviving a process restart; put post-resume logic in the downstream chunk, data.is_resume branch, or explicit resume event handler
- do not make service chunks depend on closure-captured business context when
runtime_resources would keep the handler reusable, testable, and export-friendly
- do not pass raw model stream paths directly to the UI when the workflow can translate them into stable business events
- do not hide draft/judge/revise or similar loops inside one opaque helper
- do not make DevTools or graph tooling the source of truth for workflow structure when TriggerFlow definitions already are
Read Next
references/overview.mdreferences/runtime-intervention.mdreferences/stream-bridge.mdreferences/devtools-graph.md
