Skip to content

Runtime and context

The per-node Context and the StreamEvent yielded by stream(). See The Context object.

agentflow.runtime.Context dataclass

Per-node handle: identity, event emission, and interrupts.

Source code in agentflow/runtime.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@dataclass
class Context:
    """Per-node handle: identity, event emission, and interrupts."""

    node: str
    thread: str
    step: int
    _emit_queue: asyncio.Queue[StreamEvent] | None = field(default=None, repr=False)
    _resume: _Resume | None = field(default=None, repr=False)
    _hooks: Hooks | None = field(default=None, repr=False)

    def emit(self, event: EmittableEvent) -> None:
        """Surface an event to the run's stream and telemetry hooks.

        Accepts any backend event (``TextChunk``, ``ToolCall``, ...) or a
        run-level :class:`~agentflow.events.GateEvent`.
        """
        if self._emit_queue is not None:
            self._emit_queue.put_nowait(
                StreamEvent(kind="backend", step=self.step, node=self.node, data=event)
            )
        if self._hooks is not None:
            # emit() is sync but hooks are async; schedule best-effort so a
            # telemetry sink can record backend events without blocking the
            # node. Fire-and-forget, exceptions swallowed by _safe.
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                return
            loop.create_task(_safe(self._hooks.on_event(self.thread, self.step, self.node, event)))

    async def interrupt(self, payload: Any = None) -> Any:
        """Suspend the run for a human, or return the supplied answer on resume.

        On a fresh run this raises :class:`InterruptError`; the runtime catches
        it, writes an interrupted checkpoint, and stops. When the run is later
        resumed with a value, that value is seeded here and returned directly.
        """
        if self._resume is not None and not self._resume.used:
            self._resume.used = True
            return self._resume.value
        raise InterruptError(self.node, payload)

    async def wait(self, wake_at: str, payload: Any = None) -> Any:
        """Suspend the run until ``wake_at`` (an ISO-8601 time), then re-run.

        This is a timer-based sibling of :meth:`interrupt`: a fresh call
        suspends the run with a checkpoint marked to wake no earlier than
        ``wake_at``, and no human answer is needed — the clock resolves it.
        When the run resumes (the control plane re-claims it once the time
        passes, or an inline runner sleeps until then), this call returns and
        the node runs again, typically to poll an external system once more.

        ``payload`` is persisted on the checkpoint (e.g. a watcher cursor) so
        the node can pick up where it left off. On resume the seeded value is
        returned, mirroring :meth:`interrupt`; a watcher usually ignores it and
        reads its cursor from state instead.
        """
        if self._resume is not None and not self._resume.used:
            self._resume.used = True
            return self._resume.value
        raise InterruptError(self.node, payload, wake_at=wake_at)

emit 

emit(event: EmittableEvent) -> None

Surface an event to the run's stream and telemetry hooks.

Accepts any backend event (TextChunk, ToolCall, ...) or a run-level :class:~agentflow.events.GateEvent.

Source code in agentflow/runtime.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def emit(self, event: EmittableEvent) -> None:
    """Surface an event to the run's stream and telemetry hooks.

    Accepts any backend event (``TextChunk``, ``ToolCall``, ...) or a
    run-level :class:`~agentflow.events.GateEvent`.
    """
    if self._emit_queue is not None:
        self._emit_queue.put_nowait(
            StreamEvent(kind="backend", step=self.step, node=self.node, data=event)
        )
    if self._hooks is not None:
        # emit() is sync but hooks are async; schedule best-effort so a
        # telemetry sink can record backend events without blocking the
        # node. Fire-and-forget, exceptions swallowed by _safe.
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            return
        loop.create_task(_safe(self._hooks.on_event(self.thread, self.step, self.node, event)))

interrupt async 

interrupt(payload: Any = None) -> Any

Suspend the run for a human, or return the supplied answer on resume.

On a fresh run this raises :class:InterruptError; the runtime catches it, writes an interrupted checkpoint, and stops. When the run is later resumed with a value, that value is seeded here and returned directly.

Source code in agentflow/runtime.py 
122
123
124
125
126
127
128
129
130
131
132
async def interrupt(self, payload: Any = None) -> Any:
    """Suspend the run for a human, or return the supplied answer on resume.

    On a fresh run this raises :class:`InterruptError`; the runtime catches
    it, writes an interrupted checkpoint, and stops. When the run is later
    resumed with a value, that value is seeded here and returned directly.
    """
    if self._resume is not None and not self._resume.used:
        self._resume.used = True
        return self._resume.value
    raise InterruptError(self.node, payload)

wait async 

wait(wake_at: str, payload: Any = None) -> Any

Suspend the run until wake_at (an ISO-8601 time), then re-run.

This is a timer-based sibling of :meth:interrupt: a fresh call suspends the run with a checkpoint marked to wake no earlier than wake_at, and no human answer is needed — the clock resolves it. When the run resumes (the control plane re-claims it once the time passes, or an inline runner sleeps until then), this call returns and the node runs again, typically to poll an external system once more.

payload is persisted on the checkpoint (e.g. a watcher cursor) so the node can pick up where it left off. On resume the seeded value is returned, mirroring :meth:interrupt; a watcher usually ignores it and reads its cursor from state instead.

Source code in agentflow/runtime.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
async def wait(self, wake_at: str, payload: Any = None) -> Any:
    """Suspend the run until ``wake_at`` (an ISO-8601 time), then re-run.

    This is a timer-based sibling of :meth:`interrupt`: a fresh call
    suspends the run with a checkpoint marked to wake no earlier than
    ``wake_at``, and no human answer is needed — the clock resolves it.
    When the run resumes (the control plane re-claims it once the time
    passes, or an inline runner sleeps until then), this call returns and
    the node runs again, typically to poll an external system once more.

    ``payload`` is persisted on the checkpoint (e.g. a watcher cursor) so
    the node can pick up where it left off. On resume the seeded value is
    returned, mirroring :meth:`interrupt`; a watcher usually ignores it and
    reads its cursor from state instead.
    """
    if self._resume is not None and not self._resume.used:
        self._resume.used = True
        return self._resume.value
    raise InterruptError(self.node, payload, wake_at=wake_at)

agentflow.runtime.StreamEvent dataclass

One observable moment in a run, yielded by CompiledGraph.stream.

Source code in agentflow/runtime.py 
66
67
68
69
70
71
72
73
@dataclass(frozen=True, slots=True)
class StreamEvent:
    """One observable moment in a run, yielded by ``CompiledGraph.stream``."""

    kind: str  # "node_start"|"node_end"|"backend"|"step"|"interrupt"|"wait"|"done"
    step: int
    node: str | None = None
    data: Any = None