Разработка 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мин
- Completeness профиля лида: 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 API): 2–4 недели
- Агентный цикл с обработкой ошибок: 1–2 недели
- Тестирование и permission модель: 1–2 недели
- Итого: 4–8 недель