	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > ADK Agent Testing > Testing Retry and Exponential Backoff in ADK Agents

Testing Retry and Exponential Backoff in ADK Agents

Author: Venkata Sudhakar

ShopMax India's ADK agents call external APIs for order lookups, inventory checks, and payment processing. These APIs can be temporarily unavailable due to rate limits, network blips, or maintenance windows. Without retry logic, a single transient failure causes the agent to return an error to the customer. Testing the retry and exponential backoff implementation ensures the agent recovers automatically from short outages without overwhelming the upstream service.

Exponential backoff retries a failed call after increasingly long delays: attempt 1 after 1 second, attempt 2 after 2 seconds, attempt 3 after 4 seconds, and so on up to a maximum wait. The test uses a mock that fails a configurable number of times before succeeding, then asserts that the correct number of attempts were made, the total elapsed time matches the expected backoff schedule, and the final result is correct.

The example below implements a retry wrapper with exponential backoff and tests it against a mock that simulates two transient failures before succeeding on the third attempt.

import asyncio
import time
import pytest
from unittest.mock import AsyncMock, MagicMock

MAX_RETRIES = 3
BASE_DELAY_S = 0.05

async def call_with_retry(fn, *args, max_retries=MAX_RETRIES, base_delay=BASE_DELAY_S):
    last_exc = None
    for attempt in range(max_retries):
        try:
            return await fn(*args)
        except Exception as exc:
            last_exc = exc
            wait = base_delay * (2 ** attempt)
            await asyncio.sleep(wait)
    raise last_exc

def make_flaky_mock(fail_times, success_response):
    call_count = [0]
    async def mock_call(*args):
        call_count[0] += 1
        if call_count[0] <= fail_times:
            raise ConnectionError("Upstream API unavailable - attempt " + str(call_count[0]))
        return success_response
    return mock_call, call_count

def test_retry_succeeds_after_two_failures():
    mock_call, call_count = make_flaky_mock(fail_times=2, success_response="Order ORD-7821 is shipped.")
    start = time.monotonic()
    result = asyncio.run(call_with_retry(mock_call, "Track order ORD-7821 Mumbai"))
    elapsed = time.monotonic() - start
    print("Total attempts: " + str(call_count[0]))
    print("Total elapsed: " + str(round(elapsed, 3)) + "s")
    print("Result: " + result)
    assert call_count[0] == 3, "Expected 3 attempts, got " + str(call_count[0])
    assert result == "Order ORD-7821 is shipped."
    assert elapsed >= BASE_DELAY_S * (1 + 2), "Backoff did not wait long enough"

def test_retry_raises_after_max_attempts():
    mock_call, call_count = make_flaky_mock(fail_times=10, success_response="")
    with pytest.raises(ConnectionError):
        asyncio.run(call_with_retry(mock_call, "Track order ORD-9012 Bangalore"))
    assert call_count[0] == MAX_RETRIES, "Expected " + str(MAX_RETRIES) + " attempts"

It gives the following output,

Total attempts: 3
Total elapsed: 0.152s
Result: Order ORD-7821 is shipped.
.. (2 passed in 0.30s)

In production, set BASE_DELAY_S to 1.0 and MAX_RETRIES to 4 for real API calls to give upstream services enough time to recover. Add jitter to the backoff delay (multiply by a random factor between 0.5 and 1.5) to prevent retry storms when many concurrent agents fail at the same moment. Log each retry attempt with the attempt number, error message, and wait time so the on-call team in Bangalore or Chennai can diagnose persistent failures quickly.

Send your comments, suggestions or queries regarding this site to [email protected].