rcp — Retrieval Context Protocol, Python SDK

A native, pure-standard-library Python SDK for RCP — the open, versioned JSON-RPC protocol that lets any RAG engine expose embed / rerank / retrieve / graph / index / catalog, and any client consume it uniformly.

No dependencies, no compiler, no build step — just subprocess, http, socket, and json from the standard library. It speaks the exact same wire format as the C++, Node.js, and Rust SDKs, so a Python client and a C++/Node/Rust server (or vice-versa) interoperate byte-for-byte.

Install

pip install rcp-protocol

Requires Python ≥ 3.9. The import name is rcp:

import rcp

Client

Connect to any RCP server over a subprocess (stdio) or HTTP, then make typed, capability-gated calls.

import rcp

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

print(c.server(), c.protocol_version(), list(c.capabilities()))

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

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

c.shutdown()

A call to a capability the server never advertised raises before any I/O:

try:
    c.rerank("q", ["a", "b"])
except rcp.RcpError as e:
    # e.code == rcp.Errc.CAPABILITY_MISSING  (-32003)
    ...

Client methods: embed, embed_sparse, embed_multi, rerank, retrieve, search (returns hits + usage + nextCursor), graph, transform, index_add, index_delete, feedback, memory_build, memory_recall, catalog, info, ping, call, shutdown.

Agentic & frontier RAG

The SDK carries the full spec surface for 2024–2026 RAG. retrieve accepts unit / level (granularity — chunk…subgraph…tree-node), tokenBudget (long-context packing), and sessionId (agentic trajectories) in its opts, and each returned hit preserves confidence (normalised [0,1]), unit / level, provenance (graph/tree lineage), trust (provenance + safety), and per-stage scores. Two dedicated methods complete the loop:

# RL / corrective / integrity signals back to the retriever (spec §7.16)
c.feedback([{"hitId": hit["id"], "used": True, "cited": True, "reward": 0.9}])

# MemoRAG / HippoRAG memory -> clues you fan out over retrieve/graph (spec §7.17)
mem = c.memory_build(scope="global")
for clue in c.memory_recall("my question", memory_id=mem["memoryId"])["clues"]:
    hits = c.retrieve(clue.get("query", ""), k=5, opts={"sessionId": "traj-1"})

Each surface is capability-gated (Capability.Feedback, Capability.Memory, Capability.Session) — a server that never advertised it fails fast, client-side.

Server

Expose a Python RAG engine as an RCP server.

import rcp

s = rcp.Server()
s.set_info("my-engine", "1.0")
s.advertise(rcp.Capability.Retrieve, {"maxK": 100, "modes": ["hybrid"]})

@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.serve_stdio()          # or: s.serve_http(8000)

The Server owns the initialize handshake, capability gating, JSON-RPC framing and batching, and error mapping. It answers initialize / info / ping / shutdown itself; a gated call before initialize is -32001, an unadvertised or unimplemented method is -32003, an unknown method is -32004.

Selecting a backend

Pick one engine from a registry — by id, by required capability, or by priority with liveness fallback. Connection is lazy.

sel = rcp.Selector.loads('''{
  "engines": [
    {"id": "docs", "transport": "stdio", "command": ["python3", "server.py"], "priority": 10},
    {"id": "web",  "transport": "http",  "url": "http://127.0.0.1:8000/rcp",   "priority": 5}
  ]
}''')

c = sel.select_capable(rcp.Capability.Retrieve)

Federation fusion & the vector codec

Merge per-engine ranked lists with the reference Reciprocal Rank Fusion (spec §16.3) — deterministic tie-break, richest-body dedup, origin tags in meta.engine:

fused = rcp.rrf_fuse({"dense": dense_hits, "sparse": sparse_hits}, k=10,
                     weights={"dense": 1.0, "sparse": 0.7})
# or rcp.weighted_fuse(...) when engine scores are comparable

Encode embeddings compactly for the wire (spec §7.3.1, ~4× smaller than JSON numbers):

payload, meta = rcp.encode_vectors(vectors, "f32-base64")
vectors = rcp.decode_vectors(payload, meta["encoding"], meta["dimension"])

Streaming (SSE)

A generator handler yields rcp.Progress events and returns the final result; serve_http streams notifications/progress frames then the response over one text/event-stream connection (spec §9/§13), and the same handler answers a plain unary POST:

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)

Examples & tests

python3 examples/example_server.py            # stdio (default)
python3 examples/example_server.py --http 8000
python3 examples/example_client.py            # drives the example server
python3 examples/example_streaming.py         # HTTP+SSE progress, end to end
python3 examples/example_federation.py        # two engines fanned out + RRF-fused

python3 test_bindings.py                      # smoke test incl. Python client ↔ C++ server

License

MIT © 2026 Ayush Bhat.