ASOasis - All about Tech ASOasis
Contact
AI Development

AI Invoice Processing API Tutorial: From File Upload to Accounting Sync

Build an AI-powered invoice processing pipeline with code, validation, webhooks, and ERP mapping—ready for production.

ASOasis
7 min read
AI Invoice Processing API Tutorial: From File Upload to Accounting Sync

Image used for representation purposes only.

Overview

AI-driven invoice processing turns messy PDFs, scans, and images into clean, validated data ready for your ERP or accounting system. In this tutorial, you’ll build a production-grade pipeline around a generic, vendor-neutral API. You’ll learn how to upload files, receive structured JSON with line items, validate totals, handle webhooks, and push clean data into your ledger.

What you’ll build:

  • A minimal client that uploads an invoice for extraction
  • Asynchronous job handling with polling or webhooks
  • Robust validation and duplicate detection
  • Mapping to your accounting system (chart-of-accounts, vendors, taxes)
  • Production hardening: idempotency, retries, rate limits, and security

Architecture at a Glance

A pragmatic invoice-processing flow looks like this:

  1. Ingestion: Users drag-and-drop a PDF/JPG/PNG. You store the original securely.
  2. Extraction: Send the file to an AI Invoice API (OCR + document understanding).
  3. Validation: Cross-field checks (totals, taxes, currency), vendor lookup, and business rules.
  4. Enrichment: Map vendor IDs, GL accounts, project codes, and tax codes.
  5. Review (optional): Route low-confidence fields to a human-in-the-loop UI.
  6. Export: Post the normalized payload to your ERP/AP system and archive results.

This tutorial uses example endpoints under https://api.example.com . Replace with your provider’s URLs and semantics.

Prerequisites

  • An API key from your chosen provider
  • Runtime: Python 3.10+ or Node.js 18+
  • A test invoice (single- or multi-page PDF)
  • Environment variables configured:
    • INVOICE_API_KEY — your secret token
    • INVOICE_API_BASE — e.g., https://api.example.com/v1

Quickstart: cURL

Use a simple multipart upload with an idempotency key to avoid duplicate processing on retries.

curl -X POST "$INVOICE_API_BASE/invoices:process" \
  -H "Authorization: Bearer $INVOICE_API_KEY" \
  -H "Idempotency-Key: 8f6b6f4e-5a7b-4b8f-9a1a-14a98f1a5c5b" \
  -F "file=@./samples/invoice_123.pdf" \
  -F "callback_url=https://yourapp.example.com/webhooks/invoice" \
  -F "locale_hint=en-US" \
  -F "currency_hint=USD" \
  -F "tax_region=US-CA"

Typical response (async):

{
  "job_id": "job_01HZY3XPR8T3FQ7K96G8V1Z8Z1",
  "status": "queued"
}

Poll the job or rely on webhooks for completion.

Python Client (upload + polling)

import os, time, requests

BASE = os.environ["INVOICE_API_BASE"]
KEY = os.environ["INVOICE_API_KEY"]
HEADERS = {
    "Authorization": f"Bearer {KEY}",
    "Idempotency-Key": "8f6b6f4e-5a7b-4b8f-9a1a-14a98f1a5c5b"
}

with open("./samples/invoice_123.pdf", "rb") as f:
    resp = requests.post(
        f"{BASE}/invoices:process",
        headers=HEADERS,
        files={"file": ("invoice_123.pdf", f, "application/pdf")},
        data={
            "locale_hint": "en-US",
            "currency_hint": "USD",
            "callback_url": "https://yourapp.example.com/webhooks/invoice"
        }, timeout=60
    )
    resp.raise_for_status()
    job_id = resp.json()["job_id"]

# Polling loop (use webhooks in production)
for _ in range(30):
    j = requests.get(f"{BASE}/jobs/{job_id}", headers={"Authorization": f"Bearer {KEY}"}).json()
    if j["status"] == "succeeded":
        result = requests.get(f"{BASE}/jobs/{job_id}/result", headers={"Authorization": f"Bearer {KEY}"}).json()
        print(result)
        break
    elif j["status"] == "failed":
        raise RuntimeError(j.get("error", "Job failed"))
    time.sleep(2)

Node.js Client (upload)

import fs from "node:fs";
import FormData from "form-data";
import fetch from "node-fetch";

const BASE = process.env.INVOICE_API_BASE;
const KEY = process.env.INVOICE_API_KEY;

const form = new FormData();
form.append("file", fs.createReadStream("./samples/invoice_123.pdf"));
form.append("currency_hint", "USD");
form.append("callback_url", "https://yourapp.example.com/webhooks/invoice");

const res = await fetch(`${BASE}/invoices:process`, {
  method: "POST",
  headers: {
    "Authorization": `Bearer ${KEY}`,
    "Idempotency-Key": "8f6b6f4e-5a7b-4b8f-9a1a-14a98f1a5c5b"
  },
  body: form
});
const data = await res.json();
console.log(data);

Asynchronous Jobs and Webhooks

Asynchronous processing avoids request timeouts on large or low-quality files. Provide a callback_url to receive completion events. A typical webhook delivery includes an HMAC signature you must verify.

Example webhook payload:

{
  "event": "invoice.processed",
  "job_id": "job_01HZY3XPR8T3FQ7K96G8V1Z8Z1",
  "status": "succeeded",
  "resource_url": "https://api.example.com/v1/jobs/job_01.../result",
  "timestamp": "2026-09-10T16:02:11Z"
}

Signature verification (Python, pseudo):

import hmac, hashlib

SIGNING_SECRET = os.environ["INVOICE_WEBHOOK_SECRET"]

payload = request.data  # raw bytes
sig = request.headers.get("X-Signature")  # e.g., hex HMAC-SHA256

computed = hmac.new(SIGNING_SECRET.encode(), payload, hashlib.sha256).hexdigest()
if not hmac.compare_digest(sig, computed):
    abort(401)

Response Schema Explained

Expect a normalized structure with confidence scores and optional bounding boxes. Example:

{
  "document_id": "doc_7f3...",
  "vendor": {
    "name": "Acme Supplies LLC",
    "address": "123 Market St, San Francisco, CA 94103",
    "vat_id": "US123456789",
    "confidence": 0.98
  },
  "invoice_number": {"value": "INV-10045", "confidence": 0.99},
  "issue_date": {"value": "2026-08-27", "confidence": 0.97},
  "due_date": {"value": "2026-09-26", "confidence": 0.96},
  "currency": {"value": "USD", "confidence": 0.99},
  "totals": {
    "subtotal": 1200.00,
    "tax": 96.00,
    "discount": 0.00,
    "shipping": 0.00,
    "grand_total": 1296.00
  },
  "purchase_order": "PO-7891",
  "payment_terms": "Net 30",
  "bank": {"iban": null, "account_no": "****1234"},
  "line_items": [
    {
      "description": "Printer paper A4 (500 sheets)",
      "quantity": 20,
      "unit_price": 30.0,
      "tax_rate": 0.08,
      "amount": 600.0
    },
    {
      "description": "Ink cartridges, black",
      "quantity": 12,
      "unit_price": 50.0,
      "tax_rate": 0.08,
      "amount": 600.0
    }
  ],
  "confidence_overall": 0.97,
  "pages": 2
}

Key points:

  • Use confidence to decide when to trigger human review.
  • Some providers include per-field bounding boxes for UI highlighting.
  • Multi-currency invoices should return ISO 4217 codes (e.g., “EUR”, “GBP”).

Validation and Business Rules

Always verify arithmetic and required fields before export.

from decimal import Decimal, ROUND_HALF_UP

def validate_invoice(doc: dict) -> list[str]:
    errs = []
    req = ["vendor", "invoice_number", "issue_date", "currency", "line_items", "totals"]
    for f in req:
        if f not in doc or doc[f] in (None, ""):
            errs.append(f"Missing required field: {f}")

    items_total = sum(Decimal(str(i["amount"])) for i in doc.get("line_items", []))
    subtotal = Decimal(str(doc["totals"]["subtotal"]))
    tax = Decimal(str(doc["totals"]["tax"]))
    discount = Decimal(str(doc["totals"].get("discount", 0)))
    shipping = Decimal(str(doc["totals"].get("shipping", 0)))
    grand = Decimal(str(doc["totals"]["grand_total"]))

    # Recompute and compare with 2-decimal rounding
    recomputed = (items_total - discount + shipping + tax).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    if subtotal != items_total.quantize(Decimal("0.01")):
        errs.append("Subtotal does not equal sum of line items")
    if grand != recomputed:
        errs.append("Grand total mismatch")

    return errs

Other checks:

  • Currency format and symbol agree with currency (ISO 4217)
  • Dates are valid and due_date ≥ issue_date
  • Vendor exists in your vendor master; otherwise create or flag
  • PO present when your policy requires 3-way match (PO, receipt, invoice)

Idempotency and Duplicate Detection

Idempotency prevents accidental re-processing when network retries occur. Send an Idempotency-Key header with a stable UUID per source file. The server should return the original result for repeated keys.

To catch duplicates across time, compute a fingerprint:

  • vendor_name + invoice_number + issue_date + grand_total
  • Store a SHA-256 hash and reject repeats within a policy window (e.g., 365 days)
import hashlib

def invoice_fingerprint(doc: dict) -> str:
    basis = f"{doc['vendor']['name']}|{doc['invoice_number']['value']}|{doc['issue_date']['value']}|{doc['totals']['grand_total']}"
    return hashlib.sha256(basis.encode()).hexdigest()

Mapping to Your Accounting System

Normalize fields before posting to the ERP/AP API. Example mapping stub:

def to_erp_payload(doc: dict, vendor_id: str, account_map: dict) -> dict:
    lines = []
    for it in doc["line_items"]:
        lines.append({
            "description": it["description"],
            "quantity": it["quantity"],
            "unit_amount": it["unit_price"],
            "tax_rate": it.get("tax_rate"),
            "account_code": account_map.get("office_supplies", "6000")
        })
    return {
        "vendor_id": vendor_id,
        "invoice_number": doc["invoice_number"]["value"],
        "invoice_date": doc["issue_date"]["value"],
        "due_date": doc.get("due_date", {}).get("value"),
        "currency": doc["currency"]["value"],
        "reference": doc.get("purchase_order"),
        "lines": lines,
        "total": doc["totals"]["grand_total"]
    }

Tips:

  • Maintain a vendor alias table keyed by normalized names, VAT/Tax IDs, or bank accounts.
  • Map tax rates to jurisdiction-specific codes (e.g., EU VAT, GST, US sales tax).

Errors, Retries, and Rate Limits

Handle transient errors with exponential backoff and respect Retry-After headers.

import time, math, requests

def backoff_retry(fn, max_attempts=5):
    for attempt in range(1, max_attempts + 1):
        try:
            return fn()
        except requests.HTTPError as e:
            if e.response.status_code in (429, 500, 502, 503, 504):
                ra = e.response.headers.get("Retry-After")
                delay = float(ra) if ra else min(60, 0.5 * (2 ** attempt))
                time.sleep(delay)
            else:
                raise

Also:

  • Timeouts: set client timeouts (e.g., 60s on upload)
  • Circuit breaking: stop flooding when upstream is degraded
  • Observability: log job IDs and correlate with source files

Testing and Evaluation

Build a representative test set:

  • Layout diversity: single/multi-page, tables, varying fonts, rotated pages
  • Quality: clean PDFs, scanned images at 300 DPI, light/strong skew, low contrast
  • Content: taxes, discounts, shipping, multiple currencies, credit notes

Measure quality by:

  • Field-level accuracy (precision/recall/F1) for key fields (vendor, invoice number, dates, totals)
  • Line-item correctness (row grouping and amounts)
  • End-to-end reconciliation (recomputed grand total == parsed grand total)

Use confidence thresholds to route edge cases to human review. Track false positives (incorrect but high confidence) and adjust thresholds or rules.

Security and Compliance

  • Transport: Enforce TLS 1.2+; never send keys in query strings
  • Storage: Encrypt originals and JSON results at rest; minimize retention
  • Access: Scope API keys per environment; rotate regularly
  • Webhooks: Verify signatures; use allowlists and mTLS if offered
  • PII: Redact bank accounts in logs; avoid storing unnecessary personal data
  • Compliance: Align with SOC 2, ISO 27001, and GDPR principles where applicable

Operational Tips

  • File hygiene: Prefer PDFs when available; for images, use 300 DPI and deskew/denoise
  • Locale hints: Provide locale_hint and currency_hint for better parsing
  • Page limits: Split very large PDFs if your provider imposes size/page caps
  • Versioning: Pin API versions (e.g., /v1) and track response schema changes

Production Checklist

  • Idempotency keys on all write endpoints
  • Duplicate detection and manual resolution workflow
  • Validation rules with clear user-facing error messages
  • Webhook handling with retries, signature verification, and dead-letter queues
  • Observability: metrics for success rate, latency, and review rate
  • Cost control: cap pages per invoice and auto-archive originals
  • Disaster recovery: keep originals; results are reproducible but providers evolve

Putting It All Together

  1. Accept a file upload in your app and store it securely.
  2. Call the invoice processing API with idempotency and hints.
  3. Handle completion via webhook; verify the signature.
  4. Fetch the result JSON; validate and enrich.
  5. Detect duplicates; route uncertain cases to review.
  6. Map and post to your accounting system; archive outcome.

You now have a resilient, auditable pipeline that turns raw invoices into ledger-ready entries with minimal manual effort. Extend this foundation with PO matching, fraud checks, and human-in-the-loop review to reach enterprise-grade accuracy and control.

Tags

invoice processing OCR API Python Node.js webhooks accounting automation

Related Posts