AI Agent Development with External API Access

AI Agent Development with External API Access

An AI agent with external API access can retrieve current data and perform actions in third-party systems: CRM, ERP, weather services, exchange data, government registries, payment systems. This transitions the agent from "answer questions" mode to "execute real tasks" mode.

Architecture of External API Integration

from typing import Any, Optional
import httpx
import asyncio
from pydantic import BaseModel

class APITool:
    """Base class for API integration"""

    def __init__(self, base_url: str, api_key: str = None, timeout: int = 10):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
        self.timeout = timeout

    async def request(self, method: str, endpoint: str, **kwargs) -> dict:
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.request(
                method,
                f"{self.base_url}{endpoint}",
                headers=self.headers,
                **kwargs
            )
            response.raise_for_status()
            return response.json()

# Concrete implementation for CRM
class CRMAPITool(APITool):
    async def get_customer(self, customer_id: str) -> dict:
        return await self.request("GET", f"/customers/{customer_id}")

    async def update_customer_status(self, customer_id: str, status: str) -> dict:
        return await self.request("PATCH", f"/customers/{customer_id}",
                                  json={"status": status})

    async def create_deal(self, customer_id: str, amount: float, stage: str) -> dict:
        return await self.request("POST", "/deals",
                                  json={"customer_id": customer_id, "amount": amount, "stage": stage})

Safe API Usage in Agents

Direct agent access to API requires guardrails — without them agents can perform unwanted operations:

from functools import wraps
import logging

logger = logging.getLogger(__name__)

class APIPermissionError(Exception):
    pass

# Decorator for permission control
def require_permission(permission: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, permission_context=None, **kwargs):
            if permission_context and not permission_context.has_permission(permission):
                raise APIPermissionError(f"Permission denied: {permission}")
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Log all API calls by agent
def log_api_call(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        logger.info(f"API call: {func.__name__}, args={kwargs}")
        result = await func(*args, **kwargs)
        logger.info(f"API result: {func.__name__} returned {type(result).__name__}")
        return result
    return wrapper

class SafeCRMTool(CRMAPITool):
    @log_api_call
    @require_permission("crm:read")
    async def get_customer(self, customer_id: str) -> dict:
        return await super().get_customer(customer_id)

    @log_api_call
    @require_permission("crm:write")
    async def update_customer_status(self, customer_id: str, status: str) -> dict:
        # Additional validation: allowed statuses
        allowed_statuses = ["active", "inactive", "pending"]
        if status not in allowed_statuses:
            raise ValueError(f"Status must be one of {allowed_statuses}")
        return await super().update_customer_status(customer_id, status)

API Error Handling

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class APIError(Exception):
    pass

class RateLimitError(APIError):
    pass

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(RateLimitError),
)
async def api_call_with_retry(tool, method, *args, **kwargs):
    try:
        return await getattr(tool, method)(*args, **kwargs)
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 429:
            raise RateLimitError("Rate limit exceeded")
        elif e.response.status_code >= 500:
            raise APIError(f"Server error: {e.response.status_code}")
        raise

# Wrapper function for agent
def create_api_tool_for_agent(tool_instance, method_name: str) -> callable:
    """Create sync wrapper for agent loop"""
    async def async_call(**kwargs) -> str:
        try:
            result = await api_call_with_retry(tool_instance, method_name, **kwargs)
            return json.dumps(result, ensure_ascii=False)
        except APIError as e:
            return json.dumps({"error": str(e), "retry": "automatic"})
        except Exception as e:
            return json.dumps({"error": f"Unexpected: {str(e)}"})

    def sync_call(**kwargs) -> str:
        return asyncio.run(async_call(**kwargs))

    return sync_call

Practical Case: Sales Agent with CRM and External API Access

Agent tools:

• CRM API (AmoCRM/Bitrix24): read leads, update statuses, create tasks
• Dadata API: company data enrichment (INN → full details, managers)
• Tax Authority API: counterparty verification, debts
• Telegram Bot API: manager notifications
• Email API (SendGrid): send automated emails

Scenario: new lead from legal entity → agent automatically:

1. Requests data from CRM
2. Gets full details by INN via Dadata
3. Verifies counterparty via Tax Authority
4. Creates task for manager in CRM
5. Sends welcome email with personalization

Metrics:

• Time from lead appearance to first contact: 47min → 4min
• Lead profile completeness: 42% → 91%
• Manager time on initial scoring: -68%

Rate Limiting and Cost Control

from asyncio import Semaphore

class RateLimitedAPITool:
    """API with request frequency limiting"""

    def __init__(self, api_tool, max_concurrent: int = 5, requests_per_minute: int = 60):
        self.tool = api_tool
        self.semaphore = Semaphore(max_concurrent)
        self.rpm_limit = requests_per_minute

    async def call(self, method: str, **kwargs) -> dict:
        async with self.semaphore:
            return await getattr(self.tool, method)(**kwargs)

Timeline

• API integrations (3–5): 2–4 weeks
• Agent loop with error handling: 1–2 weeks
• Testing and permission model: 1–2 weeks
• Total: 4–8 weeks