when Branches

when() can wait for chunk completion, runtime_data changes, or custom events. It is ideal for event-driven branching.

Scenario: wait for a chunk to finish

Use this for staged workflows where downstream tasks start only after upstream completion.

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
async def parse(data: TriggerFlowEventData):
  return f"parsed:{data.value}"

@flow.chunk
async def build(data: TriggerFlowEventData):
  return f"built:{data.value}"

flow.to(parse)
flow.when(parse).to(build).end()

print(flow.start("doc"))

Output:

built:parsed:doc

Scenario: wait for a runtime_data change

Use this when you need to proceed only after state is ready (for example, user info is loaded).

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
def print_user(data: TriggerFlowEventData):
  print(f"user:{data.value['id']}")

flow.when({"runtime_data": "user"}).to(print_user)

execution = flow.create_execution()
execution.set_runtime_data("user", {"id": "u-1"})

Output:

user:u-1

Scenario: wait for a custom event (emit)

Use this for external signals (message buses, webhooks, user actions).

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
def print_join(data: TriggerFlowEventData):
  print(f"join:{data.value['id']}")

flow.when("User.Join").to(print_join)

execution = flow.create_execution()
execution.emit("User.Join", {"id": "u-9"})

Output:

join:u-9

when modes: and / or / simple_or

when() can listen to multiple signals. The mode defines the triggering logic:

• and: wait for all signals, output a merged dict
• or: trigger on any signal, output (type, event, value)
• simple_or: trigger on any signal, output only value

mode="and"

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
def print_value(data: TriggerFlowEventData):
  print(data.value)

flow.when({"event": ["A", "B"]}, mode="and").to(print_value)

execution = flow.create_execution()
execution.emit("A", 1)
execution.emit("B", 2)

Output:

{'event': {'A': 1, 'B': 2}}

mode="or"

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
def print_value(data: TriggerFlowEventData):
  print(data.value)

flow.when({"event": ["A", "B"]}, mode="or").to(print_value)

execution = flow.create_execution()
execution.emit("A", 1)
execution.emit("B", 2)

Output:

('event', 'A', 1)
('event', 'B', 2)

mode="simple_or"

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
def print_value(data: TriggerFlowEventData):
  print(data.value)

flow.when({"event": ["A", "B"]}, mode="simple_or").to(print_value)

execution = flow.create_execution()
execution.emit("A", 1)
execution.emit("B", 2)

Output:

1
2