Skip to main content

Your first connector

Five-minute end-to-end: a connector that lists every organization unit currently in Nuntiq and logs how many it found. Useless in production, but it exercises every piece of the framework: settings, the API client, an entity Load, and logging.

The contract​

Every connector file lives in connectors/ and looks like this:

# connectors/get_organization.py
"""One-line summary shown in the menu."""

def run(context):
    ...
    return {'ok': True}

That's the entire contract:

  1. A top-level run(context) function.
  2. Returns a dict (or None; the framework will normalise it).

The filename (without .py) is the entrypoint key: get_supplier, get_organization, get_purchaseorder, get_receipts, get_invoice, get_invoice_status, get_payment, get_fx_rates, push_invoice or get_custom_1..N. The key decides what the run may call. A get_organization run may read and write organization units; a file with any other name may only read invoices, in the toolkit as in production. In production you paste the file into the connector under the same key.

Write it​

Create connectors/get_organization.py in your toolkit folder:

connectors/get_organization.py
"""Hello-world: count the organization units Nuntiq currently knows about."""
from lib.objects.organization import OrganizationLoad


def run(context):
    name = context.get_config('connector_name', default='hello')
    context.logger.info(f"Hello from {name}!")

    load = OrganizationLoad(context)
    units = list(load.get_all())

    context.logger.info(f"Nuntiq holds {len(units)} organization unit(s).")
    for u in units[:5]:
        context.logger.info(f"  - {u.code}: {u.name}")

    return {
        'connector_name': name,
        'organization_count': len(units),
    }

Run it​

python nuntiq.py

Pick a settings file, then get_organization from the menu.

2026-10-02 10:23:14 UTC [INFO] Hello from demo_integration!
2026-10-02 10:23:14 UTC [INFO] Nuntiq holds 2 organization unit(s).
2026-10-02 10:23:14 UTC [INFO]   - EMEA: EMEA Division
2026-10-02 10:23:14 UTC [INFO]   - NA: North America

================ result ================
{
  "connector_name": "demo_integration",
  "organization_count": 2,
  "success": true
}

The numbers come from fixtures/organization.json. Press r at the menu to restore the seed state after experimenting.

What just happened​

  1. python nuntiq.py loaded the settings file you picked from settings/ and listed every file under connectors/ that defines run().
  2. You picked get_organization. The toolkit imported connectors/get_organization.py and called run(context) with the permissions of the get_organization key.
  3. context is a JobContext, same shape as production. The only difference: context.api is the fake API client backed by fixtures/*.json.
  4. OrganizationLoad(context).get_all() translated to GET /v1/organization?limit=100 calls, following next_cursor until the last page. The fake client read them from fixtures/organization.json.
  5. Your return value was echoed back as JSON. The framework adds "success": true automatically.

In production, the same code runs unchanged: context.api becomes the real HTTP client, and OrganizationLoad.get_all() hits Nuntiq's customer API.

Make a small change​

Edit fixtures/organization.json and add a third unit. Re-run get_organization. The count goes up. Press r to reset to the seeded state.

This is the dev loop:

  1. Edit your .py file (or a fixture).
  2. Re-run python nuntiq.py → pick your entrypoint.
  3. See the result.

No restarts, no servers, no waiting.

Failing on purpose​

Add a deliberate bug:

context.logger.info(f"Found {1/0} suppliers")  # ZeroDivisionError

Run it. You'll see the traceback in stderr and the result JSON will be:

{ "success": false, "error": "division by zero", "error_type": "ZeroDivisionError" }

That's exactly what production does. The connector exits non-zero, the scheduler marks the job failed, the stderr trace lands in the job log.

What's next​