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

    Code Samples

    Python, Node.js & cURL — All three integration models

    Developer resources

    Six resources, six purposes. Pick the one that matches where you are in your integration.

    Setup

    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"
    }

    Tier 1 — Verify (RFC only)

    RFC-first checks — no CIEC required: RFC ↔ razón social, taxpayer validation, e.firma certificates, PSM publications and shareholders (SIGER/RUG).

    Resolve RFC ↔ Razón Social

    Turn a company name into an RFC, or an RFC into a legal name. Both responses include the fiscal zip code, which you can pass straight into ValidateRFC.

    # Company name -> RFC
    def get_rfc_by_razon_social(razon_social):
        r = requests.post(f"{BASE_URL}/GetRfcByRazonSocial", headers=HEADERS,
                          json={"RazonSocial": razon_social})
        return r.json()  # {"razonSocial": ..., "rfc": ..., "zipcode": ...}
    
    # RFC -> company name
    def get_razon_social_by_rfc(rfc):
        r = requests.post(f"{BASE_URL}/GetRazonSocialByRfc", headers=HEADERS,
                          json={"Rfc": rfc})
        return r.json()
    
    company = get_rfc_by_razon_social("Demo Company")
    # zipcode feeds straight into ValidateRFC
    validate_rfc(company["rfc"], company["razonSocial"], company["zipcode"])

    PSM publications (Secretaría de Economía)

    Every publication the company filed: shareholder structure, meeting calls, mergers, spin-offs, dissolutions and concurso mercantil — with folio, date and the official PDF.

    def psm_publications(rfc, transaction_id=""):
        payload = {"Rfc": rfc, "TransactionId": transaction_id}
        r = requests.post(f"{BASE_URL}/PsmPublications", headers=HEADERS, json=payload)
        return r.json()
    
    data = psm_publications("XXXX000000X00", "txn-001")
    for pub in data["result"]["publications"]:
        print(pub["publicationDate"], pub["publicationType"], pub["pdfUrl"])

    PSM publications in batch

    Several RFCs in a single request. Each row is processed independently and reported through summary and rowNumber.

    def psm_publications_batch(items, batch_id=None):
        """items: list of {"TransactionId", "Rfc"}. Rows fail independently."""
        payload = {"BatchId": batch_id, "Items": items}
        r = requests.post(f"{BASE_URL}/PsmPublicationsBatch", headers=HEADERS, json=payload)
        return r.json()
    
    batch = psm_publications_batch([
        {"TransactionId": "txn-101", "Rfc": "XXXX000000X00"},
        {"TransactionId": "txn-102", "Rfc": "XXXX000000X01"},
    ], "batch-2026-001")
    
    print(batch["summary"])  # completedItems / noRecordsItems / errorItems / totalItems
    for item in batch["items"]:
        print(item["rowNumber"], item["status"], item["result"]["matchedCount"])

    Verify e.firma certificates (synchronous)

    Returns certificate status in the same response. Leave CertificateSerialNumber empty to check every certificate for the 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")

    Verify e.firma via webhook (asynchronous)

    Same check, delivered to the callback URL of the subscription referenced by 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()

    Verify e.firma in batch (1–100 RFCs)

    Submit up to 100 items in a single call; results arrive on the registered callback.

    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)

    Shareholders (SIGER/RUG) and their documents

    Two-step flow: request the shareholders first, then the registry documents using the uuid returned on the 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()

    Tier 2 — Financial Intelligence (SAT/CIEC)

    CIEC-based onboarding that unlocks CFDI, financial statements (CFSS), payroll and FinScore.

    Model A — Approve API (Full Control)

    Step 1 — Validate 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"])

    Step 2 — Onboard the Applicant

    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"
    )

    Step 3 — Poll for Status and Get 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")

    Model B — Hosted Page

    Redirect the applicant to CRiskCo's onboarding page. No API call needed for onboarding.

    Option 1 — General onboarding

    Standard page — the applicant enters your reference code during the flow:

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

    Option 2 — White-label onboarding

    Requires white-label provisioning with CRiskCo:

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

    Then poll using 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"]

    Tier 3 — Monitoring & Alerts (Webhooks)

    Register callback subscriptions and receive asynchronous payloads as data changes.

    Register Your 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")

    Handle Incoming Webhook Events

    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

    Manage Subscriptions

    List all active subscriptions and delete by 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()

    SAT Compliance Data

    Tax Status

    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

    Historical FinScore

    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

    Financial Data

    All Financials in One Call

    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

    AR Invoices with Date Filter

    Query sales invoices filtered by date range for revenue analysis.

    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")

    AR Invoices with Line-Item Detail

    Returns header + line items per invoice in one call. The same payload works for the sibling endpoints /ar-transactions/invoices-items, /ap-transactions/invoices-by-uuid and /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()

    SAT Financial Statement

    Consolidated financial statement from filed annual SAT returns — useful as a cross-check against ERP-sourced reports.

    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()

    Monitoring & Bulk Validation

    Trigger Monitoring Refresh

    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

    Bulk RFC Validation

    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