How to Build Retry-Safe Agent Tool Calls

import os, requests, json, time
from functools import wraps

API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}

def retry_with_backoff(max_retries=3, base_delay=1.0, max_delay=10.0):
    """Decorator: retry failed calls with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.Timeout:
                    last_error = 'timeout'
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    print(f'  Retry {attempt+1}/{max_retries} after timeout ({delay:.1f}s)')
                    time.sleep(delay)
                except requests.exceptions.HTTPError as e:
                    status = e.response.status_code if e.response else 0
                    if status == 429:  # Rate limited
                        delay = min(base_delay * (2 ** attempt), max_delay)
                        print(f'  Retry {attempt+1}/{max_retries} after 429 ({delay:.1f}s)')
                        time.sleep(delay)
                    elif status >= 500:  # Server error
                        delay = min(base_delay * (2 ** attempt), max_delay)
                        print(f'  Retry {attempt+1}/{max_retries} after {status} ({delay:.1f}s)')
                        time.sleep(delay)
                    else:
                        raise  # Don't retry 4xx errors (except 429)
                except Exception as e:
                    last_error = str(e)
                    break
            return {'error': last_error or 'max_retries_exceeded', 'results': []}
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3)
def safe_search(query, **kwargs):
    resp = requests.post('https://api.scavio.dev/api/v1/search',
        headers=SH, json={'query': query, 'country_code': 'us', **kwargs}, timeout=10)
    resp.raise_for_status()
    return resp.json()

result = safe_search('python web framework 2026')
print(f'Results: {len(result.get("organic_results", []))}')

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time = 0
        self.state = 'closed'  # closed=normal, open=blocking, half_open=testing

    def call(self, func, *args, **kwargs):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = 'half_open'
                print(f'  Circuit half-open: testing...')
            else:
                remaining = self.reset_timeout - (time.time() - self.last_failure_time)
                return {'error': f'circuit_open (retry in {remaining:.0f}s)', 'results': []}
        try:
            result = func(*args, **kwargs)
            if self.state == 'half_open':
                self.state = 'closed'
                self.failure_count = 0
                print(f'  Circuit closed: service recovered')
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = 'open'
                print(f'  Circuit OPEN: {self.failure_count} failures. Blocking calls for {self.reset_timeout}s.')
            return {'error': str(e), 'results': []}

search_breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30)

def guarded_search(query):
    return search_breaker.call(safe_search, query)

# Test normal operation
result = guarded_search('test query')
print(f'Circuit state: {search_breaker.state}')
print(f'Failures: {search_breaker.failure_count}')

def classify_error(error):
    """Convert raw errors into agent-friendly classifications."""
    error_str = str(error).lower()
    if 'timeout' in error_str:
        return {'type': 'transient', 'action': 'retry',
                'message': 'Search timed out. Retry with simpler query.'}
    if '429' in error_str or 'rate' in error_str:
        return {'type': 'rate_limit', 'action': 'wait',
                'message': 'Rate limited. Wait 10 seconds before next call.'}
    if '401' in error_str or '403' in error_str:
        return {'type': 'auth', 'action': 'stop',
                'message': 'API key invalid or expired. Cannot search.'}
    if '5' in error_str[:1]:
        return {'type': 'server', 'action': 'retry',
                'message': 'Server error. Retry in 5 seconds.'}
    if 'circuit' in error_str:
        return {'type': 'circuit_open', 'action': 'fallback',
                'message': 'Search API unavailable. Use cached results or skip.'}
    return {'type': 'unknown', 'action': 'log',
            'message': f'Unexpected error: {error_str[:100]}'}

def agent_safe_search(query):
    """Search with full error handling for agent use."""
    result = guarded_search(query)
    if 'error' in result and result['error']:
        classified = classify_error(result['error'])
        print(f'  Error: [{classified["type"]}] {classified["message"]}')
        print(f'  Action: {classified["action"]}')
        return {'status': 'error', **classified, 'results': []}
    results = result.get('organic_results', [])
    return {'status': 'ok', 'results': results, 'count': len(results)}

# Test
result = agent_safe_search('best ai framework 2026')
print(f'Status: {result["status"]}, Results: {result.get("count", 0)}')