Errors

The exception hierarchy. See individual guides for where each is raised.

agentflow.errors

Exception hierarchy for agent-workflow-sdk.

All library exceptions derive from :class:AgentFlowError so callers can catch the whole family with one except clause.

AgentFlowError

Bases: Exception

Base class for every AgentFlow exception.

Source code in agentflow/errors.py

class AgentFlowError(Exception):
    """Base class for every AgentFlow exception."""

GraphError

Bases: AgentFlowError

A problem with graph structure or execution.

Source code in agentflow/errors.py

class GraphError(AgentFlowError):
    """A problem with graph structure or execution."""

CompilationError

Bases: GraphError

The graph failed validation at compile() time.

Raised for unknown edge targets, an unreachable node, a missing entry edge from START, a conditional-router mapping to an undefined node, etc.

Source code in agentflow/errors.py

class CompilationError(GraphError):
    """The graph failed validation at ``compile()`` time.

    Raised for unknown edge targets, an unreachable node, a missing entry
    edge from START, a conditional-router mapping to an undefined node, etc.
    """

NodeError

Bases: GraphError

A node raised during execution.

Wraps the original exception (as __cause__) and names the node so the last good checkpoint plus this message pinpoint where a run failed.

Source code in agentflow/errors.py

class NodeError(GraphError):
    """A node raised during execution.

    Wraps the original exception (as ``__cause__``) and names the node so the
    last good checkpoint plus this message pinpoint where a run failed.
    """

    def __init__(self, node: str, message: str):
        self.node = node
        super().__init__(f"node {node!r} failed: {message}")

RunTimeout

Bases: GraphError

A whole-run timeout fired before the graph reached END.

The last completed super-step's checkpoint is intact (each step is checkpointed before the next begins), so the run is resumable from there when a checkpointer is configured.

Source code in agentflow/errors.py

class RunTimeout(GraphError):
    """A whole-run timeout fired before the graph reached END.

    The last completed super-step's checkpoint is intact (each step is
    checkpointed before the next begins), so the run is resumable from there
    when a checkpointer is configured.
    """

    def __init__(self, thread: str, seconds: float, step: int):
        self.thread = thread
        self.seconds = seconds
        self.step = step
        super().__init__(
            f"run {thread!r} exceeded {seconds}s (reached step {step}); last checkpoint preserved"
        )

InterruptError

Bases: AgentFlowError

Internal control-flow signal that a node requested a suspend.

The runtime catches this to suspend the run and persist an interrupted checkpoint; it is not meant to escape to user code. Callers observe the suspend through the run's stream / return value, not this exception.

wake_at distinguishes the two kinds of suspend: None is a human interrupt (waits for an explicit resume value), while an ISO-8601 time is a timer wait (ctx.wait) that the clock resolves.

Source code in agentflow/errors.py

class InterruptError(AgentFlowError):
    """Internal control-flow signal that a node requested a suspend.

    The runtime catches this to suspend the run and persist an interrupted
    checkpoint; it is not meant to escape to user code. Callers observe the
    suspend through the run's stream / return value, not this exception.

    ``wake_at`` distinguishes the two kinds of suspend: ``None`` is a human
    interrupt (waits for an explicit resume value), while an ISO-8601 time is a
    timer wait (``ctx.wait``) that the clock resolves.
    """

    def __init__(self, node: str, payload: object, *, wake_at: str | None = None):
        self.node = node
        self.payload = payload
        self.wake_at = wake_at
        super().__init__(f"interrupt requested by node {node!r}")

BackendError

Bases: AgentFlowError

A recoverable backend/agent error.

Note this is distinct from the :class:~agentflow.events.BackendError event, which is data a node may branch on. This exception is raised for backend misuse (e.g. prompting before start()).

Source code in agentflow/errors.py

class BackendError(AgentFlowError):
    """A recoverable backend/agent error.

    Note this is distinct from the :class:`~agentflow.events.BackendError`
    *event*, which is data a node may branch on. This exception is raised for
    backend misuse (e.g. prompting before ``start()``).
    """

BackendTransportError

Bases: BackendError

A fatal transport failure: the process died, the socket dropped, or the wire produced unparseable data. Not recoverable within the turn.

status and headers are populated when the failure came from an HTTP response, so callers can branch on them (e.g. distinguish a 429 from a 500).

Source code in agentflow/errors.py

class BackendTransportError(BackendError):
    """A fatal transport failure: the process died, the socket dropped, or
    the wire produced unparseable data. Not recoverable within the turn.

    ``status`` and ``headers`` are populated when the failure came from an HTTP
    response, so callers can branch on them (e.g. distinguish a 429 from a 500).
    """

    def __init__(
        self,
        message: str,
        *,
        status: int | None = None,
        headers: dict[str, str] | None = None,
        retry_after: float | None = None,
    ) -> None:
        self.status = status
        self.headers = headers or {}
        self.retry_after = retry_after
        super().__init__(message)

BackendRateLimitError

Bases: BackendTransportError

A provider rate limit (HTTP 429) that outlived the retry policy.

Carries retry_after (seconds) when the provider supplied it, so a caller can back off at the application level.

Source code in agentflow/errors.py

class BackendRateLimitError(BackendTransportError):
    """A provider rate limit (HTTP 429) that outlived the retry policy.

    Carries ``retry_after`` (seconds) when the provider supplied it, so a
    caller can back off at the application level."""

CheckpointError

Bases: AgentFlowError

A checkpoint could not be written, read, or resumed.

Source code in agentflow/errors.py

class CheckpointError(AgentFlowError):
    """A checkpoint could not be written, read, or resumed."""

CheckpointConflict

Bases: CheckpointError

A conditional checkpoint write failed because the stored revision did not match the expected one — another writer updated the same (thread, step) concurrently. Re-read and retry (optimistic concurrency).

Source code in agentflow/errors.py

class CheckpointConflict(CheckpointError):
    """A conditional checkpoint write failed because the stored revision did
    not match the expected one — another writer updated the same (thread, step)
    concurrently. Re-read and retry (optimistic concurrency)."""

    def __init__(self, thread: str, step: int, expected: int, actual: int):
        self.thread = thread
        self.step = step
        self.expected = expected
        self.actual = actual
        super().__init__(
            f"checkpoint conflict on {thread!r} step {step}: "
            f"expected revision {expected}, found {actual}"
        )

StoreError

Bases: AgentFlowError

A store operation could not be completed.

Source code in agentflow/errors.py

class StoreError(AgentFlowError):
    """A store operation could not be completed."""

StoreConflict

Bases: StoreError

A conditional store write (put(..., if_absent=True)) failed because a live item already exists at (namespace, key) — another writer claimed it first. The loser re-reads and reacts (e.g. treats the effect as already in-flight or done). This is what makes a claim an atomic single-flight.

Source code in agentflow/errors.py

class StoreConflict(StoreError):
    """A conditional store write (``put(..., if_absent=True)``) failed because a
    live item already exists at ``(namespace, key)`` — another writer claimed it
    first. The loser re-reads and reacts (e.g. treats the effect as already
    in-flight or done). This is what makes a claim an atomic single-flight."""

    def __init__(self, namespace: tuple[str, ...], key: str):
        self.namespace = namespace
        self.key = key
        super().__init__(f"store conflict: {key!r} under {namespace} already exists")

ControlPlaneError

Bases: AgentFlowError

A control-plane (run queue / worker) operation failed.

Source code in agentflow/errors.py

class ControlPlaneError(AgentFlowError):
    """A control-plane (run queue / worker) operation failed."""

RunNotFound

Bases: ControlPlaneError

No run exists for the given id.

Source code in agentflow/errors.py

class RunNotFound(ControlPlaneError):
    """No run exists for the given id."""

    def __init__(self, run_id: str):
        self.run_id = run_id
        super().__init__(f"no run with id {run_id!r}")

Effect idempotency

agentflow.prebuilt.effects.IncompleteEffectError

Bases: AgentFlowError

A prior attempt at an effect died between claim and commit.

Raised by :meth:IdempotentOp.run when it finds an in_flight marker and on_incomplete="error" — the safe default for a non-idempotent effect, where blindly re-running could duplicate it. The run suspends the decision to a human or a higher-level policy rather than guessing. key and namespace identify the stuck effect.

Source code in agentflow/prebuilt/effects.py

class IncompleteEffectError(AgentFlowError):
    """A prior attempt at an effect died between claim and commit.

    Raised by :meth:`IdempotentOp.run` when it finds an ``in_flight`` marker and
    ``on_incomplete="error"`` — the safe default for a non-idempotent effect,
    where blindly re-running could duplicate it. The run suspends the decision
    to a human or a higher-level policy rather than guessing. ``key`` and
    ``namespace`` identify the stuck effect.
    """

    def __init__(self, namespace: Namespace, key: str):
        self.namespace = namespace
        self.key = key
        super().__init__(
            f"effect {key!r} under {namespace} was left in-flight by a prior attempt; "
            f"cannot tell if it completed. Choose on_incomplete='rerun' if the effect "
            f"is idempotent, or resolve it by hand."
        )