Розвиток AI-агента з доступом до зовнішніх API
AI-агент з доступом до зовнішніх API здатен отримувати актуальні дані й виконувати дії у сторонніх системах: CRM, ERP, сервісах погоди, біржових даних, державних реєстрах, платіжних системах. Це переводить агента з режиму "відповідати на питання" в режим "виконувати реальні завдання".
Архітектура інтеграції зовнішніх API
from typing import Any, Optional
import httpx
import asyncio
from pydantic import BaseModel
class APITool:
"""Базовий клас для API-інтеграції"""
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()
# Конкретна реалізація для 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})
Безпечне використання API в агенті
Прямий доступ агента до API потребує guardrails — без них агент може виконувати небажані операції:
from functools import wraps
import logging
logger = logging.getLogger(__name__)
class APIPermissionError(Exception):
pass
# Декоратор для контролю розпорядження
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
# Логування всіх API викликів агента
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:
# Додаткова валідація: дозволені статуси
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
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
# Функція-обертка для агента
def create_api_tool_for_agent(tool_instance, method_name: str) -> callable:
"""Створює синхронну обертку для агентного 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
Практичний кейс: агент продажу з доступом до CRM та зовнішніх API
Інструменти агента:
- CRM API (AmoCRM/Bitrix24): читання лідів, оновлення статусів, створення завдань
- Dadata API: збагачення даних про компанію (ІНН → повні реквізити, керівники)
- ФНС API: перевірка контрагентів, задолженості
- Telegram Bot API: сповіщення менеджеру
- Email API (SendGrid): відправлення автоматичних листів
Сценарій: новий ліді від юрособи → агент автоматично:
- Запитує дані з CRM
- За ІНН отримує повні реквізити через Dadata
- Перевіряє контрагента через ФНС
- Створює завдання менеджеру в CRM
- Відправляє привітальний лист з персоналізацією
Метрики:
- Час від появи ліді до першого контакту: 47хв → 4хв
- Повнота профілю ліді: 42% → 91%
- Менеджерський час на первинний скоринг: -68%
Rate Limiting та Cost Control
from asyncio import Semaphore
class RateLimitedAPITool:
"""API з обмеженням частоти запитів"""
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)
Графік
- Розвиток API-інтеграцій (3–5): 2–4 тижні
- Агентний цикл з обробкою помилок: 1–2 тижні
- Тестування та модель розпорядження: 1–2 тижні
- Всього: 4–8 тижнів