Skip to content

Write a connector

passageway is open at its edges: any source and any language, because a connector only has to speak the connector protocol. The bundled connectors are written the same way.

A connector is a program that answers three questions about a source: what exists, what is in it, and optionally what changed. It can be written in any language. This guide builds one in Python for a folder of text files.

flowchart LR
  L["list<br/>every id and its version"] --> C{"version differs<br/>from the index?"}
  C -- "new or changed" --> F["fetch<br/>the text of those ids"] --> I[("Index")]
  C -- "unchanged" --> K["keep what is indexed"]
  L -- "id no longer listed" --> D["delete from the index"]

The core calls list on every sync and fetch only for what is new or changed; list also decides what is deleted.

The connector

#!/usr/bin/env python3
"""A passageway connector for a folder of .txt files. Setting: FOLDER."""
import hashlib
import json
import os
import sys
from pathlib import Path

folder = Path(os.environ["FOLDER"]).resolve()
files = sorted(p for p in folder.rglob("*.txt") if p.is_file())


def item_id(path):
    return path.relative_to(folder).as_posix()


def version(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()[:12]


def emit(record):
    print(json.dumps(record))


mode = sys.argv[1]
if mode == "check":
    emit({"ok": True, "count": len(files)})
elif mode == "list":
    for p in files:
        emit({"id": item_id(p), "version": version(p)})
elif mode == "fetch":
    wanted = {json.loads(line)["id"] for line in sys.stdin if line.strip()}
    for p in files:
        if item_id(p) in wanted:
            emit({"id": item_id(p), "title": p.stem, "url": p.as_uri(), "version": version(p),
                  "sections": [{"heading": None, "text": p.read_text(encoding="utf-8")}]})
# "changed" is optional: print nothing, and edits show up at the next reconcile.

passageway runs it as <command> <mode>:

Mode What it must print
check {"ok": true, "count": n}
list one {"id", "version"} line for every item that exists
fetch an item record for each id it receives on stdin, for ids that still exist
changed --since T optional: items changed since the UTC time T

The same connector with the SDK

The script above shows the whole protocol, and a connector in any language can be written that way. In Python, the passageway-sdk package does the plumbing for you: it dispatches the mode, reads the ids from stdin, buffers the output so that a failed call prints nothing, exits 2 for bad settings and 1 for a failing source, validates each item before printing it, and keeps the JSON ASCII-safe. You write what is specific to the source:

#!/usr/bin/env python3
"""The folder connector with the SDK. Setting: FOLDER."""
import hashlib
from pathlib import Path

from passageway_sdk import Connector, SettingsError


class Folder(Connector):
    name = "folder connector"

    def __init__(self, env):
        self.root = Path(env.get("FOLDER", "")).resolve()
        if not self.root.is_dir():
            raise SettingsError("FOLDER must be an existing folder")

    def files(self):
        return {p.relative_to(self.root).as_posix(): p for p in sorted(self.root.rglob("*.txt"))}

    def version(self, path):
        return hashlib.sha256(path.read_bytes()).hexdigest()[:12]

    def versions(self):
        return {i: self.version(p) for i, p in self.files().items()}

    def fetch(self, ids):
        for i, p in self.files().items():
            if i in ids:
                yield {"id": i, "title": p.stem, "url": p.as_uri(), "version": self.version(p),
                       "sections": [{"heading": None, "text": p.read_text(encoding="utf-8")}]}


Folder.main()

Try it by hand with the SDK added for one run: FOLDER=notes uv run --with passageway-sdk python folder_sdk.py list. To run it under the server, install the SDK into passageway's own environment by adding --with passageway-sdk to the uv tool install command in Install, and register the connector with that environment's interpreter, $(uv tool dir)/passageway/bin/python (a connector runs with an empty environment, so the python on your PATH has neither passageway nor the SDK). In a project of your own, uv add passageway-sdk instead. The methods, the errors and the exit codes are in the SDK reference. Implement changed(self, since) as well if your source can answer it cheaply; without it, edits appear at the next reconcile.

Try it by hand

mkdir notes && echo "Lighthouses guide ships." > notes/a.txt
FOLDER=notes python folder_connector.py list
echo '{"id": "a.txt"}' | FOLDER=notes python folder_connector.py fetch

Test it

passageway ships the checks every connector must pass. Call them from a pytest test:

import sys
from pathlib import Path
from passageway.protocol import Connector
from passageway.testing.contract import check_connector

def test_folder_connector(tmp_path):
    (tmp_path / "a.txt").write_text("Lighthouses guide ships.")
    script = str(Path(__file__).with_name("folder_connector.py"))
    connector = Connector("notes", [sys.executable, script], env={"FOLDER": str(tmp_path)})
    check_connector(connector)

Register it

[[connector]]
name = "notes"
command = ["/usr/bin/python3", "/path/to/folder_connector.py"]
env = { FOLDER = "/path/to/notes" }

Rules that keep it correct

  • Ids are stable, and unique within the connector.
  • The version changes when the content changes, and only then. passageway fetches only items whose version differs from the one it has, so a version that changes on every call re-embeds everything every time.
  • list is the source of truth for deletions. Print every existing item; an item missing from list is deleted. If your listing can be incomplete, exit non-zero instead of printing less.
  • fetch returns only requested ids, and skips ids that no longer exist.
  • A failure means a non-zero exit and nothing worth keeping on stdout. passageway stores nothing from a failed call. Write diagnostics to stderr, never secrets.
  • Optional fields shape the ranking and the answer: weight (a multiplier), path (a breadcrumb), labels, links (ids of related items) and updated. See the protocol reference.
  • Secrets come in through the environment from the secrets table in passageway.toml, never as arguments.