Receta 3 — Sincronizar resultados (polling)¶

Lo que vas a lograr: mantener tu sistema (CRM, data warehouse, hoja de cálculo) al día con los resultados de las llamadas, mediante un loop de sincronización por polling con paginación por cursor.

Cómo se entregan los resultados hoy¶

Los resultados se obtienen consultando el endpoint GET /api/analytics/calls (polling). No hay entrega push hacia tu sistema:

La entrega por webhooks (push) está en desarrollo; hoy la sincronización es por consulta (polling). No hay forma de registrar una URL de webhook vía la API pública.

El patrón recomendado es: paginar con cursor para traer un lote completo, persistir cada llamada de tu lado, y repetir cada N minutos filtrando solo lo nuevo por fecha o por campaña.

Concepto clave: paginación por cursor¶

GET /api/analytics/calls devuelve los resultados en páginas. Cada respuesta trae un objeto pagination:

"pagination": { "limit": 100, "count": 100, "has_more": true, "next_cursor": "eyJ0..." }

• has_more: true si quedan más resultados por traer.
• next_cursor: el token que pasas en el siguiente request para traer la página siguiente.

Para recorrer todas las páginas:

1. Haz el primer request sin cursor.
2. Si has_more es true, repite el request agregando cursor=<next_cursor>.
3. Sigue hasta que has_more sea false.

limit controla el tamaño de página (default 100, máximo 100).

API_BASE_URL="https://api.neuronstudio.ai"
API_KEY="nrn_live_tu_key_aqui"

Parte A — Recorrer todas las páginas (un barrido completo)¶

Con curl, paso a paso¶

Primera página (sin cursor):

curl -s "$API_BASE_URL/api/analytics/calls?limit=100" \
  -H "X-API-Key: $API_KEY"

Si la respuesta trae "has_more": true, copia el next_cursor y pide la siguiente:

curl -s "$API_BASE_URL/api/analytics/calls?limit=100&cursor=eyJ0..." \
  -H "X-API-Key: $API_KEY"

Repite hasta que has_more sea false.

Pseudo-código del barrido completo¶

cursor = null
hacer:
    url = "{API_BASE_URL}/api/analytics/calls?limit=100"
    si cursor no es null:
        url = url + "&cursor=" + cursor

    respuesta = GET(url, header: "X-API-Key: {API_KEY}")

    para cada llamada en respuesta.calls:
        persistir(llamada)          # upsert por call_id en tu sistema

    cursor = respuesta.pagination.next_cursor
mientras respuesta.pagination.has_more == true

Ejemplo en Python¶

import requests

API_BASE_URL = "https://api.neuronstudio.ai"
API_KEY = "nrn_live_tu_key_aqui"

headers = {"X-API-Key": API_KEY}

def barrer_todas_las_llamadas(params=None):
    params = dict(params or {})
    params["limit"] = 100
    cursor = None
    while True:
        if cursor:
            params["cursor"] = cursor
        resp = requests.get(
            f"{API_BASE_URL}/api/analytics/calls",
            headers=headers,
            params=params,
        )
        resp.raise_for_status()
        data = resp.json()

        for llamada in data["calls"]:
            persistir(llamada)  # upsert por call_id de tu lado

        pag = data["pagination"]
        if not pag["has_more"]:
            break
        cursor = pag["next_cursor"]

Parte B — El loop de sincronización (incremental, cada N minutos)¶

Para no traer siempre todo el histórico, en cada corrida filtra por fecha (from_date) o por campañas (campaign_ids).

Filtros útiles de GET /api/analytics/calls¶

Estrategia incremental recomendada¶

1. Guarda del lado de tu sistema el timestamp de la última sincronización exitosa.
2. En cada corrida, barre con from_date = última_sincronización (recorriendo todas las páginas con cursor, como en la Parte A).
3. Haz upsert por call_id al persistir (así reprocesar una llamada ya vista no la duplica — los análisis pueden actualizarse poco después de terminar la llamada).
4. Al terminar el barrido sin error, actualiza la marca de última sincronización.
5. Espera N minutos y repite.

Pseudo-código del loop¶

ultima_sync = leer_marca()        # ej. "2026-02-10T08:00:00Z"; null la primera vez

repetir para siempre:
    ahora = tiempo_actual_iso()
    params = { from_date: ultima_sync }   # omitir from_date si es la primera corrida

    barrer_todas_las_paginas(params):      # ver Parte A (cursor hasta has_more=false)
        para cada llamada:
            upsert_por_call_id(llamada)

    guardar_marca(ahora)          # solo si el barrido terminó sin error
    esperar(N_minutos)

Ejemplo en Python (loop incremental)¶

import time
from datetime import datetime, timezone

def ahora_iso():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

INTERVALO_MINUTOS = 10

def loop_sincronizacion():
    ultima_sync = leer_marca()  # None la primera vez
    while True:
        inicio = ahora_iso()
        params = {}
        if ultima_sync:
            params["from_date"] = ultima_sync

        barrer_todas_las_llamadas(params)  # reutiliza la función de la Parte A

        guardar_marca(inicio)
        ultima_sync = inicio
        time.sleep(INTERVALO_MINUTOS * 60)

Elige N según tu necesidad de frescura: cada 5–15 minutos suele ser razonable. Si recibes un 429 (límite de uso), respeta el header Retry-After que viene en la respuesta y vuelve a intentar después de ese tiempo.

Detalle por llamada (cuando lo necesites)¶

GET /api/analytics/calls devuelve un resumen por llamada. Para la transcripción completa y la grabación de una llamada puntual, pide su detalle:

curl -s "$API_BASE_URL/api/analytics/call/call_abc" \
  -H "X-API-Key: $API_KEY"

Conviene traer el detalle solo de las llamadas que lo ameriten (por ejemplo, las que requieren seguimiento), no de todas en cada corrida.

Resumen¶

Barrido completo (Parte A):
   GET /api/analytics/calls?limit=100
   └─ has_more? → repetir con &cursor=next_cursor  → hasta has_more=false

Loop incremental (Parte B):
   cada N minutos:
     from_date = última_sync
     └─ barrer todas las páginas (cursor)
        └─ upsert por call_id
     guardar marca de sincronización

Cuando la entrega por webhooks (push) esté disponible, podrás complementar este loop o reemplazarlo. Por ahora, el polling con cursor + filtro por fecha es el patrón soportado.