Connect your events¶

Keel ingests envelopes, never raw rows. Every source produces the same shape, and the cartridge's handlers are the only code that reads a payload. Pick the source that matches where your events already are.

The envelope¶

One event, as the development adapter reads it from examples/events/fleet.jsonl:

{"tenant": "acme-logistics", "type": "unit.registered", "subject": "unit:MXK-4821",
 "occurred_at": "2026-09-01T09:00:00Z", "source": "fleet.vehicles",
 "payload": {"id": "MXK-4821", "plate": "MXK-4821", "model": "Freightliner Cascadia"}}

Field	Required	Meaning
`tenant`	yes	The end-customer this fact belongs to. Opaque to Keel.
`type`	yes	Dotted lowercase type from the cartridge catalog.
`subject`	yes	`<kind>:<id>` of the entity the fact is about.
`occurred_at`	yes	When it happened in the world. Timezone-aware.
`source`	yes	System of record or adapter name.
`id`	no	Generated when absent. Becomes the `evidence` of derived facts.
`recorded_at`	no	When Keel learned it. Defaults to now.
`payload`	no	Event data. Only handlers read it.

The full rules are in the specification.

A `.jsonl` file¶

For development and tests. One envelope per line; blank lines and lines starting with # are ignored.

uv run keel ingest cartridges/fleet examples/events/fleet.jsonl

A Postgres outbox table¶

Your SaaS inserts envelopes into a table. Keel polls it in order and remembers the last row it took in a cursor file, so a restart resumes instead of re-ingesting.

CREATE TABLE keel_events (
    seq          bigserial PRIMARY KEY,
    id           text NOT NULL,
    tenant       text NOT NULL,
    type         text NOT NULL,
    subject      text NOT NULL,
    occurred_at  timestamptz NOT NULL,
    recorded_at  timestamptz NOT NULL DEFAULT now(),
    source       text NOT NULL,
    payload      jsonb NOT NULL DEFAULT '{}'
);

uv sync --extra postgres

import psycopg

from keel import Ingestor, MemoryStore, SqliteStore, load_cartridge
from keel.adapters.postgres import PostgresTableAdapter

ingestor = Ingestor(load_cartridge("cartridges/fleet"), SqliteStore("keel.db"))  # durable; MemoryStore() for a throwaway graph
adapter = PostgresTableAdapter(lambda: psycopg.connect(DSN), table="keel_events", cursor_path=".keel/cursor")

ingestor.ingest(adapter)                      # one pass over everything after the cursor
ingestor.ingest(adapter.follow(interval=5))   # or poll forever

Delivery is at-least-once: the cursor advances after the consumer has taken an event, so a crash mid-event re-delivers it, and the store applies each event in one transaction, so a redelivery is skipped as a duplicate. The table name, the cursor column and the batch size are parameters. Any DB-API 2.0 connection works; pass placeholder="?" for sqlite3.

sequenceDiagram
    participant S as Your SaaS
    participant T as keel_events
    participant A as PostgresTableAdapter
    participant C as cursor file
    participant I as Ingestor
    S->>T: INSERT envelope (seq 41)
    A->>C: read position
    C-->>A: 40
    A->>T: SELECT * WHERE seq > 40 ORDER BY seq LIMIT 500
    T-->>A: row 41
    A->>I: Event 41
    I->>I: dispatch to the handler
    I->>A: next event?
    A->>C: write 41
    Note over A,I: a crash before "write 41" re-delivers event 41 on restart

Concurrent writers

A serial cursor can skip a row that commits late while a later row was already read. Use one writer per table, or a recorded_at cursor with a lag, if that bites.

A webhook¶

Keel receives envelopes over HTTP, validates them and ingests them in the request.

uv sync --extra webhook

# app.py
from keel import Ingestor, MemoryStore, SqliteStore, load_cartridge
from keel.adapters.webhook import build_app

ingestor = Ingestor(load_cartridge("cartridges/fleet"), SqliteStore("keel.db"))  # durable; MemoryStore() for a throwaway graph
app = build_app(ingestor, token="shared-secret")

uv run uvicorn app:app --port 8000
curl -X POST localhost:8000/events \
  -H "Authorization: Bearer shared-secret" -H "Content-Type: application/json" \
  -d '[{"tenant": "acme-logistics", "type": "unit.registered", "subject": "unit:MXK-4821", "occurred_at": "2026-09-01T09:00:00Z", "source": "fleet.vehicles", "payload": {"id": "MXK-4821", "plate": "MXK-4821", "model": "Freightliner Cascadia"}}]'

The body is one envelope or a list. The response is the ingest report: processed, unhandled types and errors. A malformed envelope is a 422; a wrong token is a 401. Without a token the endpoint accepts any caller, so use that only on localhost.

Consume a Kafka topic or an SQS queue¶

Your SaaS publishes envelopes as JSON messages; Keel consumes them. uv sync --extra kafka (confluent-kafka) or --extra sqs (boto3).

from keel.adapters.bus import KafkaAdapter, SqsAdapter

adapter = KafkaAdapter.connect("broker:9092", topic="keel-events", group_id="keel")
adapter = SqsAdapter.connect("https://sqs.us-east-1.amazonaws.com/123456789012/keel-events")
ingestor.ingest(adapter)            # drain what is there now, then stop
ingestor.ingest(adapter.follow())   # consume forever

Delivery is at-least-once: the offset is committed, or the SQS message deleted, only after the handler ran, so a crash mid-event re-delivers it. A message that is not a valid envelope raises and stays on the bus; pass dead_letter=lambda body, error: ... to log it somewhere and move on. Both adapters only need two calls from their client (poll/commit, receive_message/delete_message), which is how the tests run them on fakes and how you would wrap another client.

Serve the tools over MCP¶

uv sync --extra mcp
uv run keel serve cartridges/fleet --tenant acme-logistics --events examples/events/fleet.jsonl

With Claude Code, one command registers the server. The agent sees exactly five tools and nothing else:

claude mcp add keel -- uv run --directory $PWD keel serve cartridges/fleet --tenant acme-logistics --events examples/events/fleet.jsonl

With Codex CLI (OpenAI), declare the server in ~/.codex/config.toml. Keel annotates context, history, search and related as read-only, so Codex runs them without asking even under approval_policy = "never"; annotate writes, so it prompts unless you set default_tools_approval_mode = "auto" on the server:

[mcp_servers.keel]
command = "uv"
args = ["run", "--directory", "/path/to/keel", "keel", "serve", "cartridges/fleet", "--tenant", "acme-logistics", "--events", "examples/events/fleet.jsonl"]
startup_timeout_sec = 60

A headless run, for a smoke test or CI:

codex exec --json "Which tires on unit:MXK-4821 are critical according to the cartridge's own rules? Use only the keel tools." < /dev/null

Checked on 2026-10-14 with Codex CLI 0.162 and openai/gpt-5.2-codex behind an OpenAI-compatible gateway: seven calls (search, related, context, history with predicate=depth_mm), the right tire, the right depth on 2026-09-20, and the threshold quoted from the cartridge rather than assumed. Any other MCP client works the same way: a stdio server, five tools, one tenant per server instance. If the cartridge declares agents:, add --agent <id>: the server then reads and writes only what that agent may, and signs its annotations.

Where is the graph stored?

In memory with --events: the server ingests the stream at start and serves it, which is enough for a demo. For a deployment, pass --db keel.db instead: a SQLite file (stdlib, WAL mode) that an ingest process fills while keel serve reads it. Or point --db at a MySQL or MariaDB you already run: --db mysql://user:password@host:3306/keel (uv sync --extra mysql); the tables are created on first use and the same conformance suite runs against it in CI. Sharing the database with your application: ?prefix=keel_ names the tables keel_entities, keel_facts, and so on.

uv run keel ingest cartridges/fleet examples/events/fleet.jsonl --db keel.db   # or adapter.follow() in your own process
uv run keel serve cartridges/fleet --tenant acme-logistics --db keel.db
uv run keel replay cartridges/fleet examples/events/fleet.jsonl acme-logistics --db keel.db   # wipe one tenant, rebuild it from the log

The log is the truth and the file is derived: replay rebuilds one tenant from the external log (your outbox table, your bus, a .jsonl) and never from the file itself. The store is pluggable: SqliteStore is the default, and any store that passes tests/test_store.py (see contributing) can take its place.

Serve the tools over HTTP¶

For a backend, or for many tenants from one process, serve the same five operations as a JSON API (uv sync --extra http):

KEEL_SECRET=change-me uv run keel serve cartridges/fleet --transport http --db keel.db --host 0.0.0.0 --port 8000

Call	Body	Returns
`POST /v1/events`	one envelope or a list	the ingest report
`POST /v1/context`	`{"subject", "as_of"?}`	what `context` returns
`POST /v1/history`	`{"subject", "since"?, "until"?, "predicate"?, "scope"?}`	facts
`POST /v1/search`	`{"text"}`	entities and facts
`POST /v1/annotate`	`{"subject", "predicate", "value", "scope"?, "evidence"?, "by"?}`	the fact
`POST /v1/related`	`{"subject", "relation"?}`	neighbours
`GET /openapi.json`		the contract, for generated clients

Every request carries Authorization: Bearer $KEEL_SECRET, X-Keel-Tenant and, when the cartridge declares agents, X-Keel-Agent. Your product authenticates its users and sets the two headers; Keel takes them as given and builds the tools for that tenant and agent per request. A refusal (scope policy, agent registry) is a 403 with its reason. With --tenant, the same server also exposes /mcp (MCP over streamable HTTP) bound to that tenant, for agents that speak MCP and cannot run a local process.

curl -s localhost:8000/v1/context -H "Authorization: Bearer change-me" -H "X-Keel-Tenant: acme-logistics" \
  -H "Content-Type: application/json" -d '{"subject": "tire:T-9001"}'

Export the audit trail¶

One tenant, everything Keel knows about it, as JSONL: entities, relations and facts, closed ones included, each with source, evidence, scope, recorded_at and corrected_by. Hand it to the customer, an auditor, or a diff.

uv run keel export acme-logistics --db keel.db --out acme-logistics.jsonl

Each line is one record: {"record": "fact", "subject": "tire:T-9001", "predicate": "depth_mm", "value": 5.2, "valid_from": ..., "valid_to": ..., "source": "tires.measurements", "evidence": "<event id>", ...}. Another tenant's export never contains this one's rows.