Events

The request, message, and backend-event vocabulary. See the Backends overview.

agentflow.events

Structured events, messages, and requests shared across backends.

Everything a backend emits is a typed, frozen dataclass. There is exactly one vocabulary for both agent backends (Kiro/Codex/Claude) and LLM backends (Ollama and OpenAI-compatible endpoints), so a node can move a tool call from one to a message for the other without adapters.

Nothing here is async and nothing imports the engine — these are pure data types.

Message dataclass

A neutral chat message.

Used as the input unit for :class:LLMBackend and to represent turns in an agent's history. tool_calls is set on an assistant message that requested tools; tool_call_id links a role="tool" result back to the call that produced it.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class Message:
    """A neutral chat message.

    Used as the input unit for :class:`LLMBackend` and to represent turns in
    an agent's history. ``tool_calls`` is set on an assistant message that
    requested tools; ``tool_call_id`` links a ``role="tool"`` result back to
    the call that produced it.
    """

    role: Role
    content: str = ""
    tool_calls: tuple[ToolCallSpec, ...] = ()
    tool_call_id: str | None = None
    name: str | None = None

ToolSpec dataclass

Describes a tool offered to an LLM (JSON-Schema parameters).

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class ToolSpec:
    """Describes a tool offered to an LLM (JSON-Schema parameters)."""

    name: str
    description: str = ""
    schema: dict[str, Any] = field(default_factory=dict)

ToolCallSpec dataclass

A single tool call an assistant message requested.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class ToolCallSpec:
    """A single tool call an assistant message requested."""

    id: str
    name: str
    args: dict[str, Any] = field(default_factory=dict)

TextRequest dataclass

Prompt an agent backend with free text.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class TextRequest:
    """Prompt an agent backend with free text."""

    text: str

ChatRequest dataclass

Call an LLM backend with a message list and optional tool schemas.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class ChatRequest:
    """Call an LLM backend with a message list and optional tool schemas."""

    messages: tuple[Message, ...]
    tools: tuple[ToolSpec, ...] = ()
    options: dict[str, Any] = field(default_factory=dict)

TextChunk dataclass

A streamed delta of the assistant's message text.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class TextChunk:
    """A streamed delta of the assistant's message text."""

    text: str

ToolCall dataclass

The backend/model invoked a tool. For an LLM backend this is a request the graph must fulfil; for an agent backend it is informational (the agent runs the tool itself).

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class ToolCall:
    """The backend/model invoked a tool. For an LLM backend this is a
    *request* the graph must fulfil; for an agent backend it is informational
    (the agent runs the tool itself)."""

    id: str
    name: str
    args: dict[str, Any] = field(default_factory=dict)
    title: str | None = None

ToolResult dataclass

A tool call resolved (emitted by agent backends that run their own tools).

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class ToolResult:
    """A tool call resolved (emitted by agent backends that run their own
    tools)."""

    id: str
    status: Literal["ok", "error", "pending"] = "ok"
    content: Any = None

PermissionRequest dataclass

An agent backend asks the client to authorize a tool use.

Resolved by the backend's :class:PermissionPolicy; it is not normally yielded to node code (agent backends handle it inline). It is a public type so an Interactive policy can surface it to a human.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class PermissionRequest:
    """An agent backend asks the client to authorize a tool use.

    Resolved by the backend's :class:`PermissionPolicy`; it is not normally
    yielded to node code (agent backends handle it inline). It is a public
    type so an ``Interactive`` policy can surface it to a human.
    """

    id: str
    tool: str
    options: tuple[PermissionOption, ...] = ()
    detail: dict[str, Any] = field(default_factory=dict)

PermissionOption dataclass

One choice a backend offers for a permission request.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class PermissionOption:
    """One choice a backend offers for a permission request."""

    option_id: str
    name: str
    kind: str  # e.g. "allow_always", "allow_once", "reject_once"

Usage dataclass

Token (and optional cost) usage reported by a backend for one turn.

All fields are optional because backends differ in what they report: some give input/output splits, some only a total, some none at all. A field left None means "not reported" — a token budget simply cannot count what a backend does not surface, and the budget helpers treat None as zero. cost_usd is only present when a backend or the caller supplies pricing; the SDK ships no price table (it would go stale), so budget on tokens by default and set cost_usd yourself if you want a dollar cap.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class Usage:
    """Token (and optional cost) usage reported by a backend for one turn.

    All fields are optional because backends differ in what they report: some
    give input/output splits, some only a total, some none at all. A field left
    ``None`` means "not reported" — a token budget simply cannot count what a
    backend does not surface, and the budget helpers treat ``None`` as zero.
    ``cost_usd`` is only present when a backend or the caller supplies pricing;
    the SDK ships no price table (it would go stale), so budget on tokens by
    default and set ``cost_usd`` yourself if you want a dollar cap.
    """

    input_tokens: int | None = None
    output_tokens: int | None = None
    total_tokens: int | None = None
    cost_usd: float | None = None

TurnEnd dataclass

Terminal event of a turn. Carries the fully assembled assistant text, the backend's stop reason (e.g. end_turn, max_tokens), and optional token/cost :class:Usage when the backend reports it.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class TurnEnd:
    """Terminal event of a turn. Carries the fully assembled assistant text,
    the backend's stop reason (e.g. ``end_turn``, ``max_tokens``), and optional
    token/cost :class:`Usage` when the backend reports it."""

    text: str = ""
    stop_reason: str | None = None
    message: Message | None = None
    usage: Usage | None = None

ErrorEvent dataclass

A recoverable error surfaced as data (not raised). A node may branch on it. Fatal transport failures raise :class:~agentflow.errors.BackendTransportError instead.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class ErrorEvent:
    """A recoverable error surfaced as data (not raised). A node may branch
    on it. Fatal transport failures raise
    :class:`~agentflow.errors.BackendTransportError` instead."""

    message: str
    detail: dict[str, Any] = field(default_factory=dict)

GateEvent dataclass

A quality-gate lifecycle event, surfaced via ctx.emit so telemetry and the run stream show where a process gated on quality.

phase is one of "started", "passed", "failed", or "escalated". gate names the gate; attempt is the 1-based try count; score and reasons mirror the evaluator's :class:~agentflow.prebuilt.GateResult. This is an observability event — it is emitted, never returned as a node update. See the :doc:quality gates guide </durability/quality-gates>.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class GateEvent:
    """A quality-gate lifecycle event, surfaced via ``ctx.emit`` so telemetry
    and the run stream show where a process gated on quality.

    ``phase`` is one of ``"started"``, ``"passed"``, ``"failed"``, or
    ``"escalated"``. ``gate`` names the gate; ``attempt`` is the 1-based try
    count; ``score`` and ``reasons`` mirror the evaluator's
    :class:`~agentflow.prebuilt.GateResult`. This is an observability event —
    it is emitted, never returned as a node update. See the
    :doc:`quality gates guide </durability/quality-gates>`.
    """

    gate: str
    phase: str  # "started" | "passed" | "failed" | "escalated"
    attempt: int = 0
    score: float | None = None
    reasons: tuple[str, ...] = ()

Allow dataclass

Approve a permission request. option_id selects a specific offered option; when None the backend picks the most permissive allow option it was given.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class Allow:
    """Approve a permission request. ``option_id`` selects a specific offered
    option; when ``None`` the backend picks the most permissive allow option
    it was given."""

    option_id: str | None = None
    remember: bool = False

Deny dataclass

Reject a permission request.

Source code in agentflow/events.py

@dataclass(frozen=True, slots=True)
class Deny:
    """Reject a permission request."""

    reason: str | None = None