Guide for coding assistants

This page is shipped inside the toolkit zip as AGENTS.md, where tools such as Claude Code, Cursor and GitHub Copilot read it automatically. It is the same text, kept here so it can be read online and linked.

Read this before writing or changing anything in this folder. It says what a connector is, how it runs in production, where to start (a template), which helpers to use, and how this toolkit runs it on a laptop.

1. What a connector is

A connector is a set of small Python files, one per entrypoint, that move data between AP Receiving and an external system (an ERP, an accounting package, an e-invoicing network, an SFTP server). Each entrypoint:

• is one Python file with a top-level run(context) that returns a dict;
• is stored as text in AP Receiving (one code row per entrypoint, edited in the admin portal) and runs in a Docker container (Python 3.12);
• talks to AP Receiving only through context.api and the SDK in lib/.

Code that several entrypoints share goes in shared modules: shared_<name>.py files without run() (name: shared_ + up to 16 lower-case letters, digits or _). At run time they sit next to the entrypoint and are imported by file name: from shared_erp import ErpApi. No packages, no relative imports.

2. Start from a template

templates/ holds 21 ready connectors. Each folder is complete: entrypoints, shared modules, README.md (every setting and secret, what each entrypoint does, what to adapt), template.json and settings.example.json. The same templates are in the admin portal under Global → Connector Templates; a customer connector created from one starts with exactly this code.

Family	Templates (one per way of logging in)
ERP over REST: master data in, invoices out, outcome back	erp_rest_api_key, erp_rest_basic, erp_rest_oauth2_client_credentials, erp_rest_oauth2_refresh_token, erp_rest_oauth2_jwt, erp_rest_oauth1_tba (NetSuite), erp_rest_mtls, erp_rest_session (SAP Business One style)
ERP over SOAP	erp_soap_wssecurity
ERP over files: export files in, invoice files out, ack files back	erp_sftp_password, erp_sftp_key, erp_sftp_certificate, erp_s3
Invoices in (get_invoice, target capture / einvoice / json)	inbound_rest_api_key, inbound_rest_oauth2_client_credentials, inbound_rest_oauth2_refresh_token, inbound_sftp_password, inbound_sftp_key, inbound_sftp_certificate, inbound_s3
A real ERP, as a worked example	zoho_books

How to use one:

1. Pick the family, then the login method the customer's system uses.
2. Copy the folder (templates/erp_sftp_key → connectors/acme_erp/ or any folder) and copy its settings.example.json to settings/<name>.json; point connector_dir in the settings file at your folder.
3. Adapt only the marked places: the ADAPT sections of shared_mapping.py (field names) and shared_erp.py / shared_files.py (endpoints, paging, file layout). shared_conn.py changes only when the login differs.
4. Keep the entrypoint file names: the file name is the entrypoint key.
5. Run each entrypoint here (section 8), then paste the files into the customer connector in the admin portal.

Write a connector from scratch only when no template fits; build it from the same helpers.

3. The production runtime

Aspect	Production
Entrypoint keys	push_invoice, get_supplier, get_organization, get_purchaseorder, get_receipts, get_invoice, get_invoice_status, get_payment, get_fx_rates, get_custom_1..N
Container	python:3.12-slim plus requirements.txt (requests, httpx, authlib, msal, boto3, paramiko, zeep, lxml, defusedxml, pandas, openpyxl, pypdf, python-gnupg, ...) and gpg
Network	internet on tcp 443 and 22 only (another SFTP port is added to the server's reviewed list on request), from AP Receiving's fixed egress IP (customers allow-list it). No plain HTTP, FTP or FTPS. Private networks, VPNs and the cloud metadata service are unreachable
Limits	512 MB memory, 1 CPU, 4 GB disk per run; run time from the scheduler task or integration rule (60 minutes at most)
API access	a per-run token scoped to the entrypoint key (get_supplier may write suppliers but not post an invoice result); a call outside it answers 403
Result	the dict run() returns; success: false in it (or an exception) fails the run
Logging	context.logger.info/warn/error; print() also lands in the run log. Every secret value is masked as [***], but never print one anyway

Consequences: one file at a time (download, process, delete, then the next); page a big source over several runs with the delta cursor; never rely on state between runs except the delta cursor or data in AP Receiving; two runs of the same entrypoint can overlap.

4. context

def run(context):
    context.config           # settings (admin portal: connector → Settings & Secrets → settings grid)
    context.secrets          # secrets  (same tab, Secrets pane; AWS Secrets Manager)
    context.job_params       # the scheduler task's parameters, or what the engine passes to push_invoice
    context.api              # AP Receiving API (fake here)
    context.logger, context.work_dir, context.customer_number, context.entrypoint_key
    context.update_secret(name, value)   # store a secret the connector rotated itself

Read settings and secrets through Settings(context) (section 5), never directly: portal values are strings ("false", "30"), and Settings converts them and fails with a message that names the key.

5. Helpers (lib/helpers)

from lib.helpers import .... Templates are built from these; use them in your own code too. Their public names never change or disappear (new ones are only added), so code written today keeps working on later framework versions.

Helper	Use it for
Settings(context)	.str .int .bool .float .json .list .choice .date .secret .pem, .connector_name, .source_system, .param(); required unless you pass a default
RunSummary(context)	counters (.add), .warn, .error; .result() is the dict to return
ConfigError, RetryableError, PermanentError(code, message), TargetError	a bad setting; the target is down (try again later); the target refused (do not retry)
HttpClient(base_url, auth=..., context=...)	every REST call: timeouts, retries on 429/502-504/connection errors (honours Retry-After), one re-login on 401, .paginate(path, style=...) for page / offset / cursor / next_url / link_header, https only
ApiKeyAuth, BasicAuth, BearerAuth	static credentials
OAuth2ClientCredentials, OAuth2RefreshToken, OAuth2JwtBearer	OAuth 2; the refresh-token one stores a rotated refresh token itself
OAuth1Auth	OAuth 1.0a, HMAC-SHA256 (NetSuite token-based auth)
SessionAuth	login call → session cookie or header, re-login on 401, logout
client_certificate()	mTLS: certificate and key secrets → HttpClient(cert=...)
soap_client()	zeep with WS-Security and the same retry policy
TokenCache, expiry_warning()	keep an access token between runs (rotatable secret access_token_cache); warn before a secret or certificate expires
normalize_pem() / Settings.pem()	keys and certificates pasted into a one-line secret field
SftpStore, S3Store, LocalStore	one file interface (list, read_bytes, write_bytes, move, delete); SFTP with password, key, key + password or OpenSSH certificate, host key pinned; LocalStore for testing here
Inbox(store, incoming, archive=, error=)	take files one at a time: skip young files, claim, archive on success, move to error with <name>.error.txt on failure
ReadOnlyInbox	the same for a source you may not move or delete (remembers what it took in the delta cursor)
Outbox(store, folder)	write files atomically (temp name, then rename)
Pgp(context)	encrypt, decrypt, sign, verify with keys from the secrets
read_csv, read_xlsx, read_xml, write_csv	file contents to rows (encoding, BOM and delimiter detected; XML parsed safely)
pick, num, date, utc_iso, country_code, truthy, text	mapping: nested fields, Decimal amounts (decimal comma aware), dates, ISO countries
incremental(context, stream)	the delta cursor around a pull: run.since, run.seen(value), max_records for big backlogs
run_push(context, push_one)	the whole push_invoice: both callers, claims, leases, dry run; push_one(invoice) returns PushResult.posted / delivered / failed
ResultReporter(context)	get_invoice_status: report the outcome of delivered invoices (posted, failed, pending(), stale())
InvoiceIntake(context)	get_invoice: send a file to capture or e-invoice processing, or build a JSON invoice

Full docstrings are in lib/helpers/*.py; read the one you use.

6. Rules

• Secrets: never log, print or return one. Keys and certificates go in secrets (read them with Settings.pem()); OpenSSH certificates are one line.
• Rotating secrets: a secret the code writes back (refresh_token, access_token_cache) must be listed under "Secrets the connector may update" (rotatable_secrets in the settings file); the templates list theirs.
• SFTP host key: always pinned (sftp_host_key). The first run without it fails and prints the fingerprint the server offered; confirm it with the server's owner, then paste it.
• Settings names: keep the templates' names (connector_name, source_system, base_url, page_size, lookback_days, push_mode, local_folder, the sftp_* / s3_* keys). connector_name names the delta cursors: unique per connector of the customer.
• Master data: give every record its id in your system (source_record_id, PO source_document_id, PO line source_line_id, receipt source_receipt_line_id); full records with new() + save_all(), a full extract with save_all(full_load=True) (removes your records not sent), partial updates with patch_all().
• Push, sync or async (push_mode): posted when the target books the invoice on creation; delivered when it decides later (a file dropped on SFTP, an approval queue). A delivered invoice waits; get_invoice_status reports the outcome, matched on the external id stored at delivery. Never post twice: look the document up in the target first (the templates do).
• Status runs report only invoices this connector delivered; no answer yet means leave the invoice alone. An engine-pushed invoice without an answer before its integration rule's timeout is failed by AP Receiving.
• Target down: raise RetryableError (or let HttpClient raise its retryable TargetError): claimed invoices go back, files stay for the next run.

7. Entrypoints and push_invoice

Key	Direction	Usual helpers
get_organization, get_supplier, get_purchaseorder, get_receipts, get_fx_rates, get_custom_N	external → AP	incremental (APIs) or Inbox (files), the *Load objects
get_payment	external → AP	PaymentLoad, lifecycle messages; may look invoices up by external id
get_invoice	external → AP	InvoiceIntake
push_invoice	AP → external	run_push
get_invoice_status	external → AP	ResultReporter

push_invoice has two callers: the process engine (job parameter invoice_token, one invoice waiting on its integration step) and a scheduler task (claims ready invoices in batches). run_push handles both; every invoice ends with a result, an acknowledgement (delivered) or a release.

SDK objects (SupplierLoad, InvoiceLoad, ...): docs/README.md indexes one page per object.

8. Testing here

python -m venv .venv && .venv\Scripts\pip install -r requirements.txt   (once)
python nuntiq.py                                  # menus
python nuntiq.py <settings name> <entrypoint>     # one run, exit code 1 on failure

The settings file supplies secrets, config, job_params, rotatable_secrets and connector_dir (the folder whose entrypoints run; default connectors/). AP Receiving is an in-memory fake over fixtures/ (r in the menu restores them); the external system is real: point the settings at its sandbox. File templates: set local_folder to a folder on this PC and put sample files in it; no server needed. Simulated: token scoping, the master-data engine, invoice leases and acknowledgements, idempotency, file type and duplicate checks. Not simulated: OCR, webhooks, rate limits. A green local run proves the logic and the field names; the first run on a real environment is still a test.

9. In the admin portal

1. Customer → Integrations → Connectors → Add, from the template (or blank). Every entrypoint arrives disabled.
2. Settings & Secrets: fill the settings, add the secrets, check "Secrets the connector may update".
3. Entrypoints and Shared code tabs: paste your adapted files (same names). Enable the entrypoints the customer uses.
4. Scheduler Tasks: type connector_dispatch per pull entrypoint (and for batch push and status runs). Integration Rules send invoices to push_invoice from the workflow; give an async push a timeout long enough for the target's answer.
5. Run once by hand and read the run output (scheduler history → View output).

10. Checklist

1. Started from the closest template; only ADAPT sections and shared_conn.py changed, or the reason is written in the README.
2. Every setting read through Settings; every secret listed in the README.
3. Pulls use incremental or an inbox; one file or one page at a time.
4. Records carry their source ids; full loads only from full extracts.
5. push_one looks the document up before creating it; sync or async chosen on purpose; get_invoice_status scheduled for async.
6. No secret in logs, results or exceptions you raise.
7. Every entrypoint run here green; then pasted, run once on the real environment, output read.