Core Concepts

TriggerFlow is built around signals, handlers, and runtime state.

Flow and Execution

• TriggerFlow (Flow): the orchestration definition (blueprint + rules).
• Execution: one runtime instance created by start().

Signals and trigger types

TriggerFlow supports three trigger types:

• event: signals from emit()
• runtime_data: signals fired by set_runtime_data()
• flow_data: signals fired by set_flow_data() for all executions

Note: flow_data affects all running and future executions under the same Flow. Use it only when you are certain. If you use a Flow as a shared service object to create per-user executions, avoid flow_data to prevent data leakage or cross-request contamination.

All of them can be captured by when().

to and Chunk (completion signal)

to() binds a handler to the current trigger signal. When a Chunk finishes, it emits its own completion signal:

• each Chunk has a unique trigger (e.g. Chunk[xxx]-<id>)
• upstream signal triggers the Chunk
• completion emits the Chunk trigger
• downstream can be to() or when()

Chunk: explicit naming and recommended structure

For production workflows, define named chunks with @flow.chunk first, then wire them with flow.to() / flow.when().to(). This keeps the graph reusable and readable.

Passing a function directly into to() is a convenience for quick experiments. The recommended pattern is to promote handlers to chunks and bind them by name or by chunk object.

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk("normalize_input")
async def normalize(data: TriggerFlowEventData):
  return str(data.value).strip()

@flow.chunk
async def greet(data: TriggerFlowEventData):
  return f"Hello, {data.value}"

flow.to("normalize_input").to(greet).end()

execution = flow.create_execution()
result = execution.start(" Agently ")
print(result)

when (subscriptions)

when() subscribes to signals and supports:

• single event: when("Plan.Read")
• runtime data: when({"runtime_data": "user"})
• multi-trigger aggregation: when({"event": ["A", "B"]}, mode="and")

Multi-trigger subscriptions generate an internal When-xxx event for the next chain.

TriggerFlowEventData

Each handler receives TriggerFlowEventData with:

• value
• event / trigger_event
• type / trigger_type
• runtime methods: set_runtime_data / get_runtime_data
• event methods: emit / async_emit
• runtime stream: put_into_stream / stop_stream

collect

collect() aggregates branches and emits Collect-xxx when all required branches complete.

runtime_stream

Use put_into_stream() and get_runtime_stream() for progress events independent of final results.

BluePrint

Use save_blue_print() / load_blue_print() to export and reuse flow structure.

From token output to real-time signals

Background: From Token Output to Real-Time Signals.