Retry API

Auto-retry for transient model and gateway failures. Each retry resumes from the accumulated message history rather than starting over. See Retries.

RetryConfig

subagents_pydantic_ai.RetryConfig dataclass

Resolved retry policy for a subagent run.

Attributes:

Name	Type	Description
max_retries	int	Number of additional attempts after the first failure. Defaults to 3 so subagents are resilient to flaky model gateways/networks out of the box. Set 0 to disable retrying entirely (the legacy agent.run() opt-out path).
initial_delay	float	Seconds to wait before the first retry.
max_delay	float	Upper bound for the backoff delay, in seconds.
backoff_multiplier	float	The delay is multiplied by this each attempt.
jitter	bool	When True, the delay is randomised in [0, computed_delay] (full jitter) to avoid a thundering herd across many concurrent subagents.
retry_on	RetryPredicate | None	Predicate deciding whether an exception is transient. None uses :func:is_transient_error.

Source code in src/subagents_pydantic_ai/retry.py

@dataclass(frozen=True)
class RetryConfig:
    """Resolved retry policy for a subagent run.

    Attributes:
        max_retries: Number of *additional* attempts after the first
            failure. Defaults to `3` so subagents are resilient to
            flaky model gateways/networks out of the box. Set `0` to
            disable retrying entirely (the legacy `agent.run()`
            opt-out path).
        initial_delay: Seconds to wait before the first retry.
        max_delay: Upper bound for the backoff delay, in seconds.
        backoff_multiplier: The delay is multiplied by this each attempt.
        jitter: When `True`, the delay is randomised in
            `[0, computed_delay]` (full jitter) to avoid a thundering
            herd across many concurrent subagents.
        retry_on: Predicate deciding whether an exception is transient.
            `None` uses :func:`is_transient_error`.
    """

    max_retries: int = 3
    initial_delay: float = 1.0
    max_delay: float = 30.0
    backoff_multiplier: float = 2.0
    jitter: bool = True
    retry_on: RetryPredicate | None = None

    @classmethod
    def from_config(cls, config: SubAgentConfig) -> RetryConfig:
        """Build a :class:`RetryConfig` from a :class:`SubAgentConfig`.

        Missing keys fall back to the dataclass defaults, so a config
        without any `retry_*` keys yields the default policy (3
        retries with exponential backoff).
        """
        return cls(
            max_retries=config.get("max_retries", 3),
            initial_delay=config.get("retry_initial_delay", 1.0),
            max_delay=config.get("retry_max_delay", 30.0),
            backoff_multiplier=config.get("retry_backoff_multiplier", 2.0),
            jitter=config.get("retry_jitter", True),
            retry_on=config.get("retry_on"),
        )

    def should_retry(self, exc: BaseException) -> bool:
        """Return whether *exc* is retryable under this policy."""
        predicate = self.retry_on or is_transient_error
        return predicate(exc)

from_config(config) classmethod

Build a :class:RetryConfig from a :class:SubAgentConfig.

Missing keys fall back to the dataclass defaults, so a config without any retry_* keys yields the default policy (3 retries with exponential backoff).

Source code in src/subagents_pydantic_ai/retry.py

@classmethod
def from_config(cls, config: SubAgentConfig) -> RetryConfig:
    """Build a :class:`RetryConfig` from a :class:`SubAgentConfig`.

    Missing keys fall back to the dataclass defaults, so a config
    without any `retry_*` keys yields the default policy (3
    retries with exponential backoff).
    """
    return cls(
        max_retries=config.get("max_retries", 3),
        initial_delay=config.get("retry_initial_delay", 1.0),
        max_delay=config.get("retry_max_delay", 30.0),
        backoff_multiplier=config.get("retry_backoff_multiplier", 2.0),
        jitter=config.get("retry_jitter", True),
        retry_on=config.get("retry_on"),
    )

should_retry(exc)

Return whether exc is retryable under this policy.

Source code in src/subagents_pydantic_ai/retry.py

def should_retry(self, exc: BaseException) -> bool:
    """Return whether *exc* is retryable under this policy."""
    predicate = self.retry_on or is_transient_error
    return predicate(exc)

run_with_retry

subagents_pydantic_ai.run_with_retry(agent, user_prompt, *, run_kwargs, retry, on_retry=None, sleep=asyncio.sleep, event_stream_handler=None, cancel_check=None, inject_messages=None) async

Run agent with auto-retry on transient errors.

When retry.max_retries <= 0 this is exactly agent.run(...) — the legacy path, unchanged. Otherwise the agent is driven via agent.iter() so that, on a transient failure, the accumulated message history from the failed attempt is replayed via message_history on the next attempt and the subagent resumes instead of restarting from scratch.

Parameters:

Name	Type	Description	Default
agent	Any	The pydantic-ai Agent to run.	required
user_prompt	str | None	Initial prompt. After a retry that captured history it is set to None because the prompt is already replayed inside message_history.	required
run_kwargs	dict[str, Any]	Extra kwargs forwarded to agent.run/agent.iter (deps, toolsets, ...). A caller-supplied message_history is honoured as the starting history, and a caller-supplied usage as the tally every attempt adds to.	required
retry	RetryConfig	The resolved retry policy.	required
on_retry	OnRetryCallback | None	Optional callback invoked before each retry sleep with (attempt, exc, delay). May be sync or async.	None
sleep	Callable[[float], Awaitable[None]]	Async sleep function, injectable for tests.	sleep
event_stream_handler	Any | None	Optional override for the agent's configured event_stream_handler. When None the agent's own handler (agent.event_stream_handler) is used, so streaming to a platform (e.g. tool-call/reasoning events to Kafka) keeps working across retries — matching agent.run() semantics.	None
cancel_check	Callable[[], bool] | None	Optional callable polled between graph nodes for cooperative (soft) cancellation. When it returns True the run stops at the next node boundary by raising asyncio.CancelledError. Only honoured on the retry-driven path (max_retries > 0); the legacy agent.run() fast path (max_retries <= 0) does not expose node boundaries, so soft cancel is best-effort there.	None
inject_messages	Callable[[], Awaitable[list[str]]] | None	Optional async callable awaited before each model request; its returned strings are appended to that request as user instructions (unprompted parent -> child steering). Like cancel_check, only honoured on the retry-driven path (max_retries > 0); the legacy agent.run() fast path does not expose node boundaries, so steering messages stay queued there.	None

Returns:

Type	Description
Any	The AgentRunResult of the first successful attempt.

Raises:

Type	Description
Exception	The last exception when retries are exhausted or the error is not transient. asyncio.CancelledError is a BaseException and is never caught here, so cooperative/hard task cancellation propagates unchanged.

Source code in src/subagents_pydantic_ai/retry.py

async def run_with_retry(
    agent: Any,
    user_prompt: str | None,
    *,
    run_kwargs: dict[str, Any],
    retry: RetryConfig,
    on_retry: OnRetryCallback | None = None,
    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
    event_stream_handler: Any | None = None,
    cancel_check: Callable[[], bool] | None = None,
    inject_messages: Callable[[], Awaitable[list[str]]] | None = None,
) -> Any:
    """Run *agent* with auto-retry on transient errors.

    When `retry.max_retries <= 0` this is exactly `agent.run(...)` —
    the legacy path, unchanged. Otherwise the agent is driven via
    `agent.iter()` so that, on a transient failure, the accumulated
    message history from the failed attempt is replayed via
    `message_history` on the next attempt and the subagent resumes
    instead of restarting from scratch.

    Args:
        agent: The pydantic-ai `Agent` to run.
        user_prompt: Initial prompt. After a retry that captured history
            it is set to `None` because the prompt is already replayed
            inside `message_history`.
        run_kwargs: Extra kwargs forwarded to `agent.run`/`agent.iter`
            (`deps`, `toolsets`, ...). A caller-supplied
            `message_history` is honoured as the starting history, and a
            caller-supplied `usage` as the tally every attempt adds to.
        retry: The resolved retry policy.
        on_retry: Optional callback invoked before each retry sleep with
            `(attempt, exc, delay)`. May be sync or async.
        sleep: Async sleep function, injectable for tests.
        event_stream_handler: Optional override for the agent's configured
            `event_stream_handler`. When `None` the agent's own handler
            (`agent.event_stream_handler`) is used, so streaming to a
            platform (e.g. tool-call/reasoning events to Kafka) keeps working
            across retries — matching `agent.run()` semantics.
        cancel_check: Optional callable polled between graph nodes for
            cooperative (soft) cancellation. When it returns `True` the run
            stops at the next node boundary by raising
            `asyncio.CancelledError`. Only honoured on the retry-driven path
            (`max_retries > 0`); the legacy `agent.run()` fast path
            (`max_retries <= 0`) does not expose node boundaries, so soft
            cancel is best-effort there.
        inject_messages: Optional async callable awaited before each model
            request; its returned strings are appended to that request as
            user instructions (unprompted parent -> child steering). Like
            `cancel_check`, only honoured on the retry-driven path
            (`max_retries > 0`); the legacy `agent.run()` fast path does not
            expose node boundaries, so steering messages stay queued there.

    Returns:
        The `AgentRunResult` of the first successful attempt.

    Raises:
        Exception: The last exception when retries are exhausted or the error
            is not transient. `asyncio.CancelledError` is a `BaseException`
            and is never caught here, so cooperative/hard task cancellation
            propagates unchanged.
    """
    # An explicit handler overrides the agent's; otherwise inherit the agent's
    # own, exactly as agent.run() does (event_stream_handler or self.…).
    handler = event_stream_handler or getattr(agent, "event_stream_handler", None)

    if retry.max_retries <= 0:
        # Fast path: agent.run() already drives streaming and honours the
        # agent's handler. Only forward an explicit override.
        if event_stream_handler is not None:
            run_kwargs = {**run_kwargs, "event_stream_handler": event_stream_handler}
        return await agent.run(user_prompt, **run_kwargs)

    message_history = run_kwargs.pop("message_history", None)
    # One tally across every attempt. `Agent.iter` builds a fresh `RunUsage` when
    # `usage` is `None`, so without this each attempt starts counting from zero
    # while the replayed history genuinely re-spends the tokens -- and a
    # `usage_limits` ceiling would be granted again per attempt, multiplying the
    # caller's budget by `max_retries + 1`.
    caller_usage: RunUsage | None = run_kwargs.pop("usage", None)
    run_usage = caller_usage if caller_usage is not None else RunUsage()
    prompt = user_prompt
    attempt = 0
    while True:
        run = None
        try:
            async with agent.iter(
                prompt, message_history=message_history, usage=run_usage, **run_kwargs
            ) as run:
                await _drive_run(agent, run, handler, cancel_check, inject_messages)
            return run.result
        except Exception as exc:
            if attempt >= retry.max_retries or not retry.should_retry(exc):
                raise
            attempt += 1
            # Resume from wherever the failed attempt got to. `run` is
            # None only if `agent.iter()` failed before yielding.
            if run is not None:
                accumulated = run.all_messages()
                if accumulated:
                    message_history = accumulated
                    prompt = None
            delay = compute_backoff_delay(attempt, retry)
            if on_retry is not None:
                maybe_coro = on_retry(attempt, exc, delay)
                if asyncio.iscoroutine(maybe_coro):
                    await maybe_coro
            await sleep(delay)

is_transient_error

subagents_pydantic_ai.is_transient_error(exc)

Return True if exc looks like a transient networking failure.

Treated as transient (worth retrying):

• ModelHTTPError with a 408/429/5xx status code — gateway hiccups, rate limits or upstream overload, typical with proxies such as LiteLLM.
• ModelAPIError that is not an HTTP error — connection resets, read timeouts and other transport-level problems surfaced by the model client.

Everything else (auth/4xx, UnexpectedModelBehavior, UsageLimitExceeded, UserError, validation errors, task cancellation, ...) is treated as non-transient and is not retried.

Source code in src/subagents_pydantic_ai/retry.py

def is_transient_error(exc: BaseException) -> bool:
    """Return `True` if *exc* looks like a transient networking failure.

    Treated as transient (worth retrying):

    - `ModelHTTPError` with a 408/429/5xx status code — gateway hiccups,
      rate limits or upstream overload, typical with proxies such as
      LiteLLM.
    - `ModelAPIError` that is *not* an HTTP error — connection resets,
      read timeouts and other transport-level problems surfaced by the
      model client.

    Everything else (auth/4xx, `UnexpectedModelBehavior`,
    `UsageLimitExceeded`, `UserError`, validation errors, task
    cancellation, ...) is treated as non-transient and is not retried.
    """
    if isinstance(exc, ModelHTTPError):
        return exc.status_code in _TRANSIENT_STATUS_CODES
    # A bare ModelAPIError (no HTTP status) is a transport/connection
    # error from the model client — safe to retry.
    return isinstance(exc, ModelAPIError)

compute_backoff_delay

subagents_pydantic_ai.compute_backoff_delay(attempt, cfg, rng=random.uniform)

Compute the delay (seconds) before retry attempt (1-based).

Exponential backoff (initial_delay * multiplier ** (attempt - 1)) capped at cfg.max_delay. With cfg.jitter the result is randomised in [0, delay] (full jitter). rng is injectable for deterministic tests.

Source code in src/subagents_pydantic_ai/retry.py

def compute_backoff_delay(
    attempt: int,
    cfg: RetryConfig,
    rng: Callable[[float, float], float] = random.uniform,
) -> float:
    """Compute the delay (seconds) before retry *attempt* (1-based).

    Exponential backoff (`initial_delay * multiplier ** (attempt - 1)`)
    capped at `cfg.max_delay`. With `cfg.jitter` the result is
    randomised in `[0, delay]` (full jitter). `rng` is injectable for
    deterministic tests.
    """
    base = cfg.initial_delay * (cfg.backoff_multiplier ** (attempt - 1))
    delay = min(base, cfg.max_delay)
    if cfg.jitter:
        delay = rng(0.0, delay)
    return delay