API de plataforma de TotalPro.es

Integración de data.totalpro.es en 5 pasos

URL base: https://totalpro.es/api/v1 · Todas las respuestas son JSON; los errores siguen el formato RFC 9457.

1. Login con la cuenta de totalpro.es

En /v1/auth/login y /v1/auth/agent_login, data.totalpro.es comprueba las credenciales contra totalpro.es y crea o actualiza su usuario local. El saldo para la IA viene en entitlements.

import httpx
TP = "https://totalpro.es/api/v1"
HEADERS = {"Authorization": f"Bearer {TOTALPRO_API_KEY}"}   # tp_sk_… (variable de entorno)

r = httpx.post(f"{TP}/auth/verify", headers=HEADERS, json={
    "email": email, "password": password,
    "device": {"machine_id": machine_id, "name": hostname, "os": "Windows 11 24H2", "agent_version": "0.3.0"},
    "client_ip": request.client.host,   # IP real del usuario: limita los intentos fallidos por usuario, no por servidor
})
if r.status_code != 200:
    raise HTTPException(401, r.json()["detail"])          # invalid_credentials, user_blocked, too_many_attempts…
user = r.json()["user"]
ent = user["entitlements"]               # {"can_use_ai": true, "available_eur": 2.0, "gift": {...}, ...}

2. Comprobar el saldo y cobrar cada petición de IA

La IA se paga por uso desde el monedero, en euros, según el modelo (precios por modelo). Antes de llamar al modelo, comprueba can_use_ai; después, envía los tokens de la respuesta (los de usageMetadata de Gemini) y totalpro.es la cobra: primero del regalo de bienvenida y luego del saldo. Es idempotente por id. Si tras cobrar can_use_ai es false (sin saldo o límite mensual alcanzado), no hagas más llamadas al modelo para ese usuario y enséñale el enlace a https://totalpro.es/wallet.

ent = httpx.get(f"{TP}/users/{tp_user_id}/entitlements", headers=HEADERS).json()   # o la caché (máx. 60 s)
if not ent["can_use_ai"]:
    return sin_saldo(ent["blocked_reason"])          # no_balance | monthly_limit

resp = gemini.generate_content(...)
u = resp.usage_metadata
r = httpx.post(f"{TP}/usage", headers=HEADERS, json={"records": [{
    "id": request_id, "user_id": tp_user_id, "model": "gemini-3.8-flash",
    "input_tokens": u.prompt_token_count, "cached_input_tokens": u.cached_content_token_count or 0,
    "output_tokens": u.candidates_token_count, "thinking_tokens": u.thoughts_token_count or 0,
    "machine_id": machine_id, "conversation_id": conversation_id,
}]})
res = r.json()["results"][0]                          # {"charged_eur": 0.0042, "available_eur": 1.83, "can_use_ai": true, ...}

3. Abrir un ticket cuando la IA no lo resuelve

El ticket es privado. El equipo de TotalPro lee el informe interno de la IA y envía un precio cerrado (webhook ticket.quoted con pay_url); el cliente paga desde su monedero y un técnico lo resuelve.

httpx.post(f"{TP}/events", headers=HEADERS, json={
    "id": f"esc-{conversation_id}", "type": "conversation.escalated", "user_id": tp_user_id,
    "data": {"conversation_id": conversation_id, "create_ticket": True, "category": "correo",
             "title": "Outlook no sincroniza", "description": lo_que_cuenta_el_cliente,
             "internal_report": informe_para_el_tecnico, "diagnostics": inventario},
})

4. Recibir webhooks y verificar la firma

Cabecera TotalPro-Signature: t=<unix>,v1=<hex>, con v1 = HMAC-SHA256(secreto, "<t>.<cuerpo>"). Responde 2xx en menos de 5 s; si no, se reintenta durante unos 2 días. Usa TotalPro-Event-Id para no procesar dos veces el mismo evento.

import hmac, hashlib, time

@app.post("/v1/hooks/totalpro")
async def totalpro_webhook(request: Request):
    body = await request.body()
    parts = dict(p.split("=", 1) for p in request.headers["TotalPro-Signature"].split(","))
    expected = hmac.new(WEBHOOK_SECRET.encode(), f"{parts['t']}.".encode() + body, hashlib.sha256).hexdigest()
    if abs(time.time() - int(parts["t"])) > 300 or not hmac.compare_digest(expected, parts["v1"]):
        raise HTTPException(400, "firma no válida")
    event = json.loads(body)
    match event["type"]:
        case "user.blocked":         desactivar_usuario(event["data"]["user_id"])
        case "entitlements.updated": actualizar_saldo(event["data"]["user_id"], event["data"]["entitlements"])
        case "ai.blocked":           pausar_ia(event["data"]["user_id"], event["data"]["reason"])
        case "ai.balance_low":       avisar_saldo_bajo(event["data"]["user_id"])
        case "session.revoked":      cerrar_sesiones(event["data"])
        case "ticket.status_changed": notificar_agente(event["data"]["ticket"])
    return {"ok": True}

5. Agente: sesión del usuario

Si el agente habla directamente con totalpro.es, inicia sesión una vez y renueva el token antes de que caduque. El refresh token es de un solo uso.

curl -X POST https://totalpro.es/api/v1/auth/login -H "Content-Type: application/json" \
  -d '{"email":"laura@empresa.es","password":"…","device":{"machine_id":"7c2e…","name":"PORTATIL-LAURA"}}'

curl https://totalpro.es/api/v1/me -H "Authorization: Bearer tp_at_…"            # perfil, saldo para la IA y monedero
curl -X POST https://totalpro.es/api/v1/auth/refresh -d '{"refresh_token":"tp_rt_…"}' -H "Content-Type: application/json"