Skip to content

Retry Utilities

Retry decorators and utilities.

Quick Example

from mamba_agents.errors import create_retry_decorator

@create_retry_decorator(max_attempts=3, base_wait=1.0)
async def call_api():
    response = await httpx.get(url)
    return response.json()

Retry Levels

Level Max Retries Base Wait Description
1 2 1.0s Conservative
2 3 1.0s Balanced (default)
3 5 0.5s Aggressive

Configuration

from mamba_agents.config import ErrorRecoveryConfig

config = ErrorRecoveryConfig(
    retry_level=2,
    max_retries=3,
    base_wait=1.0,
    max_wait=30.0,
    exponential_base=2.0,
    jitter=True,
)

API Reference

create_retry_decorator 

create_retry_decorator(
    config: ErrorRecoveryConfig | None = None,
    *,
    max_attempts: int | None = None,
    base_wait: float | None = None,
    max_wait: float | None = None,
    retry_exceptions: tuple[type[Exception], ...]
    | None = None,
) -> Callable[[Callable[..., T]], Callable[..., T]]

Create a retry decorator with the given configuration.

PARAMETER DESCRIPTION
config

Error recovery configuration.

TYPE: ErrorRecoveryConfig | None DEFAULT: None

max_attempts

Override max retry attempts.

TYPE: int | None DEFAULT: None

base_wait

Override base wait time in seconds.

TYPE: float | None DEFAULT: None

max_wait

Override max wait time in seconds.

TYPE: float | None DEFAULT: None

retry_exceptions

Override exceptions to retry on.

TYPE: tuple[type[Exception], ...] | None DEFAULT: None

RETURNS DESCRIPTION
Callable[[Callable[..., T]], Callable[..., T]]

Retry decorator.
Source code in src/mamba_agents/errors/retry.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def create_retry_decorator(
    config: ErrorRecoveryConfig | None = None,
    *,
    max_attempts: int | None = None,
    base_wait: float | None = None,
    max_wait: float | None = None,
    retry_exceptions: tuple[type[Exception], ...] | None = None,
) -> Callable[[Callable[..., T]], Callable[..., T]]:
    """Create a retry decorator with the given configuration.

    Args:
        config: Error recovery configuration.
        max_attempts: Override max retry attempts.
        base_wait: Override base wait time in seconds.
        max_wait: Override max wait time in seconds.
        retry_exceptions: Override exceptions to retry on.

    Returns:
        Retry decorator.
    """
    if config is None:
        config = ErrorRecoveryConfig()

    # Use overrides or config values
    attempts = max_attempts if max_attempts is not None else config.max_retries + 1
    wait_base = base_wait if base_wait is not None else config.base_retry_delay
    wait_max = max_wait if max_wait is not None else config.max_retry_delay

    # Default retry exceptions if not specified
    if retry_exceptions is None:
        retry_exceptions = (RateLimitError, ConnectionError, TimeoutError)

    return retry(
        stop=stop_after_attempt(attempts),
        wait=wait_exponential(multiplier=wait_base, max=wait_max),
        retry=retry_if_exception_type(retry_exceptions),
        before_sleep=_log_retry_attempt,
        reraise=True,
    )