📘 2025 Report:Mexico Economic Review 2025 — outlook, charts, and sector signalsRead

    Ejemplos de Código

    Python, Node.js y cURL — Todos los modelos de integración

    Recursos para desarrolladores

    Seis recursos, seis propósitos. Elija el que coincida con dónde está en su integración.

    Configuración Inicial

    import requests
    
    BASE_URL = "https://service.criskco.com/apiservice.svc"
    HEADERS = {
        "apiId": "YOUR_API_ID",
        "apiKey": "YOUR_API_KEY",
        "Content-Type": "application/json",
        "Accept-Encoding": "gzip, deflate"
    }

    Nivel 1 — Verify (solo RFC)

    Validaciones a partir del RFC — sin CIEC: contribuyente, certificados de e.firma y accionistas (SIGER/RUG).

    Verificar certificados de e.firma (síncrono)

    Devuelve el estado de los certificados en la misma respuesta. Deja CertificateSerialNumber vacío para consultar todos los certificados del RFC.

    def verify_efirma_certificates(rfc, certificate_serial_number="", transaction_id=""):
        payload = {
            "TransactionId": transaction_id,
            "Rfc": rfc,
            "CertificateSerialNumber": certificate_serial_number,
        }
        r = requests.post(f"{BASE_URL}/VerifyEFirmaCertificates", headers=HEADERS, json=payload)
        return r.json()
    
    result = verify_efirma_certificates("XXXX000000X00")

    Verificar e.firma vía webhook (asíncrono)

    Misma verificación, entregada a la URL de callback de la suscripción indicada en WebhookSubscriptionId.

    # Async: CRiskCo posts the result to the callback registered on the subscription.
    def verify_efirma_async(rfc, webhook_subscription_id, certificate_serial_number="", transaction_id=""):
        payload = {
            "TransactionId": transaction_id,
            "Rfc": rfc,
            "CertificateSerialNumber": certificate_serial_number,
            "WebhookSubscriptionId": webhook_subscription_id,
        }
        r = requests.post(f"{BASE_URL}/VerifyEFirmaCertificatesWebhook", headers=HEADERS, json=payload)
        return r.json()

    Verificar e.firma en lote (1–100 RFC)

    Envía hasta 100 elementos en una sola llamada; los resultados llegan al callback registrado.

    def verify_efirma_batch(items, webhook_subscription_id):
        """items: list of {"TransactionId", "Rfc", "CertificateSerialNumber"} — 1 to 100 entries."""
        payload = {"WebhookSubscriptionId": webhook_subscription_id, "Items": items}
        r = requests.post(f"{BASE_URL}/VerifyEFirmaCertificatesBatchWebhook", headers=HEADERS, json=payload)
        return r.json()
    
    verify_efirma_batch([
        {"TransactionId": "", "Rfc": "XXXX000000X00", "CertificateSerialNumber": ""},
        {"TransactionId": "", "Rfc": "XXXX11111X11", "CertificateSerialNumber": ""},
    ], 1)

    Accionistas (SIGER/RUG) y sus documentos

    Flujo de dos pasos: primero solicita los accionistas y luego los documentos del registro con el uuid devuelto en el callback.

    # Step 1 — request the shareholders (SIGER/RUG) lookup; the result arrives on your callback.
    def request_shareholders(tax_id, subscription_id, query_socios=True):
        params = {
            "taxId": tax_id,
            "subscriptionId": subscription_id,
            "querySocios": str(query_socios).lower(),
        }
        r = requests.get(f"{BASE_URL}/siger-webhook", headers=HEADERS, params=params)
        return r.json()
    
    # Step 2 — fetch the supporting documents using the uuid returned on the callback.
    def request_shareholders_documents(tax_id, uuid, subscription_id):
        params = {"taxId": tax_id, "subscriptionId": subscription_id, "uuid": uuid}
        r = requests.get(f"{BASE_URL}/siger-pdf-webhook", headers=HEADERS, params=params)
        return r.json()

    Nivel 2 — Inteligencia Financiera (SAT/CIEC)

    Onboarding con CIEC para acceder a CFDI, estados financieros (CFSS), nómina y FinScore.

    Modelo A — Approve API (Control Total)

    Paso 1 — Validar RFC

    def validate_rfc(rfc, name=None, postal=None):
        params = {"rfc": rfc}
        if name:
            params["name"] = name
        if postal:
            params["postal"] = postal
    
        r = requests.get(f"{BASE_URL}/ValidateRFC", headers=HEADERS, params=params)
        return r.json()
    
    result = validate_rfc("GAPXXXXXXXXX", name="GAP", postal="06600")
    if not result["Success"]:
        raise Exception(f"RFC invalid: {result['message']}")
    print(result["message"])

    Paso 2 — Onboarding del Aplicante

    def onboard_applicant(rfc, ciec_password, email, ref_id=""):
        payload = {
            "IsAgreeTerms": True,
            "DateAgreeTerms": "2026-03-24",
            "VersionAgreeTerms": "1",
            "Email": email,
            "User": rfc,
            "Password": ciec_password,
            "RefApplicantId": ref_id
        }
        r = requests.post(
            f"{BASE_URL}/OnboardingSatIntegration",
            headers=HEADERS, json=payload
        )
        return r.json()
    
    result = onboard_applicant(
        rfc="GAPXXXXXXXXX",
        ciec_password="CIEC_PASSWORD",
        email="contact@empresa.com",
        ref_id="loan-app-00482"
    )

    Paso 3 — Esperar Estado y Obtener applicantId

    import time
    
    def wait_for_applicant(tax_id, max_attempts=10, interval=5):
        for _ in range(max_attempts):
            r = requests.get(
                f"{BASE_URL}/get-applicants",
                headers=HEADERS,
                params={"taxId": tax_id, "onboardingStatus": "true"}
            )
            applicants = r.json().get("applicant", [])
            if applicants and applicants[0].get("status") == "Available":
                return applicants[0]["applicantId"]
            time.sleep(interval)
        raise TimeoutError("Applicant not Available within timeout")
    
    applicant_id = wait_for_applicant("GAPXXXXXXXXX")

    Modelo B — Página Hosted

    Redirige al aplicante a la página de onboarding de CRiskCo. Sin llamada API para onboarding.

    Opción 1 — Onboarding general

    Página estándar — el aplicante ingresa tu código de referencia durante el flujo:

    https://app.criskco.com/onboarding/#!/app/referrer-es

    Opción 2 — Onboarding white-label

    Requiere configuración white-label con CRiskCo:

    https://yourbrand.criskco.com/onboarding/#!/app/

    Luego consulta por refApplicantId:

    def get_applicant_by_ref(ref_id):
        r = requests.get(
            f"{BASE_URL}/get-applicants",
            headers=HEADERS,
            params={"refApplicantId": ref_id, "onboardingStatus": "true"}
        )
        applicants = r.json().get("applicant", [])
        return applicants[0] if applicants else None
    
    applicant = get_applicant_by_ref("loan-app-00482")
    if applicant and applicant["status"] == "Available":
        applicant_id = applicant["applicantId"]

    Nivel 3 — Monitoreo y Alertas (Webhooks)

    Registra suscripciones de callback y recibe payloads asíncronos cuando los datos cambian.

    Registrar Webhook

    def register_webhook(callback_url):
        r = requests.post(
            f"{BASE_URL}/Subscriptions",
            headers=HEADERS,
            json={"CallbackUrl": callback_url}
        )
        result = r.json()
        if result.get("success"):
            sub = result["ApiSubscriptionData"]
            print(f"Subscription {sub['SubscriptionId']} created: {sub['Active']}")
        return result
    
    register_webhook("https://yourdomain.com/webhooks/criskco")

    Manejar Eventos del Webhook

    from flask import Flask, request, jsonify
    import json, requests as req
    
    app = Flask(__name__)
    
    @app.route("/webhooks/criskco", methods=["GET"])
    def webhook_validation():
        return "", 200  # Respond within 2 seconds
    
    @app.route("/webhooks/criskco", methods=["POST"])
    def webhook_handler():
        event = request.json
        applicant_id = event.get("applicantId")
        file_type = event.get("FileType")
    
        if file_type == "JSON":
            payload = json.loads(event.get("APIResponse", "{}"))
        elif file_type == "JSON_LINK":
            url = event["DownloadUrlList"][0]
            payload = req.get(url).json()
    
        if payload.get("onboardingStatus") == "Available":
            print(f"Applicant {applicant_id} ready")
            blacklists = payload.get("blackLists", [])
            if blacklists:
                print(f"WARNING: Blacklist flags: {blacklists}")
        return jsonify({"received": True}), 200

    Gestionar Suscripciones

    Listar todas las suscripciones activas y eliminar por ID.

    # List all subscriptions
    def list_subscriptions():
        r = requests.get(f"{BASE_URL}/Subscriptions", headers=HEADERS)
        return r.json()
    
    # Delete a subscription by ID (POST with id query param)
    def delete_subscription(subscription_id):
        r = requests.post(
            f"{BASE_URL}/Subscriptions",
            headers=HEADERS,
            params={"id": subscription_id}
        )
        return r.json()

    Datos de Cumplimiento SAT

    Estado Fiscal

    def get_tax_status(tax_id):
        r = requests.get(
            f"{BASE_URL}/GetCompanyTaxStatus",
            headers=HEADERS,
            params={"taxId": tax_id}
        )
        company = r.json()["CompanyTaxStatus"][0]
        print(f"Status: {company['PayingTax']}")
        if company["PayingTax"] == "NEGATIVO":
            for ob in company.get("CompanyObligationsList", []):
                print(f"  Obligation: {ob['Obligation']} ({ob['Month']}/{ob['Year']})")
        return company

    FinScore Histórico

    def get_finscore_history(tax_id):
        r = requests.get(
            f"{BASE_URL}/GetHistoricalFinscore",
            headers=HEADERS,
            params={"taxId": tax_id}
        )
        scores = r.json().get("HistoricalFinscores", [])
        for s in sorted(scores, key=lambda x: (x["Year"], x["Month"])):
            print(f"{s['Year']}-{s['Month']:02d}: {s['FinScore']}")
        return scores

    Datos Financieros

    Todos los Financieros en Una Llamada

    def get_all_financials(applicant_id):
        r = requests.post(
            f"{BASE_URL}/grouping/applicant-financials",
            headers=HEADERS,
            json={"applicantId": applicant_id, "csvInJson": False}
        )
        return r.json()
    # Returns: standardized reports, raw data, documents, analytics

    Facturas AR con Filtro de Fecha

    Consulta facturas de venta filtradas por rango de fechas para análisis de ingresos.

    def get_ar_invoices(applicant_id, from_date=None, to_date=None):
        params = {}
        if from_date:
            params["fromDate"] = from_date
        if to_date:
            params["toDate"] = to_date
    
        r = requests.post(
            f"{BASE_URL}/ar-transactions/invoices",
            headers=HEADERS,
            json={"applicantId": applicant_id},
            params=params
        )
        return r.json()
    
    # Example: last 12 months of AR invoices
    invoices = get_ar_invoices("1000143693", from_date="2025-03-01", to_date="2026-03-01")

    Facturas AR con Detalle de Items

    Devuelve cabecera + renglones de cada factura en una sola llamada. El mismo payload aplica para los endpoints hermanos: /ar-transactions/invoices-items, /ap-transactions/invoices-by-uuid y /ap-transactions/invoices-items.

    def get_ar_invoices_detailed(applicant_id, from_date, to_date):
        """Returns AR invoices with header + line items in a single call.
        Sibling endpoints (same payload):
          /ar-transactions/invoices-items   - items only
          /ap-transactions/invoices-by-uuid
          /ap-transactions/invoices-items
        """
        r = requests.post(
            f"{BASE_URL}/ar-transactions/invoices-by-uuid",
            headers=HEADERS,
            json={"applicantId": applicant_id},
            params={"fromDate": from_date, "toDate": to_date}
        )
        return r.json()

    Estado Financiero del SAT

    Estado financiero consolidado de las declaraciones anuales presentadas ante el SAT — útil como cross-check de los reportes del ERP.

    def get_financial_statement(tax_id):
        """Pulls the consolidated financial statement filed with SAT."""
        r = requests.get(
            f"{BASE_URL}/financialStatement",
            headers=HEADERS,
            params={"taxId": tax_id}
        )
        return r.json()

    Monitoreo y Validación Masiva

    Disparar Monitoreo

    def trigger_monitoring(applicant_id):
        r = requests.post(
            f"{BASE_URL}/RequestMonitoring",
            headers=HEADERS,
            json={"applicantId": applicant_id, "Source": "API"}
        )
        result = r.json()
        print(f"Monitoring triggered: {result['success']}")
        return result

    Validación Masiva de RFC

    def validate_rfc_bulk(rfc_list):
        file_content = "\n".join(rfc_list).encode("utf-8")
        headers_no_ct = {k: v for k, v in HEADERS.items() if k != "Content-Type"}
        r = requests.post(
            f"{BASE_URL}/ValidateRFCBulk",
            headers=headers_no_ct,
            files={"file": ("rfcs.txt", file_content, "text/plain")}
        )
        return r.text  # Results per RFC