Skip to content

Authoring a Connector

How to build a Connector plugin — SourceConnector and/or ActionConnector — against connector_base.

A Connector links the platform to one external system. It can:

  • Read data into the entity graph (SourceConnector)
  • Write data back (ActionConnector)
  • Both at once — a GitLab connector is typically source + action.

Package skeleton

packages/connector_<name>/
├── pyproject.toml
├── src/connector_<name>/
│   ├── __init__.py
│   ├── manifest.yaml
│   ├── connector.py          # main implementation
│   ├── webhook.py            # optional — webhook payload parsing
│   └── schema.py             # optional — pydantic models for the system's API
└── tests/
    ├── conftest.py
    └── test_connector.py

pyproject.toml:

[project]
name = "connector_my"
version = "0.1.0"
description = "My SaaS connector."
requires-python = ">=3.12"
dependencies = [
    "connector_base",
    "httpx>=0.27",
    "shared_types",
]

[project.entry-points."knowledge_os.connectors"]
my = "connector_my:MyConnector"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/connector_my"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

The manifest

A connector manifest has two top-level keys: plugin: (metadata, identical to skills/packs) and connector: (the connector-specific definition).

plugin:
  id: connector_my
  type: connector
  version: "0.1.0"
  kernel_version: ">=0.1.0,<2.0"
  display:
    name: My SaaS
    description: Ingest tickets and post comments to MySaaS.
    icon: assets/icon.svg
    category: ticketing
    tags: [tickets, comments]
  author:
    name: Threadory
    email: dev@example.com

connector:
  auth:
    type: api_key
    fields:
      - id: api_key
        label: API key
        type: secret
        required: true
        help: Generate at https://mysaas.example.com/settings/api
        validation:
          pattern: "^ms_[a-zA-Z0-9]{32}$"
      - id: webhook_secret
        label: Webhook secret
        type: secret
        required: true

  config_schema:
    type: object
    properties:
      base_url:
        type: string
        default: https://api.mysaas.example.com
        title: API base URL
      project_filter:
        type: array
        items: { type: string }
        title: Projects to sync
    required: [base_url, project_filter]
    additionalProperties: false

  provides:
    sources:
      - kind: issue
        capabilities: [read_thread, read_attachments]
        sync:
          mode: incremental
          strategy: webhook + polling
          interval_minutes: 30
    actions:
      - kind: issue
        capabilities: [post_comment, update_state]

  webhooks:
    - path: /webhooks/my/{project_id}
      events: [issue.created, issue.updated, issue.commented]
      signature_validation: hmac-sha256
      secret_field: webhook_secret

  health:
    endpoint_test: GET {base_url}/v1/me
    expected: { status: 200 }

Notable rules enforced by ConnectorManifest:

  • auth is a discriminated union on type: one of api_key, basic, webhook_secret, oauth2.
  • connector.provides must declare at least one source or action (empty connectors are rejected).
  • webhooks[].secret_field must reference a field declared in auth.fields; mismatched names raise at manifest load.
  • webhooks[].path must start with /.
  • config_schema.type must be "object".

Load and validate the manifest in code:

from pathlib import Path
from connector_base import load_manifest

manifest = load_manifest(Path(__file__).parent / "manifest.yaml")

The SourceConnector Protocol

Source connectors ingest data. The Protocol lives in connector_base.protocol:

class SourceConnector(Protocol):
    manifest: ConnectorManifest

    async def setup(
        self,
        project: Project,
        config: dict[str, Any],
        credentials: dict[str, Any],
    ) -> None: ...

    async def teardown(self, project: Project) -> None: ...

    async def health_check(self) -> HealthStatus: ...

    async def sync_full(self, project: Project, kind: str | None = None) -> SyncReport: ...

    async def sync_incremental(
        self, project: Project, since: datetime, kind: str | None = None,
    ) -> SyncReport: ...

    async def handle_webhook(
        self, path: str, headers: dict[str, str], body: bytes,
    ) -> WebhookResult: ...

    async def normalize(self, raw: dict[str, Any]) -> NormalizedEntity: ...

    async def extract_link_candidates(self, entity: Entity) -> list[LinkCandidate]: ...

Method contract:

Method Called when Must be idempotent?
setup Plugin enabled / configured for a project Yes
teardown Plugin disabled or uninstalled Yes
health_check UI test button + periodic kernel check Yes (read-only)
sync_full First connect, admin re-sync, and periodically (below) Yes (upsert)
sync_incremental Scheduler tick using the connector's interval Yes (upsert)
handle_webhook Inbound webhook hit on a registered path Yes
normalize Per-payload conversion to a NormalizedEntity Pure function
extract_link_candidates After ingestion to discover edges Pure function

What a sync report owes, and when

Two fields on SyncReport are read as claims about how far the walk got, and the kernel acts on both. Return them only when they are true.

next_cursor means everything up to here is accounted for. The kernel stores it and hands it back on the following run — as since when it parses as a timestamp, otherwise verbatim as cursor, the opaque page-token channel — so a cursor left by a walk that stopped early steps over whatever that walk never reached — permanently, since the skipped items are older than the newest it saw. Return it from any walk that reached the end, full or incremental; withhold it from any walk that did not, whether it broke (a page ceiling, a truncation) or stopped on purpose (it found the one item it was sent for).

seen_source_ids is an inventory, and the kernel deletes what an inventory omits: rows in scope that the list does not mention are tombstoned. So it belongs only to a walk that enumerated everything — never to an incremental pass, which sees a delta, and never to a truncated one, whose prefix would read as "the rest is gone". Pair it with source_system and seen_entity_kinds so the diff is scoped to the kinds this path actually produces.

The two are independent. A full pass that finished owes both; a full pass that was cut short owes neither and should say status="partial"; an incremental pass that finished owes the cursor and no inventory. A deliberate early stop is still completed — it did what it was asked — but it is a prefix, so it owes neither.

The kernel writes its own keys into install.config — the cursor, the source-tree cursor, and last_full_sync_at. If your setup validates config with extra="forbid", drop them first: import KERNEL_MANAGED_CONFIG_KEYS from shared_types rather than listing them, because setup runs on every sync and a key you have not heard of turns into a connector that fails every tick forever.

Why the kernel still asks for a full pass now and then: only an inventory can reveal a deletion, so an installation that went incremental and never came back would keep every entity ever removed upstream. The kernel alternates on last_full_sync_at, and a partial pass does not refresh that stamp.

If your full pass never returns an inventory — because "full" means a recent window rather than everything, as it does for tl;dv, Zoom and Meet — you are exempt: the kernel records that and stops asking. You get the cheap path from the first completed pass onward.

Protocol is structural; runtime_checkable is intentionally not set. The runtime trusts mypy + the contract tests to verify shape — no isinstance checks at runtime.

The ActionConnector Protocol

Action connectors write data back:

class ActionConnector(Protocol):
    manifest: ConnectorManifest

    async def can_execute(self, action: Action, target: EntityRef) -> bool:
        """Pre-flight: do we still have the credentials and permissions?"""

    async def execute(
        self,
        action: Action,
        target: EntityRef,
        payload: dict[str, Any],
    ) -> ActionResult:
        """Perform the action. Must be idempotent on payload['idempotency_key']."""

The same Python class can satisfy both Protocols — declare both surface areas in provides.

Auth methods

Four type values are supported, each enforcing its own field shape:

api_key

Simple bearer token / API key. Declare one or more secret fields:

auth:
  type: api_key
  fields:
    - { id: api_key, label: API key, type: secret, required: true }

The runtime renders a form, encrypts the secret, and passes it to setup in credentials.

basic

Username + password. Same shape as api_key; expect at least one id: username field and one id: password field.

webhook_secret

Used for connectors that only receive webhooks (no outbound API). Declares the shared secret used to verify HMAC signatures. The kernel never makes outbound requests on the plugin's behalf.

oauth2

For OAuth2 with PKCE:

auth:
  type: oauth2
  oauth2:
    authorize_url: https://gitlab.com/oauth/authorize
    token_url: https://gitlab.com/oauth/token
    scopes: [read_api, write_repository]
    pkce: true

Use the helpers in connector_base.helpers.oauth to build the authorize URL and exchange the returned code for tokens:

from connector_base import OAuth2State, build_authorize_url, exchange_code

state = OAuth2State.generate()
url = build_authorize_url(
    authorize_url=manifest.connector.auth.oauth2.authorize_url,
    client_id=client_id,
    redirect_uri="https://app.example.com/oauth/callback",
    scopes=manifest.connector.auth.oauth2.scopes,
    oauth_state=state,
)

# ... user authorizes, app receives `code` ...

tokens = exchange_code(
    token_url=manifest.connector.auth.oauth2.token_url,
    client_id=client_id,
    code=code,
    redirect_uri="https://app.example.com/oauth/callback",
    oauth_state=state,
    client_secret=client_secret,  # if confidential client
)

OAuth2State.generate() produces a CSRF token + a base64url-safe 48-byte PKCE verifier. pkce_challenge computes the S256 code challenge automatically.

Webhook signature verification

If your manifest sets signature_validation: hmac-sha256, validate every inbound request with verify_hmac_sha256 — it does timing-safe comparison and accepts both sha256=<hex> and bare <hex> signature header formats:

from connector_base import verify_hmac_sha256

async def handle_webhook(
    self,
    path: str,
    headers: dict[str, str],
    body: bytes,
) -> WebhookResult:
    signature = headers.get("X-MySaaS-Signature")
    if not verify_hmac_sha256(self._webhook_secret, body, signature):
        return WebhookResult(status="rejected", reason="invalid_signature")
    # ... parse body, emit events ...

Idempotency

Sync and action operations must be idempotent. Use idempotency_key to derive stable hashes:

from connector_base import idempotency_key

key = idempotency_key("post_comment", target.id, payload["body"])
# Stable SHA-256 across dict ordering, list order preserved, UUID/datetime normalized.

The kernel persists the key for short-term deduplication so that a retried execute call does not create a duplicate comment. Always pass through the idempotency_key your caller supplied if there is one; otherwise compute your own from action-stable inputs.

Worked example — calling an external API

from __future__ import annotations

from datetime import datetime
from pathlib import Path
from typing import Any

import httpx
from connector_base import (
    ConnectorManifest,
    idempotency_key,
    load_manifest,
    verify_hmac_sha256,
)
from shared_types import (
    Action,
    ActionResult,
    Entity,
    EntityRef,
    HealthStatus,
    LinkCandidate,
    NormalizedEntity,
    Project,
    SyncReport,
    WebhookResult,
)


class MyConnector:
    """SourceConnector + ActionConnector for MySaaS."""

    manifest: ConnectorManifest

    def __init__(self) -> None:
        self.manifest = load_manifest(Path(__file__).parent / "manifest.yaml")
        self._client: httpx.AsyncClient | None = None
        self._webhook_secret: str | None = None
        self._base_url: str | None = None

    # ------------------------------------------------------------------ setup
    async def setup(
        self,
        project: Project,
        config: dict[str, Any],
        credentials: dict[str, Any],
    ) -> None:
        self._base_url = config["base_url"]
        self._webhook_secret = credentials["webhook_secret"]
        self._client = httpx.AsyncClient(
            base_url=self._base_url,
            headers={"Authorization": f"Bearer {credentials['api_key']}"},
            timeout=httpx.Timeout(20.0),
        )

    async def teardown(self, project: Project) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    # ------------------------------------------------------------------ health
    async def health_check(self) -> HealthStatus:
        assert self._client is not None
        try:
            resp = await self._client.get("/v1/me")
            resp.raise_for_status()
        except httpx.HTTPError as exc:
            return HealthStatus(status="error", message=str(exc))
        return HealthStatus(status="ok")

    # ------------------------------------------------------------------ sync
    async def sync_full(self, project: Project, kind: str | None = None) -> SyncReport:
        return await self._sync(project, since=None, kind=kind)

    async def sync_incremental(
        self, project: Project, since: datetime, kind: str | None = None,
    ) -> SyncReport:
        return await self._sync(project, since=since, kind=kind)

    async def _sync(
        self, project: Project, since: datetime | None, kind: str | None,
    ) -> SyncReport:
        assert self._client is not None
        params: dict[str, Any] = {}
        if since is not None:
            params["updated_after"] = since.isoformat()
        resp = await self._client.get("/v1/issues", params=params)
        resp.raise_for_status()
        # ... convert via self.normalize(); persist via runtime; return SyncReport
        return SyncReport(entities_ingested=0, errors=[])

    # ------------------------------------------------------------------ webhook
    async def handle_webhook(
        self, path: str, headers: dict[str, str], body: bytes,
    ) -> WebhookResult:
        assert self._webhook_secret is not None
        signature = headers.get("X-MySaaS-Signature")
        if not verify_hmac_sha256(self._webhook_secret, body, signature):
            return WebhookResult(status="rejected", reason="invalid_signature")
        # ... parse JSON, dispatch events
        return WebhookResult(status="accepted")

    # ------------------------------------------------------------------ normalize
    async def normalize(self, raw: dict[str, Any]) -> NormalizedEntity:
        ...

    async def extract_link_candidates(self, entity: Entity) -> list[LinkCandidate]:
        return []

    # ------------------------------------------------------------------ actions
    async def can_execute(self, action: Action, target: EntityRef) -> bool:
        return self._client is not None

    async def execute(
        self,
        action: Action,
        target: EntityRef,
        payload: dict[str, Any],
    ) -> ActionResult:
        assert self._client is not None
        key = payload.get("idempotency_key") or idempotency_key(action.kind, target.id, payload)
        resp = await self._client.post(
            f"/v1/issues/{target.source_id}/comments",
            json={"body": payload["body"]},
            headers={"Idempotency-Key": key},
        )
        resp.raise_for_status()
        return ActionResult(status="completed", payload={"comment_id": resp.json()["id"]})

Testing

Mock outbound HTTP with httpx.MockTransport or respx; never hit the real API in unit tests:

import httpx
import pytest
from connector_my import MyConnector


async def test_health_check_ok(monkeypatch: pytest.MonkeyPatch) -> None:
    conn = MyConnector()

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/v1/me"
        return httpx.Response(200, json={"id": "u_1"})

    monkeypatch.setattr(
        "httpx.AsyncClient",
        lambda **kw: httpx.AsyncClient(transport=httpx.MockTransport(handler), **kw),
    )
    await conn.setup(
        project=..., config={"base_url": "https://x"}, credentials={"api_key": "k", "webhook_secret": "s"},
    )
    health = await conn.health_check()
    assert health.status == "ok"

See Testing for the full pattern, including integration tests against a Postgres testcontainer.

Connector checklist

Before opening a PR, walk through:

  • [ ] manifest.yaml validates with connector_base.load_manifest.
  • [ ] Every method on SourceConnector / ActionConnector is implemented (raise NotImplementedError if the connector doesn't act in that direction, but prefer to leave the Protocol unsatisfied and not declare actions in provides).
  • [ ] All outbound calls go through httpx.AsyncClient with an explicit timeout.
  • [ ] Webhook handler verifies signatures with verify_hmac_sha256 before parsing.
  • [ ] Idempotency key flows end-to-end on actions.
  • [ ] Unit tests cover happy path + at least one auth-failure path.
  • [ ] Entry point is registered under knowledge_os.connectors.
  • [ ] kernel_version range is set conservatively.