UTILIA OS SDK para Python

SDK Python para integrar aplicaciones externas con el sistema de soporte de UTILIA OS.

Instalación

pip install utilia-sdk

Catálogo de scopes (familia B, X-Api-Key)

Cada ExternalApp declara los permisos concedidos por el operador desde el panel de administración. Sin el scope adecuado, los endpoints responden 403 con errorCode: 'INSUFFICIENT_SCOPE'. Desde la versión 3.0.0, el SDK exporta los 56 scopes válidos como tipos Literal y constantes tipadas, espejo exacto del catálogo del backend.

Dos listas marcan lo blindado, y no dicen lo mismo: RGPD_SENSITIVE_SCOPES reúne lo que expone datos personales o hace algo irreversible; FINANCIAL_CRITICAL_SCOPES reúne lo que mueve dinero o corta un ingreso. Conceder cualquiera de las dos familias exige rol de administrador general y motivo razonado, pero quien revisa la solicitud necesita saber por qué está blindada la capacidad.

from utilia_sdk import (
    EXTERNAL_API_SCOPES,
    EXTERNAL_API_SCOPE_DESCRIPTIONS,
    FINANCIAL_CRITICAL_SCOPES,
    INSUFFICIENT_SCOPE_MESSAGE,
    RGPD_SENSITIVE_SCOPES,
    ExternalApiScope,
    is_external_api_scope,
)

# 56 scopes válidos del catálogo
assert len(EXTERNAL_API_SCOPES) == 56

# Metadatos canónicos (label, descripción, sensibilidad RGPD)
meta = EXTERNAL_API_SCOPE_DESCRIPTIONS["crm:invoices:cancel"]
print(meta.label, meta.requires_super_admin)  # 'Cancelar facturas' True

# Validar un valor recibido por configuración o webhook
if not is_external_api_scope(unknown_scope):
    raise ValueError("Scope no válido")

Dominio	Scopes	Notas
Contactos CRM	crm:contacts:read, :write, :read:secondary-email, :read:notes, :read:birthday, :read:opted-out, crm:clients:read	Notas y opted-out son RGPD CRÍTICOS (SUPER_ADMIN).
Leads CRM	crm:leads:read, :write, :read:scoring, :read:notes, :convert	Notas y convert son RGPD CRÍTICOS.
Clientes CRM	crm:clients:write, :read:fiscal-data, :read:notes	Fiscal-data y notes son RGPD CRÍTICOS.
Usuarios CRM	crm:users:read	Sprint 2026-05-25. Directorio interno.
Facturación	crm:invoices:read, :write, :issue, :cancel	Sprint 2026-05-25. cancel es RGPD CRÍTICO.
Tickets de soporte	support:tickets:read, :write, :ai	Sprint 2026-05-25. ai es RGPD CRÍTICO (IA sobre contenido sensible).
File Manager	fm:files:read, :write, :delete	Sprint 2026-05-25. delete es RGPD CRÍTICO (irreversible).
Billing	billing:payment-methods:read, :write, billing:charge	Sprint 2026-05-25. charge es RGPD CRÍTICO (mueve dinero).

Los 10 scopes RGPD CRÍTICOS están listados en RGPD_SENSITIVE_SCOPES. Asignar cualquiera de ellos a una ExternalApp exige rol SUPER_ADMIN y motivo razonado de mínimo 20 caracteres, registrado en el AuditLog inmutable (Ley 11/2021).

Mensaje OPACO del 403: cuando el backend rechaza una llamada por falta de scope, devuelve el texto literal expuesto como INSUFFICIENT_SCOPE_MESSAGE: "Esta operación requiere permisos adicionales. Contacta con el administrador del espacio de trabajo donde está instalada tu app." NO incluye el nombre del scope ni deep-link al panel admin. La app integradora NO debe parsearlo para descubrir qué scope falta; debe instruir al usuario final a contactar con el administrador. Razón: filtrar el nombre del scope facilitaría a un atacante con acceso parcial mapear los recursos accesibles vía API.

Uso rápido

Asíncrono (recomendado)

from utilia_sdk import UtiliaSDK, CreateTicketInput, CreateTicketUser, IdentifyUserInput

async with UtiliaSDK(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    # Identificar usuario
    user = await sdk.users.identify(
        IdentifyUserInput(external_id="user-123", email="user@example.com")
    )

    # Crear ticket
    ticket = await sdk.tickets.create(
        CreateTicketInput(
            user=CreateTicketUser(external_id="user-123"),
            title="Problema con facturación",
            description="No puedo ver mis facturas del mes pasado...",
        )
    )
    print(ticket.ticket_key)  # APP-0001

Síncrono

from utilia_sdk import UtiliaSDKSync, CreateTicketInput, CreateTicketUser

with UtiliaSDKSync(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    ticket = sdk.tickets.create(
        CreateTicketInput(
            user=CreateTicketUser(external_id="user-123"),
            title="Problema con facturación",
            description="No puedo ver mis facturas del mes pasado...",
        )
    )

Servicios disponibles

• sdk.tickets - Tickets de soporte (crear, listar, mensajes, cerrar, reabrir)
• sdk.users - Usuarios externos (identificar, listar)
• sdk.files - Archivos adjuntos (subir, obtener URL, quota)
• sdk.ai - IA (sugerencias, transcripción)
• sdk.errors - Errores del sistema (reportar, listar, estadísticas)
• sdk.budgets - Presupuestos CRM (CRUD, items, secciones, flujo, PDF, IA)
• sdk.budget_templates - Plantillas de presupuesto reutilizables
• sdk.budget_comments - Comentarios de presupuesto con visibilidad INTERNAL / CLIENT
• sdk.budget_signatures - Firma electrónica, magic links y certificado legal
• sdk.organization_settings - Configuración pública de la organización
• sdk.invoices - Facturación externa (crear, listar, detalle, PDF, anular, estadísticas, enviar por correo, cobrar sin el usuario delante, marcar cobrada, enlace público)
• sdk.invoices.scheduled_charges - Cobros programados a fecha futura sobre una factura
• sdk.invoices.refunds - Devolución del dinero ya cobrado
• sdk.payment_methods - Métodos de pago guardados del usuario final (SetupIntent Stripe)
• sdk.payment_authorizations - Autorizaciones de cobro: el permiso expreso del usuario para que se le cobre sin estar delante (sdk.mandates es su alias en desuso)
• sdk.payments - Cobro CON el usuario delante y consulta de PaymentIntents
• sdk.subscriptions - Suscripciones: cuotas que se repiten y se cobran solas
• sdk.webhooks - Verificación local de la firma de los avisos entrantes y lectura tipada del evento
• sdk.external_contacts - Contactos del CRM para apps externas de envío de correos (sync delta, opt-out/opt-in, webhooks CONTACT_*)
• sdk.external_leads - Leads del CRM para apps externas de prospección y sincronización CRM bidireccional (listado, sync delta, qualify / disqualify / convert, webhooks LEAD_*)
• sdk.external_clients - Clientes (empresas / cuentas) del CRM para apps externas de facturación y sincronización CRM bidireccional (listado, búsqueda por NIF/CIF timing-uniform, contactos vinculados, deactivate, webhooks CLIENT_*)
• sdk.chat - Plataforma de chat extensible (/external/v1/chat): publicar mensajes como app (send_message, v1) y leer canales, miembros e historial visible (list_channels / get_channel / list_members / list_messages / get_message, v2), más webhooks.verify() para validar la firma HMAC de los webhooks entrantes

Comentarios y firmas de presupuestos

Desde la versión 2.1.0 el SDK cubre los dos dominios centrales del flujo de aprobación de presupuestos.

Comentarios (async)

from utilia_sdk import (
    UtiliaSDK,
    CreateBudgetCommentInput,
    BudgetCommentVisibility,
    ListBudgetCommentsFilter,
)

async with UtiliaSDK(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    # Listar comentarios del cliente
    page = await sdk.budget_comments.list(
        budget_id,
        ListBudgetCommentsFilter(visibility=BudgetCommentVisibility.CLIENT, page=1, limit=20),
    )

    # Crear un comentario interno con menciones
    comment = await sdk.budget_comments.create(
        budget_id,
        CreateBudgetCommentInput(
            body="Revisar el descuento del item 2 antes de enviar.",
            visibility=BudgetCommentVisibility.INTERNAL,
            mentioned_user_ids=["6d1a4c6b-3f8b-4a0e-a0d1-b29f1f1c21cb"],
        ),
    )

Firmas y magic link (async)

from utilia_sdk import UtiliaSDK, SigningLinkRequest

async with UtiliaSDK(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    # Emitir magic link (idempotente por email)
    link = await sdk.budget_signatures.generate_signing_link(
        budget_id,
        SigningLinkRequest(
            signer_email="cliente@empresa.com",
            signer_name="María García",
            expires_in_hours=72,
            send_email=True,
        ),
    )

    if not link.reused and link.signing_url:
        print("Enviar al cliente:", link.signing_url)

    # Listar, verificar y certificar
    signatures = await sdk.budget_signatures.list(budget_id)
    verification = await sdk.budget_signatures.verify(budget_id, signatures[0].id)
    if not verification.valid:
        print("El documento ha cambiado después de la firma")

    pdf_bytes = await sdk.budget_signatures.download_certificate(
        budget_id, signatures[0].id
    )

Variantes síncronas

Todas las operaciones están disponibles también en UtiliaSDKSync:

from utilia_sdk import UtiliaSDKSync, SigningLinkRequest

with UtiliaSDKSync(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    link = sdk.budget_signatures.generate_signing_link(
        budget_id,
        SigningLinkRequest(signer_email="cliente@empresa.com"),
    )

Facturación y cobros

Desde la versión 3.0.0 el ciclo completo se recorre con la clave de API, sin salir del SDK: dar de alta al usuario, guardar su tarjeta, recoger su autorización de cobro, emitir la factura y enviarla por correo, montar la suscripción que se cobra sola, reintentar lo que falle, escuchar los avisos y devolver el dinero. Las facturas que emite tu aplicación son facturas de UTILIA OS: misma numeración, misma fiscalidad, mismo sellado VeriFactu y mismo motor de envío que las del equipo.

Requiere que la aplicación externa tenga billingEnabled = true en la configuración de la organización.

Catálogo de servicios y su credencial

Servicio	Qué hace	Credencial
sdk.users	Da de alta al usuario final y lo vincula con su ficha de cliente	X-Api-Key
sdk.payment_methods	Guarda y gestiona las tarjetas del usuario	X-Api-Key
sdk.payment_authorizations	Recoge y registra el permiso para cobrar sin el usuario delante	X-Api-Key
sdk.invoices	Emite, lista, envía por correo, cobra, marca cobrada y devuelve	X-Api-Key
sdk.invoices.scheduled_charges	Programa el cobro de una factura a fecha futura	X-Api-Key
sdk.invoices.refunds	Devuelve dinero ya cobrado	X-Api-Key
sdk.subscriptions	Cuotas que se repiten y se cobran solas	X-Api-Key
sdk.payments	Cobro CON el usuario delante, montando la pasarela en tu interfaz	X-Api-Key
sdk.webhooks	Verifica la firma de los avisos entrantes (no hace red)	ninguna
sdk.mcp.*	Herramientas para copilotos	sesión OAuth

Todo lo demás del producto que no aparezca en esa tabla vive en rutas internas que exigen una sesión de usuario iniciada, y el SDK no las alcanza.

Emitir una factura (síncrono)

from utilia_sdk import (
    UtiliaSDKSync,
    CreateInvoiceInput,
    CreateInvoiceLine,
    CreateInvoiceRecipient,
    CreateInvoiceUser,
)

with UtiliaSDKSync(base_url="https://os.utilia.ai/api", api_key="utl_...") as sdk:
    invoice = sdk.invoices.create(
        CreateInvoiceInput(
            user=CreateInvoiceUser(
                external_id="user_123",
                email="cliente@empresa.com",
                tax_id="B76543210",
            ),
            recipient=CreateInvoiceRecipient(
                name="Empresa Receptora SL",
                tax_id="B12345678",
                address="Calle Mayor 1",
                city="Madrid",
                postal_code="28001",
                country="España",
            ),
            lines=[
                CreateInvoiceLine(
                    name="Plan Premium mensual",
                    quantity=1,
                    unit_price=49.95,
                    tax_type="IVA_21",
                ),
            ],
            idempotency_key="ord_01HXYZ-invoice",
        )
    )
    print(invoice.invoice_number)  # p. ej. 202600001
    print(invoice.pdf_url)

Emitir una factura (asíncrono)

from utilia_sdk import (
    UtiliaSDK,
    CreateInvoiceInput,
    CreateInvoiceLine,
    CreateInvoiceRecipient,
    CreateInvoiceUser,
)

async with UtiliaSDK(base_url="https://os.utilia.ai/api", api_key="utl_...") as sdk:
    invoice = await sdk.invoices.create(
        CreateInvoiceInput(
            user=CreateInvoiceUser(external_id="user_123", email="x@y.com"),
            recipient=CreateInvoiceRecipient(
                name="Empresa SL",
                address="Calle Mayor 1",
                city="Madrid",
                postal_code="28001",
                country="España",
            ),
            lines=[
                CreateInvoiceLine(
                    name="Plan",
                    quantity=1,
                    unit_price=49.95,
                    tax_type="IVA_21",
                ),
            ],
        )
    )

Listado, detalle, PDF y anulación

from utilia_sdk import UtiliaSDKSync, InvoiceStatus, ListInvoicesFilters

with UtiliaSDKSync(base_url="https://os.utilia.ai/api", api_key="utl_...") as sdk:
    # Listar facturas del usuario con filtros
    page = sdk.invoices.list(
        "user_123",
        ListInvoicesFilters(status=InvoiceStatus.ISSUED, page=1, limit=20),
    )
    for invoice in page.invoices:
        print(invoice.invoice_number, invoice.total_amount, invoice.currency)

    # Detalle con líneas e impuestos
    detail = sdk.invoices.get(page.invoices[0].id, "user_123")
    print(detail.recipient_name, detail.lines)

    # PDF como bytes
    pdf_bytes = sdk.invoices.download_pdf(detail.id, "user_123")
    with open("factura.pdf", "wb") as fh:
        fh.write(pdf_bytes)

    # Anular
    cancelled = sdk.invoices.cancel(detail.id, "user_123", reason="Duplicada")
    assert cancelled.status == InvoiceStatus.CANCELLED

    # Estadísticas agregadas del usuario
    stats = sdk.invoices.get_stats("user_123")
    print(stats.total, stats.total_paid, stats.total_pending)

Guardar tarjeta (SetupIntent) y pagar

La app externa obtiene un clientSecret de UTILIA para montar Stripe Elements. La respuesta incluye publishable_key (clave de la plataforma de UTILIA, modelo Direct charges) y stripe_account_id (la cuenta Connect de la organización); se pasan a load_stripe/loadStripe con la opción stripeAccount. No configuras ninguna clave publishable propia.

from utilia_sdk import UtiliaSDKSync, CreateInvoiceUser

with UtiliaSDKSync(base_url="https://os.utilia.ai/api", api_key="utl_...") as sdk:
    # SetupIntent para que el cliente guarde una tarjeta nueva
    setup = sdk.payment_methods.create_setup_intent(
        CreateInvoiceUser(external_id="user_123", email="cliente@empresa.com")
    )
    print(setup.client_secret, setup.publishable_key, setup.stripe_account_id)

    # Métodos guardados del usuario
    methods = sdk.payment_methods.list("user_123")
    default = next((m for m in methods if m.is_default), None)

    # Marcar otro como predeterminado
    if methods:
        sdk.payment_methods.set_default(methods[-1].id, "user_123")

    # Desvincular un método (se marca como DETACHED, no se borra)
    sdk.payment_methods.remove(methods[0].id, "user_123")

Cobrar una factura

from utilia_sdk import UtiliaSDKSync

with UtiliaSDKSync(base_url="https://os.utilia.ai/api", api_key="utl_...") as sdk:
    # Pagar con método guardado (on_session, sin 3DS -> already_succeeded)
    result = sdk.payments.pay_invoice("inv_abc", "user_123", "pm_xyz")
    if result.already_succeeded:
        print("Cobro realizado")
    else:
        # La app externa completa el cobro con Stripe Elements
        print("Confirmar con clientSecret:", result.client_secret)

    # Pagar con tarjeta nueva (devuelve clientSecret para montar PaymentElement)
    result = sdk.payments.pay_invoice(
        "inv_abc", "user_123", save_card=True
    )

    # Histórico de intentos de cobro
    for intent in sdk.payments.get_payment_intents("inv_abc", "user_123"):
        print(intent.id, intent.status, intent.error_code or "")

El recorrido completo

1. El usuario y su ficha de cliente. Todo cuelga de aquí: facturas, tarjetas, autorizaciones y suscripciones.

usuario = await sdk.users.identify(
    IdentifyUserInput(
        external_id="user_01HXYZ",
        email="ana@estudiomarea.example",
        name="Estudio Marea SL",
        tax_id="B76543210",
        billing_address=CreateInvoiceBillingAddress(
            street="Avenida de Canarias 12, planta 3",
            city="Las Palmas de Gran Canaria",
            postal_code="35001",
            country="España",
        ),
    )
)
print(usuario.client_id)  # ficha del cliente en el CRM

tax_id y billing_address se propagan a la ficha del CRM y a la de la pasarela de pago: son los datos que salen impresos en el PDF de las facturas, y el NIF es lo que mira primero el puente para reconocer una ficha que ya existía. Mándalos antes de emitir la primera factura; corregirlos después obliga a rectificarla.

2. La tarjeta y la autorización de cobro. Guardar una tarjeta NO autoriza cobros. La autorización exige que el usuario acepte el texto de consentimiento canónico, que sirve el backend y debes mostrar TAL CUAL: es prueba legal.

consentimiento = await sdk.payment_authorizations.get_consent_text(
    ExternalRequestablePaymentAuthorizationScope.RECURRING
)
# ... muestras `consentimiento.text` sin cambiar ni una coma ...

autorizacion = await sdk.payment_authorizations.create(
    CreateExternalPaymentAuthorizationInput(
        user_id="user_01HXYZ",
        setup_intent_id=intent.setup_intent_id,
        consent_text=consentimiento.text,
        consent_version=consentimiento.version,
        accepted_at=datetime.now(timezone.utc).isoformat(),
        accepted_ip=peticion.client.host,
        accepted_user_agent=peticion.headers.get("user-agent"),
        scope=ExternalRequestablePaymentAuthorizationScope.RECURRING,
    )
)

3. La factura y su correo. Con send_email=True la factura sale por correo al emitirla, con su PDF adjunto, usando la plantilla y el remitente de la organización.

factura = await sdk.invoices.create(
    CreateInvoiceInput(..., send_email=True)
)
if factura.email_sent is False:
    # La factura ESTÁ emitida y numerada: solo falló el correo.
    # Reenvíala; no vuelvas a crearla.
    await sdk.invoices.send(
        factura.id, SendExternalInvoiceInput(user_id="user_01HXYZ")
    )

4. La suscripción. Con charge_mode = AUTO_STRIPE y sin autorización, el alta NO se degrada a cobro manual: falla con PAYMENT_AUTHORIZATION_REQUIRED, para que no creas que has montado un cobro que nunca se ejecutará.

suscripcion = await sdk.subscriptions.create(
    CreateExternalSubscriptionInput(
        user_id="user_01HXYZ",
        name="Plan Profesional mensual",
        start_date="2026-10-01",
        frequency=ExternalSubscriptionFrequency.MONTHLY_FIXED_DAY,
        day_of_month=1,
        charge_mode=ExternalSubscriptionChargeMode.AUTO_STRIPE,
        payment_authorization_id=autorizacion.id,
        send_email=True,
        lines=[
            CreateExternalSubscriptionLineInput(
                name="Plan Profesional", unit_price=49.95, tax_type="IGIC_7"
            )
        ],
    )
)

5. Cobros y reintentos. charge cobra sin el usuario delante; invoices.scheduled_charges.schedule deja el cobro programado para una fecha futura. Un status = REQUIRES_ACTION no es un error: el banco pide autenticación reforzada y hay que llevar al usuario al public_url de la respuesta.

Para programar un cobro sobre una factura suelta hace falta una autorización de alcance SCHEDULED_INVOICE. La de alcance RECURRING cubre las cuotas de una suscripción y el backend la rechaza con 409 PAYMENT_AUTHORIZATION_NOT_ACTIVE. Si tu aplicación hace las dos cosas, recoge las dos autorizaciones.

invoices.scheduled_charges.retry_now tiene un tope de cinco reintentos manuales por factura en 24 horas, no por cobro programado. Al superarlo responde 409 con el mensaje del límite, y ese 409 no trae error_code: se reconoce por el estado y el mensaje.

resultado = await sdk.invoices.charge(
    factura.id,
    ChargeExternalInvoiceInput(
        user_id="user_01HXYZ", idempotency_key=f"cuota-{factura.id}"
    ),
)
if resultado.status is ExternalChargeResultStatus.REQUIRES_ACTION:
    enviar_al_usuario(resultado.public_url)

6. Los avisos. El resultado de un cobro programado NO llega en la respuesta de la llamada que lo creó: llega por webhook. parse verifica la firma y devuelve el evento tipado.

@app.post("/webhooks/utilia")
async def recibir(request: Request) -> Response:
    crudo = (await request.body()).decode("utf-8")
    evento = sdk.webhooks.parse(crudo, request.headers, SECRETO)
    if evento.event == "CHARGE_SUCCEEDED":
        marcar_como_pagado(evento.data.invoice_id)
    elif evento.event == "CHARGE_REQUIRES_ACTION":
        # `public_url` llega con valor si el cobro lo lanzó la
        # aplicación, y NULO si viene del motor de cobros programados:
        # ahí hay que emitir un enlace nuevo y entregárselo al usuario.
        direccion = evento.data.public_url
        if direccion is None:
            enlace = await sdk.invoices.regenerate_public_link(
                evento.data.invoice_id,
                RegenerateExternalInvoicePublicLinkInput(
                    user_id=evento.data.external_user_id
                ),
            )
            direccion = enlace.public_url
        avisar_al_usuario(direccion)
    elif evento.event == "CHARGE_RETRIES_EXHAUSTED":
        suspender_servicio(evento.data.external_user_id)
    return Response(status_code=200)

7. El reembolso. Anular una factura no devuelve dinero; esto sí.

await sdk.invoices.refunds.create(
    factura.id,
    RefundExternalInvoiceInput(
        user_id="user_01HXYZ",
        reason=ExternalRefundReason.REQUESTED_BY_CUSTOMER,
        idempotency_key=f"devolucion-{factura.id}",
    ),
)

El PDF nunca sale de tu servidor, y el enlace público se entrega una vez

La ruta del PDF de la API externa exige la clave de API, y esa clave no debe salir de tu servidor. Para entregar la factura a tu usuario final hay dos vías y ninguna incluye la clave:

• sdk.invoices.download_pdf(id, user_id) devuelve los bytes en tu servidor.
• sdk.invoices.regenerate_public_link(id, datos) emite un enlace y devuelve la dirección pública, que sí puedes enviar por correo o abrir en el navegador.

Guarda la dirección cuando la emitas. Del enlace emitido el backend guarda solo su huella, así que no se puede recuperar: sdk.invoices.get_public_link(id, user_id=...) es una lectura pura que informa de la vigencia, la caducidad y las acciones permitidas, y devuelve public_url = None SIEMPRE. Sin enlace vigente responde 404 con INVOICE_PUBLIC_LINK_NOT_AVAILABLE.

Emitir revoca el anterior. Si vuelves a emitir, el enlace que ya habías repartido deja de servir. Un cobro con charge que acabe en REQUIRES_ACTION entrega un enlace adicional, y ese NO revoca el que el usuario ya tuviera.

Todos los enlaces que emite la API externa permiten solo ver y pagar (VIEW, PAY), también los de una factura de suscripción. Firmar o retirar una autorización de cobro recurrente se recoge por su camino propio.

Contactos para apps externas de envío de correos

Desde la versión 2.12.1 el SDK expone sdk.external_contacts para que aplicativos de envío de correos (Instantly, Smartlead, Apollo, Mailerlite, etc.) sincronicen el directorio de contactos del CRM, registren bajas y reactivaciones, y reciban los cambios en tiempo real vía webhooks. La autenticación es por X-Api-Key de la ExternalApp, con scopes específicos por campo.

Documentación completa de la API REST subyacente: docs/integrations/external-contacts-api.md.

Async

import os

from utilia_sdk import (
    ExternalContactSyncOptions,
    ResubscribeContactInput,
    ResubscribeProof,
    ResubscribeProofType,
    UnsubscribeContactInput,
    UnsubscribeContactSource,
    UtiliaSDK,
)
from utilia_sdk.models.external_contact import ExternalContactInclude


async def sincronizar() -> None:
    async with UtiliaSDK(
        base_url="https://os.utilia.ai/api",
        api_key=os.environ["UTILIA_API_KEY"],
    ) as sdk:
        cursor: str | None = None
        has_more = True
        while has_more:
            page = await sdk.external_contacts.sync(
                ExternalContactSyncOptions(
                    cursor=cursor,
                    updated_after=(None if cursor else "1970-01-01T00:00:00.000Z"),
                    limit=200,
                    include=[ExternalContactInclude.SECONDARY_EMAIL],
                )
            )
            for contact in page.data:
                await mi_app.upsert_suscriptor(contact)
            cursor = page.next_cursor
            has_more = page.has_more

        contact = await sdk.external_contacts.by_email("juan@empresa.com")
        if contact is None:
            return

        opt_out = await sdk.external_contacts.unsubscribe(
            contact.id,
            UnsubscribeContactInput(
                reason="Solicitud del propio contacto desde el footer",
                source=UnsubscribeContactSource.EXTERNAL_APP,
            ),
        )
        if not opt_out.changed:
            print("El contacto ya estaba dado de baja")

        await sdk.external_contacts.resubscribe(
            contact.id,
            ResubscribeContactInput(
                reason="El contacto se ha vuelto a registrar desde la web",
                proof=ResubscribeProof(
                    type=ResubscribeProofType.DOUBLE_OPT_IN,
                    captured_at="2026-05-21T09:15:00.000Z",
                    source_url="https://aplicativo-externo.com/proofs/abc",
                ),
            ),
        )

Síncrono

from utilia_sdk import (
    UnsubscribeContactInput,
    UnsubscribeContactSource,
    UtiliaSDKSync,
)


def registrar_baja(api_key: str, contact_id: str) -> None:
    with UtiliaSDKSync(
        base_url="https://os.utilia.ai/api", api_key=api_key
    ) as sdk:
        sdk.external_contacts.unsubscribe(
            contact_id,
            UnsubscribeContactInput(
                reason="Rebote permanente notificado por el proveedor SMTP",
                source=UnsubscribeContactSource.BOUNCE,
            ),
        )

by_email devuelve None cuando el backend responde 404 EMAIL_NOT_INDEXED. unsubscribe y resubscribe devuelven OptOutResult / OptInResult con contact, changed y recorded_at para soportar idempotencia desde el aplicativo externo. Si la ExternalApp no tiene crm:clients:read, client_ids llega como [] y primary_client como None; primary_client_id se mantiene siempre para correlación opaca.

Leads para apps externas

Desde la versión 2.13.0 el SDK expone sdk.external_leads para que aplicativos de prospección, plataformas de sincronización CRM bidireccional y herramientas de soporte gestionen los leads del CRM de UTILIA OS. La autenticación es por X-Api-Key de la ExternalApp con scopes específicos por campo (crm:leads:read, crm:leads:write, crm:leads:read:scoring, crm:leads:read:notes, crm:leads:convert).

Async

import os

from utilia_sdk import (
    ConvertLeadInput,
    DisqualifyLeadInput,
    DisqualifyLeadSource,
    ExternalLeadSyncOptions,
    QualifyLeadInput,
    QualifyLeadSource,
    UtiliaSDK,
)
from utilia_sdk.models.external_lead import ExternalLeadInclude


async def sincronizar_leads() -> None:
    async with UtiliaSDK(
        base_url="https://os.utilia.ai/api",
        api_key=os.environ["UTILIA_API_KEY"],
    ) as sdk:
        # Sincronización delta inicial (catálogo completo)
        cursor: str | None = None
        has_more = True
        while has_more:
            page = await sdk.external_leads.sync(
                ExternalLeadSyncOptions(
                    cursor=cursor,
                    updated_after=(None if cursor else "1970-01-01T00:00:00.000Z"),
                    limit=100,
                    include=[ExternalLeadInclude.SCORING],
                )
            )
            for lead in page.data:
                await mi_app.upsert_lead(lead)
            cursor = page.next_cursor
            has_more = page.has_more

        # Búsqueda exacta por email (devuelve None si no existe)
        lead = await sdk.external_leads.by_email("juan@acme.com")
        if lead is None:
            return

        # Cualificación tras detectar interés real
        await sdk.external_leads.qualify(
            lead.id,
            QualifyLeadInput(
                reason="El lead ha solicitado una demo concreta del producto",
                source=QualifyLeadSource.EXTERNAL_APP,
            ),
        )

        # Descarte por rebote permanente
        await sdk.external_leads.disqualify(
            lead.id,
            DisqualifyLeadInput(
                reason="Rebote permanente notificado por el proveedor SMTP",
                source=DisqualifyLeadSource.BOUNCE,
            ),
        )

        # Conversión (crea Contact y, opcionalmente, Opportunity)
        result = await sdk.external_leads.convert(
            lead.id,
            ConvertLeadInput(
                reason="El lead ha aceptado el presupuesto y solicita formalizar el contrato",
                create_opportunity=True,
            ),
        )
        print(result.contact_id, result.opportunity_id, result.client_id)

Síncrono

from utilia_sdk import (
    QualifyLeadInput,
    QualifyLeadSource,
    UtiliaSDKSync,
)


def cualificar(api_key: str, lead_id: str) -> None:
    with UtiliaSDKSync(
        base_url="https://os.utilia.ai/api", api_key=api_key
    ) as sdk:
        result = sdk.external_leads.qualify(
            lead_id,
            QualifyLeadInput(
                reason="Conversación cerrada con compromiso de presupuesto",
                source=QualifyLeadSource.MANUAL,
            ),
        )
        if not result.changed:
            print("El lead ya estaba cualificado")

by_email devuelve None cuando el backend responde 404 EMAIL_NOT_INDEXED_LEAD. Las mutaciones devuelven LeadMutationResult / LeadConvertResult con lead, changed y recorded_at (más contact_id / opportunity_id / client_id en convert). ConvertLeadInput.reason exige al menos 20 caracteres (queda en AuditLog inmutable). score y temperature solo llegan con scope crm:leads:read:scoring + include=scoring; notes solo con scope crm:leads:read:notes + include=notes.

Clientes para apps externas

Desde la versión 2.13.0 el SDK expone sdk.external_clients para que aplicativos de facturación, plataformas de sincronización CRM bidireccional y herramientas de soporte gestionen los clientes (empresas / cuentas) del CRM de UTILIA OS. La autenticación es por X-Api-Key de la ExternalApp con scopes específicos por campo (crm:clients:read, crm:clients:write, crm:clients:read:fiscal-data, crm:clients:read:notes).

Restricción de seguridad importante: el tax_id (NIF/CIF) JAMÁS aparece en listados ni en resultados de search. Solo se expone en get y by_tax_id con scope crm:clients:read:fiscal-data.

Async

import os

from utilia_sdk import (
    DeactivateClientInput,
    DeactivateClientSource,
    ExternalClientContactsFilters,
    ExternalClientSyncOptions,
    UtiliaSDK,
)
from utilia_sdk.models.external_client import ExternalClientInclude


async def sincronizar_clientes() -> None:
    async with UtiliaSDK(
        base_url="https://os.utilia.ai/api",
        api_key=os.environ["UTILIA_API_KEY"],
    ) as sdk:
        # Sincronización delta inicial
        cursor: str | None = None
        has_more = True
        while has_more:
            page = await sdk.external_clients.sync(
                ExternalClientSyncOptions(
                    cursor=cursor,
                    updated_after=(None if cursor else "1970-01-01T00:00:00.000Z"),
                    limit=100,
                )
            )
            for client in page.data:
                await mi_app.upsert_client(client)
            cursor = page.next_cursor
            has_more = page.has_more

        # Búsqueda exacta por NIF/CIF (devuelve None si no existe)
        client = await sdk.external_clients.by_tax_id(
            "B12345678", include=[ExternalClientInclude.FISCAL_DATA]
        )
        if client is None:
            return

        # Contactos vinculados al cliente
        contactos = await sdk.external_clients.list_contacts(
            client.id,
            ExternalClientContactsFilters(can_receive_emails=True),
        )
        for contact in contactos.data:
            print(contact.full_name, contact.email)

        # Desactivación auditada (no existe DELETE vía API externa)
        await sdk.external_clients.deactivate(
            client.id,
            DeactivateClientInput(
                reason="Solicitud explícita del cliente desde la app externa",
                source=DeactivateClientSource.EXTERNAL_APP,
            ),
        )

Síncrono

from utilia_sdk import (
    DeactivateClientInput,
    DeactivateClientSource,
    UtiliaSDKSync,
)


def desactivar(api_key: str, client_id: str) -> None:
    with UtiliaSDKSync(
        base_url="https://os.utilia.ai/api", api_key=api_key
    ) as sdk:
        result = sdk.external_clients.deactivate(
            client_id,
            DeactivateClientInput(
                reason="Política de baja por inactividad de la cuenta",
                source=DeactivateClientSource.AUTOMATION,
            ),
        )
        if not result.changed:
            print("El cliente ya estaba inactivo")

by_tax_id devuelve None cuando el backend responde 404 TAX_ID_NOT_INDEXED y aplica un rate limit más estricto (100 req/h). deactivate devuelve ClientMutationResult con client, changed y recorded_at. DeactivateClientInput.reason exige entre 20 y 500 caracteres (queda en AuditLog inmutable y se propaga en el webhook CLIENT_DEACTIVATED). La API NUNCA expone IBAN, condiciones de pago, creditLimit, discount, metadata ni asignaciones del equipo comercial.

Chat para apps externas (sdk.chat)

Desde la versión 2.20.0 el SDK expone sdk.chat (async y sync) para que una aplicación externa participe en los canales de chat de UTILIA OS como una identidad propia (EXTERNAL_APP). Superficie REST /external/v1/chat.

• v1 (publicar): send_message(). Requiere scope chat:messages:send y una concesión (AppChannelGrant) activa en el canal.
• v2 (leer): list_channels(), get_channel(), list_members(), list_messages(), get_message(). Requieren chat:channels:view / chat:messages:view. El backend aplica un filtro de visibilidad fail-closed: la app nunca recibe TEAM_ONLY/PRIVATE, respuestas privadas de IA ni mensajes borrados.

Doble puerta: el scope de la app autoriza la acción y la concesión por canal autoriza el recurso. El organizationId lo fija el backend desde la credencial; para un canal no concedido o de otro tenant responde 404.

from utilia_sdk import UtiliaSDK, SendChatMessageInput

async with UtiliaSDK(base_url="https://os.utilia.ai/api", api_key="...") as sdk:
    # Publicar una lectura como la propia app en un canal concedido.
    await sdk.chat.send_message(
        channel_id,
        SendChatMessageInput(content="Termómetro del cliente: 72/100."),
    )

    # Leer el historial visible (cursor descendente).
    page = await sdk.chat.list_messages(channel_id, limit=50)

Verificar la firma de un webhook

sdk.chat.webhooks.verify() valida localmente (sin red) la firma HMAC de los webhooks entrantes: HMAC-SHA256 sobre {timestamp}.{raw_body} con hmac.compare_digest y ventana anti-replay (300 s por defecto). Devuelve el ChatWebhookPayload tipado o lanza UtiliaSDKError.

payload = sdk.chat.webhooks.verify(
    raw_body,
    signature=headers["X-Utilia-Signature"],
    timestamp=headers["X-Utilia-Timestamp"],
    secret=os.environ["UTILIA_WEBHOOK_SECRET"],
)

Ejemplo completo: el "Termómetro" (anti-bucle)

El "Termómetro" publica lecturas como app y reacciona a los mensajes de personas. El anti-bucle es esencial: solo responde si el autor NO es una app (author.kind != "APP"), de modo que no reacciona a sí mismo ni a otras apps.

# Manejador del webhook en el servidor del integrador.
async def on_utilia_webhook(raw_body: str, headers: dict[str, str]) -> None:
    try:
        payload = sdk.chat.webhooks.verify(
            raw_body,
            signature=headers["X-Utilia-Signature"],
            timestamp=headers["X-Utilia-Timestamp"],
            secret=os.environ["UTILIA_WEBHOOK_SECRET"],
        )
    except UtiliaSDKError:
        return  # firma inválida o fuera de la ventana anti-replay

    if payload.event == "chat.message.created":
        message = await sdk.chat.get_message(payload.data["messageId"])
        # Anti-bucle: no reaccionar a mensajes de apps (incluida la nuestra).
        if message.author.kind != "APP":
            await sdk.chat.send_message(
                message.channel_id,
                SendChatMessageInput(
                    content="Recibido. Recalculo el termómetro…",
                    parent_message_id=message.id,
                ),
            )

Rectificativas y notas de crédito: reservadas al equipo

Desde la versión 3.0.0, el SDK no expone la emisión ni el parche de metadata legal de una rectificativa. Cualquier llamada a sdk.invoices.rectifications.update_legal_metadata(...) lanza UtiliaSDKError con la vía correcta.

El motivo: emitir una rectificativa exige elegir su código legal (R1 a R5 del RD 1619/2012) y de esa elección depende cómo la Agencia Tributaria clasifica el documento. Es una decisión del equipo que lleva la contabilidad, no de una aplicación que factura. La vía oficial es el panel de UTILIA OS o la herramienta MCP crm_invoices_rectifications.

Lo que sí puede hacer una aplicación externa es devolver el dinero:

reembolso = await sdk.invoices.refunds.create(
    "inv-uuid",
    RefundExternalInvoiceInput(
        user_id="user_01HXYZ",
        reason=ExternalRefundReason.REQUESTED_BY_CUSTOMER,
        # Por defecto NO se emite la rectificativa: la emite el equipo
        # para elegir su código legal.
        create_rectifying_invoice=False,
    ),
)

Hasta la 2.28.0 estos métodos apuntaban a /finance/*, rutas internas que exigen una sesión de usuario iniciada: devolvían 401 con la clave de API y también con OAuth, porque el token OAuth solo se resuelve en el carril MCP.

Errores tipados de facturación

Desde la versión 2.6.0, el SDK exporta modelos canónicos para el código INVOICE_TAX_COHERENCE_ISSUES. El backend lo emite con HTTP 400 cuando la factura tiene incongruencias fiscales (IGIC en peninsular, IVA en Canarias, taxType nulo, tipo no perteneciente al sistema, etc.) antes de persistir o emitir. Endpoints emisores: POST /finance/invoices, POST /finance/invoices/:id/issue y POST /external/v1/invoices.

from utilia_sdk import (
    UtiliaSDK,
    UtiliaSDKError,
    INVOICE_TAX_COHERENCE_ERROR_CODE,
    InvoiceTaxCoherenceErrorPayload,
)

async with UtiliaSDK(
    base_url="https://os.utilia.ai/api",
    api_key="tu-api-key",
) as sdk:
    try:
        await sdk.invoices.create({...})
    except UtiliaSDKError as exc:
        if exc.error_code == INVOICE_TAX_COHERENCE_ERROR_CODE:
            # Cuando el backend devuelve este código, la corrección es
            # local: arregla los `taxType` de las líneas y vuelve a
            # intentarlo. NO reintentes automáticamente.
            print("La factura tiene incoherencias fiscales.")
        raise

Si tu cliente HTTP captura el cuerpo crudo del 400 y necesitas el detalle por línea, parséalo con el modelo importado o consulta el endpoint POST /finance/invoices/preview-tax-coherence (interno, OAuth) antes de intentar crear o emitir. Su respuesta tiene exactamente la forma de InvoiceTaxCoherenceErrorPayload:

def describe_tax_coherence_error(raw: dict) -> list[str]:
    payload = InvoiceTaxCoherenceErrorPayload.model_validate(raw)
    descriptions: list[str] = []
    for issue in payload.issues:
        idx = issue.line_index + 1
        if issue.code == "NULL_TAX_TYPE":
            sug = (
                f" Sugerencia: {issue.suggested_tax_type}."
                if issue.suggested_tax_type
                else ""
            )
            descriptions.append(f"Línea {idx}: falta el tipo impositivo.{sug}")
        elif issue.code == "INCOHERENT_WITH_SYSTEM":
            cur = issue.current_tax_type or "sin tipo"
            sug = (
                f" Cambia a {issue.suggested_tax_type}."
                if issue.suggested_tax_type
                else ""
            )
            descriptions.append(
                f"Línea {idx}: {cur} no es compatible con el sistema "
                f"{payload.tax_system}.{sug}"
            )
        elif issue.code == "RATE_NOT_IN_SYSTEM":
            descriptions.append(
                f"Línea {idx}: el porcentaje {issue.current_tax_rate}% "
                f"no existe en el catálogo del sistema {payload.tax_system}."
            )
        elif issue.code == "LEGACY_HEADER_RATE_MISMATCH":
            descriptions.append(
                f"Línea {idx}: tipo legado en la cabecera no coincide "
                "con el calculado por línea. Migración pendiente."
            )
    return descriptions

Actualizaciones en tiempo real (SSE)

Recibe notificaciones cuando un agente responde, cambia el estado, resuelve o cierra un ticket:

Asíncrono

async with UtiliaSDK(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    async for event in sdk.tickets.stream_updates(user_id="user-123"):
        print(f"Ticket {event['ticketKey']} actualizado: {event['type']}")
        # event['type']: 'comment-added' | 'status-changed'

Síncrono

with UtiliaSDKSync(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    for event in sdk.tickets.stream_updates(user_id="user-123"):
        print(f"Ticket {event['ticketKey']} actualizado: {event['type']}")

Reportar errores del sistema

import traceback
from utilia_sdk import UtiliaSDK, ReportErrorInput

async with UtiliaSDK(base_url="https://os.utilia.ai/api", api_key="tu-api-key") as sdk:
    try:
        await procesar_pago(orden)
    except Exception as e:
        result = await sdk.errors.report(ReportErrorInput(
            message=str(e),
            module="pagos",
            severity="critical",
            stack=traceback.format_exc(),
            endpoint="/api/payments",
            method="POST",
        ))
        print(result.hash)         # Hash de deduplicación
        print(result.deduplicated) # True si ya existía

OAuth y Sign In

Desde la versión 0.5.0, el SDK incluye soporte nativo para OAuth 2.1 con PKCE:

from utilia_sdk import UtiliaSDK

sdk = UtiliaSDK(
    base_url="https://os.utilia.ai/api",
    oauth={
        "client_id": "client_xxxxxxxxxx",
        "redirect_uri": "http://localhost:8000/callback",
        "scopes": ["openid", "profile", "email"],
    },
)

# Generar URL de autorización
auth_url = await sdk.oauth.get_authorization_url()

# Opcionalmente, solicitar un tema específico para la pantalla de consentimiento
dark_url = await sdk.oauth.get_authorization_url(theme="dark")

# Manejar callback
tokens = await sdk.oauth.handle_callback(code)
user_info = sdk.oauth.get_user_info()

Documentación completa: https://os.utilia.ai/dashboard/docs/integrar-sdk/sdk-python-guia-oauth

Manejo de errores

from utilia_sdk import UtiliaSDKError, ErrorCode

try:
    ticket = await sdk.tickets.create(data)
except UtiliaSDKError as e:
    if e.is_unauthorized:
        print("API Key inválida")
    elif e.is_rate_limited:
        print("Demasiadas peticiones")
    elif e.is_retryable:
        print("Error temporal, reintentar")
    else:
        print(f"Error: {e.code} - {e.message}")

Licencia

MIT