ulmp 0.1.1

Python client for UMP (Ulmen Message Protocol)

ulmp

Python client for UMP (Ulmen Message Protocol) -- the wire protocol for agentic AI database workloads.

Backed entirely by a Rust implementation via PyO3. Zero protocol logic duplicated in Python. Maximum performance. Full type safety.

Install

pip install ulmp

Requires Python 3.10+. The package includes a compiled Rust extension, no Rust toolchain needed to install.

30-second quickstart

Start the built-in test server:

python -m ulmp.testserver --port 7771 --token mytoken

In another terminal:

import asyncio
from ulmp import AsyncClient

async def main():
    db = await AsyncClient.connect("localhost", 7771, token=b"mytoken")
    ns = db.namespace("my_project")

    await ns.put("auth.py::AuthService", b"class AuthService: pass")
    value = await ns.get("auth.py::AuthService")
    print(value)  # b"class AuthService: pass"

    await db.close()

asyncio.run(main())

That is it. You are connected, authenticated over TLS, and reading/writing records.

Connection

from ulmp import AsyncClient

# Basic connection
db = await AsyncClient.connect(
    host="localhost",
    port=7771,
    token=b"your_secret_token",
)

# With all options
db = await AsyncClient.connect(
    host="db.internal",
    port=7771,
    token=b"secret",
    server_name="db.internal",      # TLS certificate name
    client_name="my-agent/1.0",     # identifies your client
    verify_cert=True,               # verify server certificate
    timeout=10.0,                   # connection timeout in seconds
)

# Always close when done
await db.close()

Use as an async context manager for automatic cleanup:

async with await AsyncClient.connect("localhost", 7771, token=b"secret") as db:
    ns = db.namespace("my_project")
    await ns.put("key", b"value")
    # connection closed automatically on exit

Check connection health:

latency_us = await db.ping()        # round-trip in microseconds
print(db.is_connected)              # True
print(db.info.server_name)          # "ulmendb/0.1.0"
print(db.info.session_id)           # server-assigned session ID

Namespaces

A namespace is a scoped view into one indexed codebase. All record operations are scoped to a namespace.

ns = db.namespace("github.com/org/repo@abc123")

Write

await ns.put("auth.py::AuthService", b"class AuthService: pass")
await ns.put("auth.py::validate_token", b"def validate_token(t): ...")

Read

value = await ns.get("auth.py::AuthService")
# Returns bytes if found, None if not found
if value is None:
    print("key not found")

Delete

await ns.delete("auth.py::old_function")

Batch write

await ns.put_batch([
    ("auth.py::AuthService", b"class AuthService: ..."),
    ("auth.py::validate_token", b"def validate_token(): ..."),
    ("auth.py::hash_password", b"def hash_password(p): ..."),
])

Range scan

# All keys starting with "auth.py::"
pairs = await ns.scan("auth.py::", "auth.py::zzz", limit=100)
for key, value in pairs:
    print(key, len(value), "bytes")

Range delete

# Delete all keys in auth.py in one operation
await ns.range_delete("auth.py::", "auth.py::zzz")

Query builder

Build queries with a fluent chaining API. The server runs all matching indices in parallel and merges results.

results = await ns.query("JWT authentication validation") \
    .top_k(10) \
    .lang("python") \
    .type("function") \
    .relations("calls", "imports") \
    .file("src/*.py") \
    .timeout(5000) \
    .execute()

for row in results:
    print(f"{row.key}  score={row.final_score:.4f}  rank={row.rank}")

Stream results

For large result sets, stream results one at a time and stop early:

async for row in ns.query("validate token").top_k(50).stream():
    print(row.key, row.final_score)
    if row.final_score < 0.5:
        break  # sends ENOUGH to server, stops streaming

Query methods

Method	Description
.top_k(n)	Maximum results (default: 10)
.lang("python", "typescript")	Filter by programming language
.type("function", "class")	Filter by node type
.file("src/*.py")	Filter by file glob pattern
.relations("calls", "imports")	Follow relation edges in the graph
.depth(n)	Max graph traversal depth (default: 3)
.vector([0.1, 0.2, ...])	Pre-computed embedding vector
.timeout(ms)	Cancel if not done in N milliseconds
.merge_rrf()	Reciprocal Rank Fusion merge (default)
.merge_linear()	Linear combination merge
.merge_max()	Max score merge
.execute()	Run and return all results
.stream()	Run and yield results as async iterator

Branches

Branch and merge like git. Native to the protocol, not an application-level hack.

async with ns.branch("feat/refactor", description="refactor auth") as b:
    await b.put("auth.py::validate_token", new_implementation)
    await b.delete("auth.py::old_helper")
    # merges automatically on clean exit
    # rolls back automatically on any exception

If you need manual control:

branch = ns.branch("experiment/risky-change")
async with branch as b:
    await b.put("auth.py::service", b"new code")
    # raise an exception here -> automatic rollback
    raise ValueError("this approach does not work")
    # branch is rolled back, original code is preserved

List and diff branches:

branches = await ns.branch_list()
diff = await ns.branch_diff("feat/a", "feat/b")

Transactions

ACID transactions with three isolation levels.

async with db.transaction() as tx:
    val = await tx.get("counter")
    current = int(val or b"0")
    await tx.put("counter", str(current + 1).encode())
    # auto-commit on clean exit
    # auto-rollback on exception

Explicit control:

async with db.transaction(isolation="serializable") as tx:
    await tx.put("key", b"value")
    await tx.commit()  # explicit commit

async with db.transaction() as tx:
    await tx.put("key", b"risky")
    await tx.rollback()  # explicit rollback

Isolation levels:

Level	Description
"read_committed"	See committed writes from other transactions
"snapshot"	See a consistent snapshot from transaction start (default)
"serializable"	Full conflict detection, rejects on write-write conflict

Watch

Push notifications when keys change. No polling.

async with ns.watch("auth.py::") as stream:
    async for event in stream:
        print(f"{event.key} changed: {event.change_type}")
        # change_type: "put", "delete", "branch_create", "branch_merge"

Snapshots

Create and restore point-in-time snapshots.

# Save current state
await ns.snapshot_create("before refactor")

# List snapshots
snaps = await ns.snapshot_list()

# Restore to a previous state
await ns.snapshot_restore("snap-001")

Distributed tracing

Every ULMP frame can carry a W3C-compatible trace context.

from ulmp import TraceContext

# Root span
tc = TraceContext.generate()
print(tc.to_traceparent())  # "00-{trace_id}-{span_id}-01"

# Child span (same trace, new span)
child = tc.child_span()

# For custom protocol work
encoded = tc.encode()  # 25 bytes

Low-level API

For building raw UMP payloads when implementing custom protocol logic:

from ulmp import PayloadBuilder, decode_payload

# Build a payload
pb = PayloadBuilder()
pb.push_string("auth.py::AuthService")
pb.push_bytes(b"class AuthService: pass")
pb.push_u32(42)
pb.push_bool(True)
pb.push_null()
payload = pb.finish()  # bytes ending with TAG_END

# Decode a payload
values = decode_payload(payload)
# values = ["auth.py::AuthService", b"class AuthService: pass", 42, True, None]

Frame encoding

from ulmp import Header, HEADER_SIZE, MAGIC_V1

# Create a frame header
h = Header(opcode=0x06, stream_id=1, sequence=0, payload_length=9)
buf = h.encode()  # 20 bytes with CRC32

# Decode a frame header
h = Header.decode(buf)
print(h.opcode, h.stream_id, h.payload_length)

Checkpoint tokens

For resuming long-running streaming operations after disconnect:

from ulmp import CheckpointToken

# Server creates a checkpoint
token = CheckpointToken(
    session_id=42,
    stream_id_high=0,
    stream_id_low=1,
    last_confirmed_row=1000,
    ttl_seconds=3600,
)
encoded = token.encode()  # 64 bytes

# Client decodes after reconnect
token = CheckpointToken.decode(encoded)
print(token.last_confirmed_row)  # 1000
print(token.is_expired())        # False

Sequence tracking

Detect missing or duplicate frames:

from ulmp import SequenceTracker

tracker = SequenceTracker()
assert tracker.check(0)   # True: first frame
assert tracker.check(1)   # True: sequential
assert not tracker.check(3)  # False: gap detected
print(tracker.expected)   # 2

Crypto

All crypto runs in Rust at native speed. Zero external dependencies.

from ulmp import crc32, sha256, hmac_sha256

# CRC32 (IEEE 802.3)
assert crc32(b"123456789") == 0xCBF43926

# SHA-256 (FIPS 180-4)
digest = sha256(b"hello world")
assert len(digest) == 32

# HMAC-SHA256 (FIPS 198-1)
mac = hmac_sha256(key=b"secret", msg=b"data")
assert len(mac) == 32

Error handling

All errors have typed exceptions with error codes.

from ulmp import (
    UlmpError,           # base exception
    ConnectionError,     # TCP/TLS failure
    AuthError,           # authentication rejected
    ProtocolError,       # frame/payload decode failure
    ServerError,         # server returned an error
    KeyNotFoundError,    # key does not exist
    NamespaceError,      # namespace not found or access denied
    QueryError,          # query execution failed
    TransactionError,    # transaction conflict or failure
    BranchError,         # branch not found or merge conflict
    RateLimitError,      # request rate exceeded
    TimeoutError,        # request deadline exceeded
)

# get() returns None for missing keys (no exception)
val = await ns.get("nonexistent")
assert val is None

# Server errors have codes
try:
    async with db.transaction(isolation="serializable") as tx:
        await tx.put("key", b"val")
except TransactionError as e:
    print(f"conflict: code=0x{e.code:02x}")

Exception hierarchy:

UlmpError
  ConnectionError
  AuthError
  ProtocolError
  ServerError
    KeyNotFoundError
    NamespaceError
    QueryError
    TransactionError
    BranchError
    RateLimitError
  TimeoutError

Constants

All ULMP protocol constants are available from the Rust extension:

from ulmp._core import (
    # Frame
    HEADER_SIZE,       # 20
    MAGIC_V1,          # 0x554D5001

    # 72 opcodes
    OP_PUT,            # 0x30
    OP_GET,            # 0x31
    OP_QUERY,          # 0x40
    OP_RESULT_ROW,     # 0x51
    OP_BRANCH_CREATE,  # 0x74

    # 13 capability bits
    CAP_TRACING,       # 1 << 3
    CAP_WATCH,         # 1 << 4
    ULMP_V1_CAPS,      # full v1 capability set

    # 31 error codes
    ERR_KEY_NOT_FOUND,  # 0x40
    ERR_TX_CONFLICT,    # 0x61

    # 8 flag bits
    FLAG_HAS_TRACE,     # 0b00000001
    FLAG_STREAM_END,    # 0b00010000
)

Test server

The built-in test server lets you develop and test without a full ulmendb installation.

# Start the test server
python -m ulmp.testserver --port 7771 --token mytoken

The test server:

• accepts TLS connections with a self-signed certificate
• authenticates via HMAC-SHA256 challenge-response
• stores records in memory (not persisted)
• supports PUT, GET, DELETE, SCAN, PING/PONG
• runs single-threaded, one connection at a time Not for production use. Use ulmendb for production.

Protocol

ulmp implements the UMP (Ulmen Message Protocol) specification:

• 72 opcodes across 16 groups
• 22-type self-describing payload system
• 20-byte frame header with CRC32 integrity
• TLS 1.3 mandatory, Ed25519 preferred
• HMAC-SHA256 challenge-response authentication
• Distributed tracing (W3C Trace Context compatible)
• Frame fragmentation for large payloads
• Connection-level and per-stream flow control
• In-session token rotation
• Checkpoint and stream resume
• Zero-downtime upgrade with drain/redirect Full specification: [./docs]

Architecture

Python layer (ulmp)           Rust core (ulmp._core)
  AsyncClient                   Frame encode/decode
  Namespace                     Payload type system
  QueryBuilder                  SHA-256, HMAC-SHA256, CRC32
  BranchContext                 TLS via rustls
  Transaction                   Checkpoint tokens
  WatchStream                   Sequence tracking
  Error hierarchy               72 opcode constants
  TestServer                    13 capability bits
                                31 error codes

All protocol logic runs in Rust. Python handles ergonomics and async I/O.

Performance

The Rust extension handles all CPU-intensive work:

Operation	Backend
Frame encode/decode	Rust
CRC32	Rust
SHA-256	Rust
HMAC-SHA256	Rust
Payload serialization	Rust
TLS handshake	Rust (rustls)
Async I/O	Python (asyncio)

Requirements

• Python 3.10+
• No external Python dependencies
• Rust extension included in wheel (no Rust toolchain needed to install)

License

BSL 1.1

Author

El Mehdi Makroumi