utilia-sdk 2.20.0

SDK Python para integrar aplicaciones externas con UTILIA OS

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 2.15.0, el SDK exporta los 29 scopes válidos como tipos Literal y constantes tipadas.

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

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

# 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="[e-mail]")
    )

    # 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)
• sdk.payment_methods - Métodos de pago guardados del usuario final (SetupIntent Stripe)
• sdk.payments - Cobro de facturas y consulta de PaymentIntents
• 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="[e-mail]",
            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="[e-mail]"),
    )

Facturación y pagos

Desde la versión 2.2.0, el SDK cubre el ciclo completo de facturación externa respaldado por Stripe Connect: emisión de facturas numeradas (AEAT/VeriFactu), guardado de tarjetas del usuario final y cobro on-session u off-session.

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

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="[e-mail]",
                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="[e-mail]"),
            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="[e-mail]")
    )
    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 "")

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("[e-mail]")
        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("[e-mail]")
        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

Desde la versión 2.5.0, el SDK expone los tipos canónicos del flujo de rectificativas y notas de crédito definido por el RD 1619/2012 art. 15.3. La creación vive en endpoints INTERNOS (POST /finance/invoices/:id/rectify y POST /finance/invoices/:id/credit-note), por lo que requieren OAuth con permiso INVOICES_RECTIFY. La metadata legal sobre borradores se parchea con client.invoices.rectifications.update_legal_metadata(...).

Modalidad rectification_type y default

• COMPLETE (sustitución total): reemplaza la factura original. Importe libre. Sólo puede existir UNA rectificativa COMPLETE viva por factura original.
• DIFFERENCE (por diferencias): recoge solo el delta. El cap acumulado por diferencias no puede rebasar el total de la original.

Cambio en 2.5.0: el endpoint credit-note ahora aplica DIFFERENCE por defecto cuando no se envía rectification_type. Si tu integración espera sustitución total, envía rectification_type="COMPLETE" (o RectificationType.COMPLETE) de forma explícita.

from utilia_sdk import (
    UtiliaSDK,
    RectificationCode,
    RectificationType,
    UpdateRectificationMetadataInput,
)

async with UtiliaSDK(
    base_url="https://os.utilia.ai/api",
    oauth={"client_id": "...", "redirect_uri": "..."},
) as sdk:
    # Parchea la metadata legal sobre una rectificativa o nota de
    # crédito en estado DRAFT antes de emitirla.
    updated = await sdk.invoices.rectifications.update_legal_metadata(
        invoice_id="inv-uuid",
        body=UpdateRectificationMetadataInput(
            rectification_code=RectificationCode.R1,
            rectification_reason=(
                "Error en el NIF del cliente, ahora B12345678."
            ),
            rectification_type=RectificationType.COMPLETE,
        ),
    )

Tipar los errores de negocio

from utilia_sdk import (
    RECTIFICATION_BUSINESS_ERROR_CODES,
    RectificationDifferenceExceedsRemainingError,
    RectificationSubstitutionAlreadyExistsError,
)

print(RECTIFICATION_BUSINESS_ERROR_CODES)
# (
#   'RECTIFICATION_DIFFERENCE_EXCEEDS_REMAINING',
#   'RECTIFICATION_SUBSTITUTION_ALREADY_EXISTS',
# )

# Si capturas el cuerpo del 400 desde tu propio cliente HTTP:
def describe_rectification_error(raw: dict) -> str:
    code = raw.get("code")
    if code == "RECTIFICATION_DIFFERENCE_EXCEEDS_REMAINING":
        err = RectificationDifferenceExceedsRemainingError.model_validate(raw)
        return (
            "La rectificativa por diferencias rebasa el saldo restante. "
            f"Acumulado previo: {err.previous_total} €, intento: "
            f"{err.new_amount} €, total original: {err.original_total} €."
        )
    if code == "RECTIFICATION_SUBSTITUTION_ALREADY_EXISTS":
        err = RectificationSubstitutionAlreadyExistsError.model_validate(raw)
        return (
            f"Ya existe una rectificativa por sustitución activa "
            f"({err.existing_invoice_number}, estado {err.existing_status}). "
            "Anúlala antes de emitir otra."
        )
    return f"Error desconocido: {code}"

Crear rectificativas y notas de crédito

El SDK aún no expone un método dedicado para rectify y credit-note; mientras tanto, llama al endpoint con tu cliente HTTP autenticado vía OAuth. Recuerda enviar rectification_type explícito para no depender del default (que ahora es DIFFERENCE en notas de crédito).

body_credit_note = {
    "creditNoteReason": "Devolución parcial de las horas no consumidas en marzo.",
    "rectificationCode": "R4",
    "rectificationType": "DIFFERENCE",  # explícito; default desde 2.5.0
    "lines": [
        {
            "name": "Horas no consumidas",
            "quantity": 1,
            "unitPrice": 200.0,
            "taxType": "IGIC_7",
        }
    ],
}

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/integraciones-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