Интеграция AI-контакт-центра с облачной АТС (Mango Office, Zadarma, Sipuni, UIS)

Интеграция AI-контакт-центра с облачной АТС (Mango Office, Zadarma, Sipuni, UIS)

Российские облачные АТС предоставляют API для управления звонками и доступа к записям. Интеграция AI добавляет умную маршрутизацию, автотранскрибацию и аналитику без развёртывания собственной телефонии.

Общая архитектура интеграции

Облачная АТС (Mango/Zadarma/UIS)
      ↓ Webhook при входящем
AI-Platform API
      ↓ Инструкция АТС (переключить/ответить)
      ↓ URL аудиофайла при завершении
      ↓ STT → NLP → CRM update

Mango Office интеграция

import hashlib
import hmac
import json
import requests

class MangoOfficeIntegration:
    def __init__(self, api_key: str, api_salt: str):
        self.api_key = api_key
        self.api_salt = api_salt
        self.base_url = "https://app.mango-office.ru/vpbx"

    def sign(self, json_data: str) -> str:
        return hashlib.sha256(
            f"{self.api_key}{json_data}{self.api_salt}".encode()
        ).hexdigest()

    async def get_call_recording(self, recording_id: str) -> bytes:
        data = json.dumps({"recording_id": recording_id, "action": "download"})
        response = requests.post(
            f"{self.base_url}/queries/recording/post_load",
            data={"vpbx_api_key": self.api_key, "sign": self.sign(data), "json": data}
        )
        return response.content

    async def set_call_routing(self, from_number: str, to_extension: str):
        """Динамическая маршрутизация входящего звонка"""
        data = json.dumps({
            "from_number": from_number,
            "to_number": to_extension,
            "sip_headers": {"X-AI-Routed": "true"}
        })
        requests.post(
            f"{self.base_url}/routing/transfer",
            data={"vpbx_api_key": self.api_key, "sign": self.sign(data), "json": data}
        )

UIS (CloudTalk) интеграция

class UISIntegration:
    async def handle_call_event(self, event: dict) -> None:
        if event["type"] == "call.finished":
            recording_url = event.get("recording_url")
            if recording_url:
                audio = await self.download_recording(recording_url)
                analysis = await self.analyze_call(audio, event)
                await self.push_to_crm(event["contact_id"], analysis)

    async def set_smart_routing(self, caller_id: str) -> str:
        """Определяем куда направить звонок на основе истории клиента"""
        customer = await crm.lookup_by_phone(caller_id)
        if not customer:
            return "general_queue"

        if customer.get("open_tickets"):
            return "support_queue"
        elif customer.get("segment") == "vip":
            return "vip_queue"
        return "general_queue"

Post-call обработка (общий паттерн)

@app.post("/webhook/call-completed")
async def handle_completed_call(payload: dict):
    """Единый обработчик завершённых звонков от разных АТС"""
    recording_url = payload.get("recording_url") or payload.get("record")
    call_id = payload.get("call_id") or payload.get("uid")

    if not recording_url:
        return {"status": "no_recording"}

    # Асинхронная обработка в фоне
    asyncio.create_task(process_call_recording(call_id, recording_url))
    return {"status": "processing"}

async def process_call_recording(call_id: str, recording_url: str):
    audio = await download_audio(recording_url)
    transcript = await transcribe(audio)
    analysis = await analyze_call(transcript)
    await update_crm(call_id, transcript, analysis)

Сроки: webhook + post-call анализ для одной АТС — 1 неделя. Мультисистемная интеграция — 3–4 недели.