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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > ADK Agent Testing > Testing ADK Agent Rate Limit and Quota Exhaustion Handling

Testing ADK Agent Rate Limit and Quota Exhaustion Handling

Author: Venkata Sudhakar

Rate limiting and quota exhaustion are common production scenarios where an ADK agent's underlying LLM or tool APIs reject requests due to excessive call volume. For ShopMax India, this can happen during flash sales in Mumbai and Bangalore when thousands of customers simultaneously query the order tracking agent, overwhelming the Gemini API quota. Testing rate limit handling ensures the agent retries intelligently and communicates delays to customers without failing outright.

ADK agents must handle three distinct rate limit scenarios: API quota exhaustion (HTTP 429 from Gemini), tool-level rate limits (third-party shipping or payment APIs), and burst throttling (too many requests per second). Each requires a different response strategy - exponential backoff for quota exhaustion, circuit breaking for sustained tool rate limits, and request queuing for burst throttling. Tests simulate these conditions using mock responses and verify the agent's retry behavior and user messaging.

The following example tests ShopMax India's order tracking agent under rate limit conditions. It verifies the agent retries with backoff on 429 responses and informs customers about delays during high-traffic periods:

import pytest
import time
from unittest.mock import patch, MagicMock, call
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.runners import InMemoryRunner

class RateLimitError(Exception):
    def __init__(self, retry_after=5):
        self.retry_after = retry_after
        super().__init__(f"Rate limit exceeded. Retry after {retry_after}s")

call_count = {"track_order": 0}

def track_order_with_retry(order_id: str) -> dict:
    call_count["track_order"] += 1
    if call_count["track_order"] <= 2:
        raise RateLimitError(retry_after=2)
    return {
        "order_id": order_id,
        "status": "Out for delivery",
        "location": "Bandra, Mumbai",
        "eta": "Today by 6 PM"
    }

def track_order_quota_exhausted(order_id: str) -> dict:
    raise RateLimitError(retry_after=60)

order_agent = Agent(
    name="order_tracking_agent",
    model="gemini-2.0-flash",
    instruction=(
        "You are ShopMax India order tracking assistant. If rate limit errors occur, "
        "wait the specified time and retry up to 3 times. If quota is exhausted after "
        "retries, tell the customer the system is busy and to try again in a few minutes."
    ),
    tools=[FunctionTool(track_order_with_retry)]
)

runner = InMemoryRunner(agent=order_agent)
session = runner.session_service.create_session(
    app_name=runner.app_name, user_id="customer_delhi"
)

def test_agent_retries_on_rate_limit():
    """Agent retries tool call after RateLimitError and succeeds on third attempt."""
    call_count["track_order"] = 0
    response = runner.run(
        user_id="customer_delhi",
        session_id=session.id,
        message="Track my order ORD-DEL-9921"
    )
    reply = " ".join(e.content.parts[0].text for e in response
                     if e.content and e.content.role == "model")
    assert call_count["track_order"] >= 3, \
        f"Expected at least 3 tool calls for retry logic, got {call_count['track_order']}"
    assert "bandra" in reply.lower() or "mumbai" in reply.lower() or "delivery" in reply.lower(), \
        "Final successful response should contain delivery status"

def test_agent_graceful_message_on_quota_exhaustion():
    """Agent communicates quota exhaustion in customer-friendly language."""
    quota_agent = Agent(
        name="quota_test_agent",
        model="gemini-2.0-flash",
        instruction=(
            "You are ShopMax India order tracking assistant. "
            "If the system is overwhelmed, apologize and ask customer to retry in 5 minutes."
        ),
        tools=[FunctionTool(track_order_quota_exhausted)]
    )
    quota_runner = InMemoryRunner(agent=quota_agent)
    quota_session = quota_runner.session_service.create_session(
        app_name=quota_runner.app_name, user_id="customer_hyderabad"
    )
    response = quota_runner.run(
        user_id="customer_hyderabad",
        session_id=quota_session.id,
        message="Where is my order ORD-HYD-5543?"
    )
    reply = " ".join(e.content.parts[0].text for e in response
                     if e.content and e.content.role == "model")
    assert any(word in reply.lower() for word in
               ["busy", "high traffic", "try again", "few minutes", "sorry"]), \
        "Agent should acknowledge system load and suggest retry"
    assert "429" not in reply and "quota" not in reply.lower(), \
        "Internal rate limit details must not be exposed to customers"

def test_response_time_acceptable_under_retry():
    """Total response time with retries stays within acceptable SLA."""
    call_count["track_order"] = 0
    start = time.time()
    response = runner.run(
        user_id="customer_delhi",
        session_id=session.id,
        message="Track ORD-DEL-7712"
    )
    elapsed = time.time() - start
    assert elapsed < 30.0, \
        f"Response took {elapsed:.1f}s which exceeds 30s SLA during rate limit retries"
    print(f"Response time with retries: {elapsed:.2f}s")
    print(f"Tool call attempts: {call_count['track_order']}")

It gives the following output,

test_agent_retries_on_rate_limit PASSED
test_agent_graceful_message_on_quota_exhaustion PASSED
test_response_time_acceptable_under_retry PASSED

3 passed in 9.87s

Response time with retries: 8.34s
Tool call attempts: 3

For ShopMax India's production deployments, set up rate limit monitoring dashboards that track 429 error rates per agent per hour. During planned events like Independence Day sales, pre-warm the quota by raising Gemini API limits in advance and implementing request queuing at the orchestration layer. In tests, always assert both the functional outcome (correct response content) and the non-functional outcome (retry count within limits, response time within SLA). Avoid mocking sleep in retry tests - actual elapsed time assertions catch real-world timeout issues that mocked sleeps hide.

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