Skip to main content

Your first connector

Five-minute end-to-end: a connector that lists every supplier 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/hello.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. In production you register this key against your connector instance and the scheduler dispatches by it.

Write it

Create connectors/hello.py in your toolkit folder:

connectors/hello.py
"""Hello-world: count the suppliers Nuntiq currently knows about."""
from lib.objects.supplier import SupplierLoad


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

load = SupplierLoad(context)
suppliers = list(load.get_all())

context.logger.info(f"Nuntiq holds {len(suppliers)} supplier(s).")
for s in suppliers[:5]:
context.logger.info(f" - {s.supplier_number}: {s.supplier_name}")

return {
'connector_name': name,
'supplier_count': len(suppliers),
}

Run it

python nuntiq.py

hello will appear in the menu. Pick it.

2026-05-16 10:23:14 UTC [INFO] Hello from demo!
2026-05-16 10:23:14 UTC [INFO] Nuntiq holds 3 supplier(s).
2026-05-16 10:23:14 UTC [INFO] - SUP-001: Acme Corp
2026-05-16 10:23:14 UTC [INFO] - SUP-002: Globex Inc
2026-05-16 10:23:14 UTC [INFO] - SUP-003: Initech

================ result ================
{
"connector_name": "demo",
"supplier_count": 3,
"success": true
}

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

What just happened

  1. python nuntiq.py loaded your settings.json and discovered every .py file under connectors/.
  2. You picked hello. The toolkit imported connectors/hello.py and called run(context).
  3. context is a JobContext — same shape as production. The only difference: context.api is the fake API client backed by fixtures/*.json.
  4. SupplierLoad(context).get_all() translated to a GET /v1/supplier?page=1... call. The fake client read it from fixtures/supplier.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 SupplierLoad.get_all() hits Nuntiq's customer API.

Make a small change

Edit fixtures/supplier.json and add a fourth supplier. Re-run hello. 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" }

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