Invoice ingestion

The pull-side patterns (REST API pull, SFTP pull, SFTP push) all live in your customer's domain. They assume Nuntiq already has the invoice and you're either pulling supporting master data in or pushing the invoice out to an ERP.

This page covers the other direction: putting invoices into Nuntiq from a connector (an e-invoicing network, a scanning service, a supplier portal, a folder on SFTP). The inbound_* templates are complete get_invoice connectors for REST, SFTP and S3 sources; start there.

With the helpers: InvoiceIntake

InvoiceIntake sends each invoice the way the customer wants it read, picked by the target setting:

target	What Nuntiq does	The customer configures
capture	the PDF (or a ZIP of PDFs) goes through capture, like an e-mail to the receiving inbox; other files ride along as attachments	a receiving inbox (optional: the default inbox is used) and capture templates
einvoice	the data file (UBL, XML, EDI, CSV, JSON) is read by the customer's inbound file formats; PDFs and images ride along, linked to the invoice whose data file names them	an inbound file format for the sender's layout
json	the connector builds the invoice (header, lines, addresses) and it is imported as it is	nothing extra; json_workflow decides how far Nuntiq processes it

get_invoice.py

from lib.helpers import Settings, HttpClient, ApiKeyAuth, InvoiceIntake, incremental


def run(context):
    s = Settings(context)
    portal = HttpClient(s.str('base_url'), auth=ApiKeyAuth(s.secret('api_key')), context=context)
    intake = InvoiceIntake(context)
    with incremental(context, 'invoices') as run:
        for doc in portal.paginate('documents', params={'since': run.since}, items='data'):
            pdf = portal.request('GET', f"documents/{doc['id']}/pdf", raw=True).content
            if intake.target == 'json':
                inv = intake.new_invoice()
                inv.header.invoice_number = doc['number']
                inv.header.gross_amount = doc['total']
                inv.attach_image(filename=f"{doc['number']}.pdf", content=pdf)
                intake.submit_invoice(inv)
            else:
                intake.submit_file(f"{doc['number']}.pdf", pdf)
            run.seen(doc['modifiedAt'])
    return intake.summary.result(cursor=run.new_cursor)

The same code works for a folder on SFTP or S3 with an Inbox instead of the API pull: see SFTP pull. The rest of this page is the object underneath, IngestionLoad, for when you need more control.

Underneath: IngestionLoad

Two flows, both wrapped in IngestionLoad.

Flow	Use when
IngestionLoad.ingest_pdf(...)	You have a PDF / image / ZIP and want Nuntiq's standard pipeline (OCR → template extraction → enrichment). Same path as email intake.
IngestionLoad.new_invoice() + .submit()	You already have parsed invoice data (from an XML / EDI feed) plus optionally the original PDF, and you want to import the structured fields directly without OCR.

Files as they are (a PDF to capture, an e-invoice to map, with their attachments) go through context.api.upload_files(): see Sending files: capture or e-invoice. This page is about IngestionLoad.

The raw context.api.upload_source_documents / upload_invoice_attachments / invoice_import helpers are still available if you need them, but IngestionLoad is the recommended entry point: it handles the upload-then-import token-threading dance for you, validates inputs before the HTTP call, and returns typed results.

Run this code under the get_invoice entrypoint key. Its job token may upload source documents and invoice attachments and import invoices; other keys get a 403. In the toolkit, name the file connectors/get_invoice.py: the file name is the key.

Pattern 1: Drop a PDF into the standard pipeline

The simplest case. You have a file, you want Nuntiq to process it the same way it processes emailed invoices. Nuntiq runs OCR, picks a template, extracts fields, runs validation + enrichment, ends up with a fully-processed invoice at status 100 (or whichever stop you configure).

connectors/get_invoice.py

"""Pull invoice PDFs from a supplier portal and ingest them."""
import requests

from lib.objects.ingestion import IngestionLoad


def run(context):
    portal_url = context.get_config('portal_url')
    api_token = context.get_secret('portal_api_token')
    receiving_inbox = context.get_config('receiving_inbox')   # e.g. 'invoices@acme.apreceiving.com'

    ingest = IngestionLoad(context)

    # Fetch the list of new invoices from the supplier portal
    headers = {'Authorization': f'Bearer {api_token}'}
    resp = requests.get(f'{portal_url}/new-invoices', headers=headers, timeout=30)
    resp.raise_for_status()

    uploaded = 0
    for inv in resp.json()['invoices']:
        # Download the PDF
        pdf_resp = requests.get(inv['pdf_url'], headers=headers, timeout=60)
        pdf_resp.raise_for_status()

        # Ingest
        result = ingest.ingest_pdf(
            filename=inv['filename'],
            content=pdf_resp.content,
            receiving_inbox=receiving_inbox,
            supplier_code=inv.get('supplier_code'),
        )

        if result.duplicate:
            # Sent within the last 24 hours already: nothing was processed again.
            context.logger.info(
                f"{inv['filename']} already ingested as source_document_id={result.duplicate_of}",
                step='ingest',
            )
            continue

        context.logger.info(
            f"Ingested {inv['filename']} → source_document_id={result.source_document_id}",
            step='ingest',
            detail={'supplier_code': inv.get('supplier_code')},
        )
        uploaded += 1

    return {'uploaded': uploaded}

What receiving_inbox does

It's the address Nuntiq uses for template lookup. Different receiving inboxes can map to different templates, e.g. xml@acme.apreceiving.com might use a structured-data template, invoices@acme.apreceiving.com uses the standard OCR template. Default to your customer's primary inbox unless you have a specific routing reason.

Hints

• company_code and supplier_code are both optional. Pass them when you know: the worker uses them to skip the AI-match step for those fields.
• ingest_files() sends up to 20 files as one source document.
• The API reads the file type from the bytes, not the name. PDFs are captured, loose or in a ZIP. Next to them, loose or in the ZIP, PNG, JPEG, TIFF, GIF, DOCX, XLSX, PPTX, DOC, XLS, CSV, TXT, XML, JSON, EML and MSG files are kept as attachments of the source document, the way the attachments of an email are: not captured, shown with the source document and attached to every invoice captured from the upload. Anything else (executables, scripts, HTML, macro-enabled Office files, an archive inside the ZIP) is refused with 415 unsupported_file_type before anything is stored. An upload without a PDF is refused with 422 no_processable_file.
• ZIP files are opened under limits: at most 200 entries and 250 MB unpacked, no encrypted entries (422 zip_rejected).
• Files up to 50 MB go as a normal upload. A larger file (up to 500 MB) goes through a presigned upload; the framework does that by itself.

Duplicates and processing status

The same files from the same connector within 24 hours are not processed again. result.duplicate is True and result.duplicate_of names the earlier source document. Pass allow_duplicate=True to process them anyway.

To follow the processing:

status = ingest.status(result)        # or a source document id / source token
print(status['status'])               # received, processing, completed, failed, timeout
for invoice in status.get('invoices', []):
    print(invoice)

Pattern 2: Submit structured invoice + attach the original PDF

When you already have parsed invoice data (an XML feed, an EDI 810, a JSON payload from an upstream system) AND the original PDF, this is the right pattern. The builder handles the upload-then-import dance; you never touch document tokens directly.

connectors/get_invoice.py

from lib.objects.ingestion import IngestionLoad


def run(context):
    receiving_inbox = context.get_config('receiving_inbox')
    ingest = IngestionLoad(context)

    imported = 0
    for raw in fetch_structured_invoices():           # your code
        inv = ingest.new_invoice(
            workflow='payment_portal',
            receiving_inbox=receiving_inbox,
        )

        # Header: set whatever fields you have
        inv.header.invoice_number = raw['number']
        inv.header.invoice_date = raw['date']         # ISO date string
        inv.header.due_date = raw['due_date']
        inv.header.currency_code = raw['currency']
        inv.header.net_amount = raw['net']
        inv.header.tax_amount = raw['tax']
        inv.header.gross_amount = raw['gross']
        inv.header.supplier_name = raw['supplier']
        inv.header.order_number_1 = raw.get('po_number')

        # Lines
        for i, line in enumerate(raw['lines']):
            l = inv.new_line()
            l.invoice_line_number = i + 1
            l.product_name = line['desc']
            l.quantity = line['qty']
            l.unit_price = line['price']
            l.net_amount = line['net']
            l.tax_amount = line.get('tax', 0)

        # Addresses: address_type must be REMITTO | SUPPLIER | SHIPTO | BILLTO
        addr = inv.new_address('SUPPLIER')
        addr.street = raw['supplier_addr']['street']
        addr.city = raw['supplier_addr']['city']
        addr.country = raw['supplier_addr']['country']

        # Optional tax breakdown
        tax = inv.new_tax()
        tax.tax_name = 'VAT'
        tax.tax_rate = 21
        tax.tax_amount = raw['tax']

        # Attach the original PDF (and any supporting docs)
        inv.attach_image(filename=raw['pdf_name'], content=raw['pdf_bytes'])
        for att in raw.get('supporting_docs', []):
            inv.attach_file(filename=att['name'], content=att['bytes'])

        # Submit: uploads attachments first, then POSTs the structured invoice
        result = inv.submit()

        context.logger.info(
            f"Imported {raw['number']} → invoice_token={result.invoice_token}",
            step='import',
            detail={
                'invoice_status': result.invoice_status,
                'template_id': result.template_id,
            },
        )
        imported += 1

    return {'imported': imported}

Workflow choice

workflow controls how far Nuntiq processes the invoice on import:

Value	Resulting status	Use when
payment_portal (default)	100 (Processed)	The invoice arrives fully verified; skip enrichment entirely
basic_enrichment	10	Run light enrichment but no AI
full_enrichment	1	Run the full pipeline (AI match, validation)

If you trust the upstream data, payment_portal is faster: it sends the invoice straight to the payable stage without re-running OCR or AI match.

Attachment rules

• inv.attach_image(filename, content): at most one per invoice. This is the PDF the AP team will view in the Nuntiq viewer. Calling attach_image() twice raises ValueError.
• inv.attach_file(filename, content): 0..N supporting documents (delivery notes, supporting POs, etc.). Visible to the AP team but not the primary view.
• If you don't pass any IMAGE, the invoice has no viewable PDF, so the AP team sees the structured fields only. Usually a poor UX. Aim to always include at least one IMAGE.

Pattern 3: Structured invoice only, no PDF

Same as Pattern 2, just skip the attach_image() and attach_file() calls. The invoice gets created with no viewable image. Make sure the AP team is OK with that before you build a pipeline that does it at scale.

What submit() checks

The attachments are uploaded first (PDF, PNG, JPEG, TIFF, XML and ZIPs of those; other types answer 415 unsupported_file_type). The import then refuses a document token it does not know or that was already used, with 422 unknown_attachment_token, before it creates anything.

What submit() returns

IngestedInvoice.submit() returns an InvoiceImportResult:

Field	Type	Notes
invoice_token	str (UUID)	Pass to InvoiceLoad.get_by_token()
invoice_id	int	Internal numeric id
source_document_id	int	Underlying source-document id
workflow	str	The workflow used
invoice_status	int	Final numeric invoice status (e.g. 100 = Processed)
source_document_status	int	Final source-document status
template_id / template_type	int / str	Template Nuntiq picked + whether it's 'global' or 'customer'
attachments_linked	int	Count of attachments tied to the invoice
receiving_inbox	str	The inbox used for template lookup

IngestionLoad.ingest_pdf() / ingest_files() return a smaller SourceDocumentResult:

Field	Type	Notes
source_document_id	int	The new source document
source_token	str (UUID)	Its token; status() takes it
files_uploaded	int	Files stored, ZIP entries included
pdf_files_queued	int	PDFs queued for capture
upload_session_id	str (UUID)	Groups the upload for log searches
duplicate	bool	True when the same files were sent within 24 hours; nothing was processed
duplicate_of	int	The earlier source document, when duplicate

Both objects carry a _raw attribute with the full HTTP response if you need to read fields the wrappers haven't surfaced.

Error handling

Local validation (workflow value, address_type value, max-one-image) raises ValueError before any HTTP call, so you find out at code-write time.

ingest_pdf() and submit() raise ApiError on transport / HTTP failure. See ApiClient → Error handling.

Common server-side rejections:

Status	Code	Cause
400		Validation of the invoice body failed, or the receiving_inbox cannot be resolved
413		A file over 50 MB sent as a normal upload, or too many files
415	unsupported_file_type	A file type the endpoint does not take; nothing was stored
422	zip_rejected	A ZIP over the limits or encrypted (an archive inside a ZIP is a 415)
422	no_processable_file	ingest_pdf() / ingest_files() without a PDF
422	unknown_attachment_token	An import names a document token that is unknown or already used
503	queue_failed	The source document was stored but not queued (it is marked failed); send it again

Choosing between Pattern 1 and Pattern 2

The simplest filter:

• Have a PDF only → Pattern 1.
• Have structured data only or structured data + PDF → Pattern 2.
• Have structured data and don't care about a viewable PDF → Pattern 3.

If you have both options available (structured data AND a PDF), Pattern 2 is strictly better than Pattern 1: Nuntiq skips OCR + template extraction (faster and more deterministic) and you still get the PDF for the AP team to view.

Idempotency

• ingest_pdf() / ingest_files(): the same files from the same connector within 24 hours answer the earlier source document (duplicate) and are not processed again, unless you pass allow_duplicate=True.
• inv.submit(): an import has no duplicate check on its content. Sending the same invoice twice creates two invoices. The API takes an Idempotency-Key header on POST /v1/invoice-import (the same key and body within 24 hours replay the first answer); submit() does not send one, so when you need it, send the import yourself: context.api.post('/v1/invoice-import', data=body, extra_headers={'Idempotency-Key': key}).

Either way, track which upstream records you've already submitted in your own state (typically a delta_run cursor) and don't re-submit.

Raw helpers (escape hatch)

If you need finer control than IngestionLoad offers, the raw helpers on context.api are still available:

• context.api.upload_source_documents(files, ...): upload to /v1/source-document. Returns the raw {duplicate, data: {sourceDocumentId, source_token, ...}} envelope.
• context.api.get_source_document(id_or_token): processing status, files and invoices of a source document.
• context.api.upload_invoice_attachments(files): upload to /v1/invoice-attachments. Returns {data: {documents: [{document_token, file_name, content_type, size}, ...]}}.
• context.api.create_upload(filename, size) and context.api.put_presigned(upload, content): a presigned upload for one file up to 500 MB; send the returned id as upload_ids in a JSON body. The two upload helpers above do this by themselves for files over 50 MB.
• context.api.invoice_import(body): POST to /v1/invoice-import with the body shape documented in the Customer API reference.

The IngestionLoad implementation in lib/objects/ingestion.py is the canonical example of how to thread tokens between them.

What's next

• IngestionLoad reference: field-level reference for the builder.
• Invoice claim flow: once an invoice is ingested, this is how downstream connectors push it onward to an ERP.
• Lifecycle messages: once your upstream system knows the invoice's Nuntiq token, write lifecycle events back so the Nuntiq timeline reflects what's happening in your system.