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 explains what a connector is, how it runs in production, how this toolkit simulates that on a laptop, and how the files here map onto what an administrator configures in the AP Receiving admin portal.

1. What a connector is

A connector is a set of small Python scripts, one per entrypoint, that move data between AP Receiving (the AP automation platform) and an external system (an ERP, an accounting package, a procurement tool). AP Receiving runs them on a schedule or in reaction to an invoice reaching its integration step. Each script:

• is one Python file with a top-level run(context);
• is stored as text in the AP Receiving database, one row per entrypoint, pasted into the admin portal's code editor;
• runs in a Docker container (Python 3.12) that can reach the internet and the AP Receiving API, nothing else;
• talks to AP Receiving only through context.api and the SDK objects in lib/, never a database;
• returns a dict; the platform records success or failure from it.

Code that several entrypoints need (the HTTP client for the external system, config coercion, mapping helpers) goes in a shared code module: a second kind of file, named shared_<name>.py, that has no run(). In the admin portal it is added on the connector's "Shared code" tab, where the shared_ prefix is fixed and the name is yours. At run time the platform writes every shared module next to the entrypoint's handler.py and puts that directory on sys.path, so an entrypoint uses it with a plain import by file name:

from shared_zoho import ZohoBooks, SOURCE_SYSTEM, cfg_bool

Shared modules may import each other the same way. There is no package and no relative import (from . import x does not work); only import shared_<name>. The SDK in lib/ is imported as before.

2. This folder (the toolkit)

<toolkit>/
├── nuntiq.py            run this: pick a settings file, pick an entrypoint, it runs locally
├── connectors/          one .py per entrypoint; the file name is the entrypoint key
│   └── examples/        the toolkit's samples (a subfolder is not listed in the menu)
├── settings/            one .json per connector instance (secrets, config, job params)
│   └── example.json     template; every other file here is git-ignored
├── fixtures/            the fake AP Receiving tenant: one JSON list per table
│   ├── _seed/           reset state (press r in the menu)
│   └── attachment_files/ bytes served for attachment downloads
├── lib/                 the SDK your entrypoints import (same code as production)
├── docs/                one page per SDK object; docs/README.md is the index
├── temp_work/           context.work_dir output per run (inspect, then delete)
└── _toolkit/            the sandbox itself: fake API over the fixtures. Do not import from it.

Locally, nuntiq.py imports the entrypoint file and calls run(context) with a context whose api is an in-memory fake backed by fixtures/. In production the same run(context) gets a real HTTP client. Everything else about context is identical.

3. The production runtime, precisely

Aspect	Production
Entrypoint keys	push_invoice, get_supplier, get_organization, get_purchaseorder, get_receipts, get_invoice, get_invoice_status, get_payment, and get_custom_1..N for custom connectors
Code storage	admin portal, one code version per (connector, entrypoint); every save is a new version; the server caches by version number
Container	python:3.12-slim plus the packages in requirements.txt (requests, httpx, pandas, openpyxl, pypdf, pdfplumber, paramiko, zeep, lxml, ...)
Network	outbound internet via NAT (your external system) and the AP Receiving API; private networks blocked
Invocation	python3 main.py --entrypoint <key> --connector-path <dir> --config config.json; the platform writes config.json with secrets, config, job_params, customer_number, job_token
Result	run() returns a dict; the framework adds success: true; an exception makes the run fail with the traceback in the log
Logging	context.logger.info/warn/error/debug go to stderr, which the platform stores per run; print() also lands there
Timeout	default 10 minutes per run (configurable per task / integration rule); the container is killed at the limit
Limits	512 MB memory, 1 CPU
API auth	a per-run job token, scoped to the operations the entrypoint key needs (e.g. get_supplier may read and write suppliers but may not submit an invoice result); AP Receiving API rate limits apply
Work dir	context.work_dir is a scratch folder inside the container, gone when the run ends

Consequences for code:

• Everything your entrypoint needs must be in its file, in a shared_<name> module of the same connector, in the SDK (lib/) or in the preinstalled packages. Nothing else is on the path.
• Handle rate limits of the external system yourself (sleep and retry on 429).
• Never rely on state between runs except through the delta-state service (lib/delta.py) or data stored in AP Receiving.
• Do not assume the entrypoint runs alone; two scheduled tasks can run the same entrypoint with different job_params.

4. context: where each part comes from

def run(context):
    context.entrypoint_key   # 'get_supplier'
    context.customer_number  # '10000001'
    context.secrets          # dict   <- admin portal: connector instance, Secrets pane
    context.config           # dict   <- admin portal: connector instance, Configuration Parameters
                             #           (+ per-entrypoint parameters, which win on the same key)
    context.job_params       # dict   <- admin portal: the scheduler task's Parameters JSON,
                             #           or what the process engine passes for push_invoice
    context.api              # API client (fake here, HTTP in production)
    context.logger           # info/warn/error/debug
    context.work_dir         # scratch folder path
    context.get_secret('client_id'); context.get_config('dc'); context.get_param('lookback_days', 7)

Type warning. The admin portal's parameter grids store every value as a string: "false", "1", '{"a": "b"}'. A settings file here carries real JSON types. Coerce in code (str(v).lower() in ('1','true'), int(v), json.loads(v) when a string) so both behave the same. Never test a config flag with a bare if context.get_config('flag'):.

The settings file mirrors the connector instance

settings/<name>.json:

{
  "customer_number": "10000001",
  "secrets":    { "client_id": "...", "client_secret": "...", "refresh_token": "..." },
  "config":     { "connector_name": "zoho_books_acme", "dc": "com", "organization_id": "922278034" },
  "job_params": { "lookback_days": 3650, "limit": 10 },
  "forward_structured_logs": true
}

settings key	admin portal location	read with
secrets	Customer → Connectors → instance → Settings & Secrets → Secrets pane (stored in AWS Secrets Manager, never exported)	context.get_secret()
config	same tab → Configuration Parameters (pre-filled from the definition's Default Config)	context.get_config()
job_params	Customer → Scheduler Tasks → task → Parameters JSON	context.get_param()
customer_number	the customer the instance belongs to	context.customer_number

One settings file = one connector instance. Two files let you run the same entrypoint against two instances (two ERPs, two organizations).

5. Entrypoints: what each one is for

Key	Direction	Typical flow	SDK objects
get_organization	external → AP	list companies/organizations, upsert organization units	OrganizationLoad
get_supplier	external → AP	vendors → suppliers + one location per vendor (systems without locations mirror the supplier) + address, incremental by modified time	SupplierLoad, delta_run
get_purchaseorder	external → AP	POs + lines → purchase orders / PO lines	PurchaseOrderLoad, PurchaseOrderLineLoad, delta_run
get_receipts	external → AP	goods receipts → receipts	ReceiptLoad, delta_run
get_invoice	external → AP	invoices already in the external system → AP Receiving (structured or PDF)	IngestionLoad
get_invoice_status	external → AP	status of invoices posted earlier → lifecycle messages	LifecycleMessageLoad, delta_run
get_payment	external → AP	payments → PAID / PARTIALLY_PAID lifecycle messages or payment records	LifecycleMessageLoad, PaymentLoad, delta_run
push_invoice	AP → external	post an invoice (bill, expense, journal) and report the result	InvoiceLoad
get_custom_N	any	anything else, e.g. cost centres into a customer data table	CustomerDataTableLoad

Name the local file after the key (connectors/get_supplier.py). The menu shows the first line of the file's docstring; write one.

The menu lists only files in connectors/ that define run(. Any other .py there (shared_zoho.py) is a shared module; the menu names them on one line so you can see what the entrypoints can import.

push_invoice has two callers

1. Process engine (the normal case). An invoice reaches the integration step of its workflow; an integration rule picks the connector instance; the platform runs push_invoice with job_params = {invoice_id, invoice_token, customer_number, integration_rule_id, integration_rule_name}. Handle exactly that invoice: InvoiceLoad.get_by_token(token), post it, then integration_result(token, success=..., ...). No claim, no acknowledge.
2. Scheduler (batch mode, optional). No invoice_token in job_params: claim() a batch of PendingIntegration invoices, acknowledge(), post, integration_result() per invoice.

Contract either way: every invoice you handle gets an integration_result, success or failure. A run that exits without one counts as failed and the invoice is routed to the failure path.

6. SDK essentials (details in docs/)

• Load classes: load = SupplierLoad(context); load.get_all(), load.search(field=value), load.new() then load.save_all(), obj.save(), load.delete_where(...). Full field lists in docs/<object>.md.
• save() writes the whole object. The API overwrites the stored row with what you send; fields you did not set become NULL. Two patterns:
  • records only your connector edits (suppliers, POs, receipts, CDT rows): build with load.new() and replace;
  • records people also edit in the portal (organization units): fetch, change only the fields your system owns, obj.save().
• External identity: set source_system (e.g. 'MY_ERP') and source_record_id on suppliers, supplier locations and organization units. The API matches on that pair before the natural key, the portal shows the row as managed by your system, and load.index_by_source('MY_ERP') returns {source_record_id: obj}. Organization units may be created without a code when they carry a source identity.
• Invoice fields are template-driven, keyed by the capture template's camelCase names: invoiceNumber, invoiceDate, dueDate, grossAmount, netAmount, taxAmount, currencyCode, supplierName, matched_supplier_number, matched_organization_code, orderNumber1, ext_reference_1..5; lines: productName, quantity, unitPrice, netAmount. Read with invoice.get_field('grossAmount'). Enrichment rules copy supplier-location or organization custom fields into ext_reference_N, which is how master-data settings (a default GL account, a posting mode) reach the invoice header.
• Incremental sync: with delta_run(context, connector_name, stream, cursor_type='timestamp_utc') as r: gives r.lower_bound_cursor; call r.set_new_cursor(...) before the block ends; an exception leaves the cursor untouched. connector_name comes from config, so two instances keep separate cursors.
• Lifecycle messages: LifecycleMessageLoad(context).new(), set code (PAID, APPROVED, ...), post_to(token) or post_by_reference(invoice_number, supplier_code); pass an idempotency_key from the external event id.
• Errors: from lib.api_client import ApiError; e.status_code.

7. Setting a connector up in the admin portal

Two levels: a template (global: the code as it leaves the developer) and a customer connector (per customer: its own copy of the code plus settings and secrets). Creating a customer connector from a template copies the template's default config, every entrypoint and every shared module into the customer as version 1. From then on the code is the customer's; later template versions do not change it. A customer connector can also start blank, with no code, when the work is specific to one customer.

A. Template (once per connector, Global → Connector Templates)

1. New template: connector key (stable, e.g. zoho_books), display name, version, description.
2. Default Config grid: every config key your code reads, with a default or blank value. Customer connectors start from this.
3. Shared code tab: for each connectors/shared_<name>.py, add a module with that name and paste the file.
4. Entrypoints & Code tab: for each entrypoint you implemented, select the key and paste the whole file from connectors/<key>.py. Save. Each save is a new template version; the next customer connector created from the template gets the newest.

B. Customer connector (Customer → Connectors → Add)

1. Start from: Template (pick one) or Blank. Give it a name. The description field is for implementation notes to the next consultant.
2. Settings & Secrets: fill the Configuration Parameters (the template's defaults appear; set the customer-specific ones such as the external organization id); Load Secrets, add one row per secret key your code reads (client_id, client_secret, refresh_token, ...), save.
3. Entrypoints tab: every entrypoint arrives disabled; enable only the ones this customer uses. Enabling push_invoice makes the connector a candidate for integration rules. The code of each entrypoint is editable here; customer-specific changes go in this copy.
4. Shared code tab: the copied modules, editable the same way.
5. Enable the connector itself.

C. Running it

• Customer → Scheduler Tasks → New: task type connector_dispatch, scope customer, pick the connector and the entrypoint, Parameters JSON = job_params. Run once by hand, then schedule. The task history shows connector_succeeded / connector_failed and the run's log (connector_output) with your logger lines.
• push_invoice is wired through Customer → Connectors → Integration Rules: a rule with an activation expression selects the instance for an invoice at its integration step.

D. Updating code

Edit connectors/<key>.py, test with nuntiq.py, paste the file again under the same key: on the template when the change is for every future customer, on the customer connector when it is for that customer. Settings and secrets survive code updates. A shared module is updated the same way on the Shared code tab; every entrypoint picks the new version up on its next run. Renaming a module there renames the file, so update the imports too. A template change does not reach existing customer connectors; paste it there as well when they need it.

8. Testing here

python -m venv .venv && .venv\Scripts\pip install -r requirements.txt   (once)
python nuntiq.py

Pick a settings file, then an entrypoint. r reseeds fixtures/ from fixtures/_seed/; edit fixtures freely to build a scenario. Attachment downloads read fixtures/attachment_files/<token>.bin.

Not simulated: token operation scoping (locally every call is allowed), file OCR (ingest_pdf returns a stub), status-machine validation (the fake accepts more transitions than production), lifecycle paging, rate limits. A green local run means the logic and field names are right; the first run on a real environment is still a test.

For a connector to a real external system, the settings file holds real credentials and the entrypoints call the real system while AP Receiving is still the fake. That is the intended way to develop: real source, fake target, no side effects in AP Receiving.

9. Checklist for implementing an entrypoint

1. Name the file after the key; write a docstring whose first line says what it does and which lists every config, secrets and job_params key it reads.
2. Read all settings at the top of run() with coercion; fail early with a clear message when a required one is missing.
3. Put the external-system client in a shared_<name>.py module when more than one entrypoint needs it, otherwise in the entrypoint file. Retry on 429 and on a 401 once (refresh the token). Never log secrets.
4. For inbound master data: decide who owns the record (connector only: new() + save_all(); people too: fetch, merge, save()), stamp source_system / source_record_id, wrap in delta_run if incremental.
5. For push_invoice: support both callers (section 5), be idempotent (look the document up in the external system by invoice number + vendor before creating it), attach the PDF when the external system supports it, and always submit an integration_result.
6. Log one line per record with the identifiers on both sides; return a small summary dict (counts, cursor, calls made).
7. Test locally with nuntiq.py, then paste into the template, create the customer connector from it, run the scheduler task once, read the run log.

10. Worked example in this folder: Zoho Books

connectors/get_organization.py, get_supplier.py, get_purchaseorder.py, push_invoice.py, get_payment.py. Settings in settings/zoho_live.json (git-ignored):

config key	meaning
connector_name	delta-state identity for this instance
dc	Zoho data centre suffix (com, eu, in, ...)
organization_id	the Zoho Books instance to work in (from GET /organizations); a connector-level setting, comparable to an SAP client
organization_unit_code	code of the hand-made AP Receiving unit that the primary Zoho Location adopts when get_organization syncs; for POs without a synced unit, the fallback code
sync_organization	true = get_organization syncs Zoho Locations (Settings → Locations, the companies / operating units) into organization units; false = connectivity check only. Tenants without Locations keep it false and maintain their unit by hand
location_custom_fields	JSON map of AP location custom fields to Zoho vendor custom fields; default puts the vendor's "Default Account ID" in custom field 1 and "Posting Mode" in custom field 2
paid_through_account_id	Zoho bank/card account debited by "auto charged" vendors; required for expenses
default_posting_mode, default_expense_account_id, default_tax_id, attach_pdf, auto_create_vendor, vendor_map	push_invoice behaviour; posting mode and GL account normally come from the invoice header (ext_reference_2, ext_reference_1, filled by enrichment from the supplier location's custom fields 2 and 1) with the organization unit's custom field 1 as the GL fallback

Secrets: client_id, client_secret, refresh_token from a Zoho "self client" OAuth app; tools/zoho_oauth.py exchanges a grant code for the refresh token once, after which the connector mints one-hour access tokens by itself.

Choices made there worth copying: the Zoho contact id doubles as supplier_number because Zoho has no vendor code; each vendor gets one location that mirrors it (same code and name), because Zoho has no locations; Zoho's "organization" is the whole instance (a connector parameter) while its "Locations" are what AP Receiving calls organization units, adopted by code on the first sync and matched by source_record_id afterwards, custom fields left to the customer; posting mode "invoice" creates a Bill (idempotent on vendor + bill number), "auto charged" creates an Expense (idempotent on vendor + reference number) with the PDF as receipt; dry_run in job params logs the payload and posts nothing.