--- title: "Code Samples — CRiskCo API" description: "Python, Node.js, and cURL code samples for integrating the CRiskCo API: onboarding, SAT data, webhooks, and more." lang: en json-ld: | [ { "@context": "https://schema.org", "@type": "TechArticle", "headline": "Code Samples — CRiskCo API", "description": "Python, Node.js, and cURL code samples for integrating the CRiskCo API: onboarding, SAT data, webhooks, and more.", "author": { "@type": "Organization", "name": "CRiskCo" }, "publisher": { "@type": "Organization", "name": "CRiskCo", "url": "https://criskco.com" }, "url": "https://criskco.com/developers/code-samples", "datePublished": "2026-03-24", "dateModified": "2026-03-25", "inLanguage": "en-US" }, { "@context": "https://schema.org", "@type": "BreadcrumbList", "itemListElement": [ { "@type": "ListItem", "position": 1, "name": "Home", "item": "https://criskco.com/" }, { "@type": "ListItem", "position": 2, "name": "Developers", "item": "https://criskco.com/developers/api-guide" }, { "@type": "ListItem", "position": 3, "name": "Code Samples", "item": "https://criskco.com/developers/code-samples" } ] } ] --- 📘 2025 Report: [Mexico Economic Review 2025 — outlook, charts, and sector signals ](/mexico-economic-review-2025)[Read](/mexico-economic-review-2025) [![CRiskCo](/assets/criskco-logo-KYPBr-8b.png)](/en) Solutions Developers [Pricing](/en/pricing) Resources Sign In ES[Book a Demo](https://meetings.hubspot.com/israel-madrid/lead-discovery) [Home](/en)[Developers](/en/developers/api-guide) Code Samples # 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. [ ### SAT Integration Hub Start here: what the SAT API exposes, how CRiskCo abstracts SOAP/CIEC, and technical FAQ. Go to page ](/en/integracion-sat-api)[ ### API Integration Guide End-to-end walkthrough: authentication, integration models (Approve / White-label / Webhook), and the full endpoint catalog. Go to page ](/en/developers/api-guide)[ You are here ### Code Samples Copy-paste snippets in Python, Node.js, and cURL for every common flow. ](/en/developers/code-samples)[ ### Tutorials Step-by-step beginner guides: first request, authentication, and monitoring. Go to page ](/en/tutorials)[ ### API Explorer Every endpoint and every field in one searchable view. Built for teams evaluating or migrating their integration. Go to page ](/en/developers/api-explorer)[ ### API Docs Complete technical reference: every endpoint, parameter, response schema, and error code. Open docs ](https://api-docs.criskco.com/) PythonNode.jscURL ### 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 ``` ### Resources - [API Integration Guide](/developers/api-guide) - [Full API reference: api-docs.criskco.com](https://api-docs.criskco.com) - [SAT service status](/sat-service-status-mexico) [Book a technical demo](https://meetings.hubspot.com/israel-madrid/lead-discovery) [![CRiskCo](/assets/criskco-logo-KYPBr-8b.png)](/en) Risk and compliance intelligence for Mexico. We connect multi-source regulatory data for reliable enterprise decisions. [+52 55 6428 4571](tel:+525564284571)[WhatsApp](https://wa.me/525564284571)[contacto@criskco.com](mailto:contacto@criskco.com) Platform - [Platform](/en/#platform) - [How It Works](/en/#how-it-works) - [Solutions](/en/#solutions) - [Pricing](/en/pricing) - [SAT Status](/en/sat-service-status-mexico) - [Satisfied Customers](/en/success-stories) - [Security](https://trust.delve.co/criskco) Developers - [SAT API Integration](/en/integracion-sat-api) - [CFSS Standard](/en/cfss) - [API Guide](/en/developers/api-guide) - [Code Samples](/en/developers/code-samples) - [API Tutorials](/en/tutorials) - [CRiskCo Labs](/en/solutions/labs) - [MCP Integration](/en/solutions/mcp) - [API Documentation](https://api-docs.criskco.com/) Company - [About](/en/about) - [Success Stories](/en/success-stories) - [Careers](/en/careers) - [Press](/en/blog) - [Contact](/en/about) © 2026 CRiskCo. All rights reserved. [Privacy Policy](/en/privacy)[Terms of Service](/en/terms) [](https://wa.me/525564284571?text=Hola%2C%20me%20gustar%C3%ADa%20conocer%20m%C3%A1s%20sobre%20CRiskCo)