Python SDK - Retrieval Context Protocol

The Python SDK is written entirely against the standard library — no C extension, no pybind11, no compiled module. It mirrors the C++ SDK’s public surface and wire behaviour, so a Python client and a C++/Node/Rust server interoperate byte-for-byte.

Requirements

Python ≥ 3.9. No third-party dependencies.

pip install rcp-protocol

The package installs as rcp-protocol and imports as import rcp.

Server

import rcp

s = rcp.Server()
s.set_info("my-engine", "1.0")
s.advertise(rcp.Capability.Retrieve, {"maxK": 200, "modes": ["dense", "hybrid"]})
s.advertise(rcp.Capability.Embed, {"dimension": 384, "identity": "bge-small-en"})

@s.on(rcp.Method.RETRIEVE)
def _(params):
    hits = my_index.search(params["query"], params.get("k", 10))
    return {"hits": [{"id": h.id, "score": h.score, "text": h.text} for h in hits]}

@s.on(rcp.Method.EMBED)
def _(params):
    items = params.get("inputs") or params.get("texts", [])   # inputs preferred
    return {"vectors": [embed_one(t) for t in items]}

s.serve_stdio()          # or s.serve_http(8000)

Client

import rcp

c = rcp.connect_stdio(["python3", "my_engine.py"])
# or: c = rcp.connect_http("http://127.0.0.1:8000/rcp")

print(c.server, "RCP/" + str(c.protocol_version))

if c.supports(rcp.Capability.Retrieve):
    for h in c.retrieve("eiffel tower", k=3):
        print(h["id"], h["score"], h["text"])

if c.supports(rcp.Capability.Embed):
    (vec,) = c.embed(["hello world"])
    print("dim", len(vec))

c.shutdown()

A gated call to an unadvertised capability raises before any I/O:

try:
    c.rerank("q", ["a", "b"])
except RuntimeError as e:
    # str(e) == "[RCP -32003] server does not advertise 'rerank'"
    ...

Client methods: embed, embed_sparse, embed_multi, rerank, retrieve, search (returns {hits, usage, next_cursor}), graph, transform, index_add, index_delete, catalog, info, ping, call, shutdown.

Selecting a backend

sel = rcp.Selector.load("rcp.json")
c = sel.select_capable(rcp.Capability.Retrieve)   # highest-priority live engine

Fusion, streaming & the vector codec

The reference implementations of three spec algorithms ship as importable, tested helpers — identical across all four SDKs, so no adopter re-derives them:

# Reciprocal Rank Fusion of per-engine ranked lists (spec §16.3).
fused = rcp.rrf_fuse({"dense": dense, "sparse": sparse}, k=10,
                     weights={"dense": 1.0, "sparse": 0.7})
# rcp.weighted_fuse(...) when engine scores are comparable.

# Compact f32-base64 embedding codec (spec §7.3.1, ~4× smaller than JSON).
payload, meta = rcp.encode_vectors(vectors, "f32-base64")
vectors = rcp.decode_vectors(payload, meta["encoding"], meta["dimension"])

A streaming handler is a generator that yields rcp.Progress and returns the result; over HTTP+SSE the frames stream live, over a plain request the same handler returns one buffered response:

def retrieve(params):
    yield rcp.Progress(0.5, "recall")
    yield rcp.Progress(1.0, "rerank")
    return {"hits": hits}

s.advertise(rcp.Capability.Streaming)
s.stream("retrieve", retrieve)     # serve_http honours Accept: text/event-stream

See Federation, Streaming, and compact encoding. Runnable: examples/example_federation.py and examples/example_streaming.py.

Test

cd sdk/python
python3 test_bindings.py     # pure stdlib — nothing to build

The suite includes a case where the Python client drives the C++ example server, proving cross-language interop.

​Requirements

​Server

​Client

​Selecting a backend

​Fusion, streaming & the vector codec

​Test

Requirements

Server

Client

Selecting a backend

Fusion, streaming & the vector codec

Test