skeg-py

Python client for skeg, an SSD-primary KV+vector store designed for Personal AI Inference machines.

pip install skeg              # pure-Python
pip install 'skeg[fast]'      # PyO3-backed (binary wheels for macOS arm64, Linux x86_64, Linux aarch64)

RESP3 is the supported protocol. It carries the whole command surface, and it is the one to reach for unless you have a specific reason not to. The native binary client covers a smaller set of operations and is kept for existing callers.

import skeg

with skeg.connect() as c:  # RESP3 on 6379, HELLO already negotiated
    c.set(b"hello", b"world")

Two entry points, and they are not interchangeable:

Call	Wire	Port	Surface
skeg.connect()	RESP3	6379	Everything
skeg.client()	native binary	7379	KV + basic vectors

What's in the package

Two synchronous clients sharing one error hierarchy:

Module	Wire	Server binary	Default port	Use case
skeg.RespClient	RESP2/3 (Redis)	skeg-resp3	6379	Recommended. Full surface: KV, vectors, tenancy, quotas. Also a drop-in for redis-cli / redis-py
skeg.BinaryClient	skeg-proto (native)	skeg	7379	Lower framing overhead, smaller surface. No payloads, filters or tenancy

What only RESP3 can do: vector payloads (WITHPAYLOAD), search filters, VMSET bulk insert, index consolidation, and every tenancy, quota and QoS command. The native protocol has no equivalent for any of them.

Three backends behind one selector (skeg.client(...)):

• Pure-Python (default, zero deps) - just pip install skeg
• PyO3 / Rust (optional, faster framing) - pip install skeg[fast] (requires Rust toolchain at install time, or a pre-built wheel)

The PyO3 path mirrors BinaryClient's public API, so you can swap with no code changes:

import skeg

c = skeg.client(prefer_native=True)  # uses PyO3 if available
c = skeg.client(prefer_native=False)  # forces pure-Python

Both backends speak native protocol v1 and v2, including native_hello(), supports_kind() and the TurboQuant kinds:

import skeg
from skeg import _wire as wire

c = skeg.client(prefer_native=True, version=wire.VERSION_V2)
if c.supports_kind("tq2"):
    c.vindex_create("notes", dim=1024, kind="tq2", backend="disk")

tests/test_backend_parity.py compares the two surfaces and fails both ways: when the compiled backend loses a method, and when a method listed as missing comes back. So the two cannot drift apart quietly.

Install

pip install skeg              # pure-Python, zero dependencies
pip install 'skeg[fast]'      # PyO3-backed; binary wheels for macOS arm64, Linux x86_64, Linux aarch64

To build from source (e.g. on Windows or another arch), pip install will compile the PyO3 backend; set SKEG_PY_PURE=1 to skip it.

Quick start

KV

import skeg

with skeg.connect("127.0.0.1", 6379) as c:
    c.set(b"hello", b"world")
    print(c.get(b"hello"))  # b"world"
    print(c.mget([b"hello", b"nope"]))  # [b"world", None]
    print(c.append(b"hello", b"!"))  # 6 - the new length
    print(c.incr(b"counter"))  # 1
    c.delete(b"hello")

Vectors

import skeg

with skeg.connect() as c:
    # kind: f32 | int8 | binary | tq1 | tq2 | tq4. Omit it for the
    # server default, tq2 - near-f32 recall at a fraction of the RAM.
    c.vindex_create("notes", dim=1024, backend="flat")
    c.vset("notes", 1, my_embedding_1024d)
    c.vset("notes", 2, another_embedding, payload=b"chapter-3")

    for hit in c.vsearch("notes", query_embedding, k=10):
        print(hit.id, hit.score)

    # Payloads come back only when asked for, and can be filtered on.
    for hit in c.vsearch(
        "notes", query_embedding, k=10, with_payload=True, filter="tenant=alice"
    ):
        print(hit.id, hit.score, hit.payload)

Bulk insert in one round trip:

c.vmset("notes", [(1, vec_a, b"payload-a"), (2, vec_b, None)])
c.vindex_consolidate("notes")  # fold the disk delta into the graph

Vectors (on-disk Vamana)

Build the index offline (one-shot), then point the client at the served copy:

skeg-cli build --input embeddings.npy --output ./data --name notes
skeg --mode serve --data-dir ./data --tier pq:128:256

# Same client code as above; the server handles the disk-backed index.
hits = c.vsearch("notes", query, k=10)

Multi-tenant

c.hello(3, auth=("alice", "hunter2"))
print(c.skeg_whoami())  # "tenant=<hex> mode=tenant-aware"
# All subsequent GET/SET are auto-scoped to alice's namespace.

c.subject_erase(b"user:42:")  # erase one data subject, returns a count
c.quota_set("alice", max_vectors=1_000_000, max_disk_bytes=None)  # None = unlimited
c.qos_set("alice", rate=500, burst=1000, max_concurrent=8)
c.reclaim()  # admin: physically free erased bytes

Erasure is logical: the bytes leave the disk on a later reclaim().

Native binary protocol

Smaller surface, less framing overhead. Two wire versions:

from skeg import BinaryClient, VectorKindV2
from skeg import _wire as wire

# v1 (default): f32, int8 and binary kinds only.
with BinaryClient.connect("127.0.0.1", 7379) as c:
    c.vindex_create("notes", dim=1024, kind="int8")

# v2: adds the TurboQuant kinds and capability negotiation.
with BinaryClient.connect("127.0.0.1", 7379, version=wire.VERSION_V2) as c:
    protocol_version, kind_mask = c.native_hello()
    if c.supports_kind("tq2"):
        c.vindex_create("notes", dim=1024, kind="tq2")

Kind byte 3 means different things in the two versions: PQ in v1, TQ1 in v2. That collision is why v2 exists, and why VectorKindV2 is a separate enum from VectorKind rather than extra variants on it. A v1 connection asking for byte 3 is refused by the server rather than silently given the wrong index.

Testing

brew tap skegdb/tap
brew install skeg
git clone https://github.com/skegdb/skeg-py
cd skeg-py
pip install -e '.[test]'
SKEG_BIN=$(which skeg) SKEG_RESP3_BIN=$(which skeg-resp3) pytest

The fixture spawns one server per session and tears it down at the end. Both env vars are required: skeg-py used to live inside the skeg repo, where the binaries could be found by relative path, and no longer does. Tests that need a server skip without them.

Conformance suite

Command coverage is not asserted by hand here. tests/test_conformance_resp3.py and tests/test_conformance_native.py run the shared case files that every skeg client is checked against, driving this SDK's public methods:

SKEG_CONFORMANCE_DIR=<skeg-internal>/conformance \
SKEG_BIN=$(which skeg) SKEG_RESP3_BIN=$(which skeg-resp3) pytest

Without SKEG_CONFORMANCE_DIR those tests skip. Cases marked wire_only are skipped by design: they are things a typed SDK cannot express - a GET with no key, an unknown command - and the case files ship standalone validators that cover them against the raw wire.

Test-suite safety vs your data

The pytest suite is designed for a server it owns. It writes keys under names like doc:N, counter:N, sk-N, and creates/drops VINDEX entries prefixed with pytst-. If you ever override the default fixture to point the suite at a server that already holds real data:

• KV: keys with the names above will be overwritten or deleted.
• VINDEX: only entries that match the pytst- prefix are touched by the cleanup loop. Unrelated VINDEX are left alone.

In short: never point pytest at a server you can't afford to lose KV state on. The pytest-spawned fixture is the safe default.

Compatibility

• Python 3.10+ (one abi3 wheel covers 3.10, 3.11, 3.12, 3.13).
• macOS arm64, Linux x86_64, Linux aarch64 (wheels). Other targets build from sdist; set SKEG_PY_PURE=1 to skip the PyO3 backend.
• RESP2 and RESP3, and native protocol versions 1 and 2. All stable.

License

Apache-2.0. See LICENSE.