synap-sdk 1.2.0

Official Python SDK for Synap - High-Performance In-Memory Key-Value Store & Message Broker

Synap Python SDK

Official Python client library for Synap - High-Performance In-Memory Key-Value Store & Message Broker.

Features

• 💾 Key-Value Store: Fast in-memory KV operations with TTL support
• 📨 Message Queues: RabbitMQ-style queues with ACK/NACK
• 📡 Event Streams: Kafka-style event streams with offset tracking
• 🔔 Pub/Sub: Topic-based messaging with wildcards
• ⚡ StreamableHTTP Protocol: Unified endpoint for all operations
• 🛡️ Type-Safe: Full type hints with mypy strict mode
• 📦 Async/Await: Built on modern async patterns with httpx

Requirements

• Python 3.11 or higher
• Synap Server running

Installation

pip install synap-sdk

Quick Start

import asyncio
from synap_sdk import SynapClient, SynapConfig

async def main():
    # Create client
    config = SynapConfig.create("http://localhost:15500")
    async with SynapClient(config) as client:
        # Key-Value operations
        await client.kv.set("user:1", "John Doe")
        value = await client.kv.get("user:1")
        print(f"Value: {value}")

        # Queue operations
        await client.queue.create_queue("tasks")
        msg_id = await client.queue.publish("tasks", {"task": "process-video"}, priority=9)
        message = await client.queue.consume("tasks", "worker-1")
        
        if message:
            # Process message
            print(f"Received: {message.payload}")
            await client.queue.ack("tasks", message.id)

        # Event Stream
        await client.stream.create_room("chat-room-1")
        offset = await client.stream.publish("chat-room-1", "message", {
            "user": "alice",
            "text": "Hello!"
        })

        # Pub/Sub
        await client.pubsub.subscribe_topics("user-123", ["notifications.*"])
        delivered = await client.pubsub.publish("notifications.email", {
            "to": "[e-mail]",
            "subject": "Welcome"
        })

asyncio.run(main())

Transports

Since v0.11.0 the SDK selects the transport via URL scheme — no extra arguments required:

URL scheme	Default port	When to use
synap://	15501	✅ Recommended default — MessagePack over persistent TCP, lowest latency, preserves int/float/bool/bytes.
resp3://	6379	Redis-compatible text protocol — interop with existing Redis tooling.
http:// / https://	15500	Original REST transport — full command coverage.

All commands (KV, Hash, List, Set, Sorted Set, Queue, Stream, Pub/Sub, Transactions, Scripts, Geo, HyperLogLog) are fully supported on every transport. Native transports raise UnsupportedCommandError instead of silently falling back to HTTP.

from synap_sdk import SynapClient, SynapConfig

# SynapRPC — recommended default
config = SynapConfig("synap://127.0.0.1:15501")
client = SynapClient(config)

# RESP3 — Redis-compatible
config = SynapConfig("resp3://127.0.0.1:6379")
client = SynapClient(config)

# HTTP — full REST access
config = SynapConfig("http://127.0.0.1:15500")
client = SynapClient(config)

The synap:// transport is Thunder

The binary transport is not hand-written in this SDK. It is Thunder (hivellm-thunder, imported as thunder_rpc) — the HiveLLM family's shared binary RPC client, the same protocol the Synap server runs on, so the two ends of the wire cannot drift.

Pipelining	Concurrent commands multiplex over one connection, demultiplexed by frame id.
Frame cap	512 MiB, validated against the length prefix before allocating. The previous transport called readexactly() with whatever a 4-byte prefix claimed.
Timeouts	Connect and per-call, both from SynapConfig.timeout.
Reconnect	Lazy, with capped retries.
Authentication	auth_token / username+password travel in the handshake — the previous transport never sent AUTH, so it could not reach a require_auth server on 15501.
Push	SUBSCRIBE delivery uses Thunder's push hook, registered before the command is sent, so a message published between the acknowledgement and the reader starting is not lost.

Queue, stream and pub/sub over synap://:

import asyncio
from synap_sdk import SynapClient, SynapConfig

async def main():
    async with SynapClient(SynapConfig("synap://127.0.0.1:15501")) as client:
        # Queue round-trip
        await client.queue.create_queue("tasks", max_size=1000, message_ttl=60)
        msg_id = await client.queue.publish("tasks", {"job": "resize"}, priority=5)
        msg = await client.queue.consume("tasks", "worker-1")
        await client.queue.ack("tasks", msg.id)

        # Stream publish + read
        await client.stream.create_room("events")
        await client.stream.publish("events", "user.created", {"id": "u1"})
        events = await client.stream.read("events", offset=0)

        # Reactive pub/sub (server-push over dedicated TCP connection on synap://)
        async for msg in client.pubsub.observe(["news.*"]):
            print("got:", msg)
            break

API Reference

Configuration

from synap_sdk import SynapConfig

config = SynapConfig.create("http://localhost:15500") \
    .with_timeout(60) \
    .with_auth_token("your-token") \
    .with_max_retries(5)

async with SynapClient(config) as client:
    # Use client
    pass

Key-Value Store

# Set value with TTL
await client.kv.set("session:abc", session_data, ttl=3600)

# Get value
data = await client.kv.get("session:abc")

# Atomic operations
new_value = await client.kv.incr("counter", delta=1)
exists = await client.kv.exists("session:abc")

# Scan keys
keys = await client.kv.scan("user:*", limit=100)

# Get stats
stats = await client.kv.stats()

# Delete key
await client.kv.delete("session:abc")

KV Watch

Observe a key — or a wildcard pattern — and receive its new value on every change, without polling. Requires the synap:// transport; closing the iterator issues KV.UNWATCH:

# Watch one key, or a whole prefix
async for event in client.kv.watch("user:*"):
    # event: WatchEvent(key, event, version, value, truncated)
    print(event.event, event.key, event.version, event.value)

# Notify-only mode: change signals without value bandwidth (re-GET on demand)
async for event in client.kv.watch("hot:key", mode="notify"):
    value = await client.kv.get(event.key)

Delivery is best-effort, latest-value; WatchEvent.version resets when the key is deleted, expires or is evicted — version 1 marks a new incarnation.

Message Queues

# Create queue
await client.queue.create_queue("tasks", max_size=10000, message_ttl=3600)

# Publish message with priority
message_id = await client.queue.publish(
    "tasks",
    {"action": "encode", "file": "video.mp4"},
    priority=9,
    max_retries=3
)

# Consume and process
message = await client.queue.consume("tasks", "worker-1")
if message:
    try:
        # Process message
        await process_message(message.payload)
        await client.queue.ack("tasks", message.id)
    except Exception:
        await client.queue.nack("tasks", message.id)  # Requeue

# Get queue stats
stats = await client.queue.stats("tasks")
queues = await client.queue.list()

# Delete queue
await client.queue.delete_queue("tasks")

Event Streams

# First publish to a new stream? Use the idempotent
# get_or_create_room — it never errors when the room already exists,
# replacing the publish-or-create-then-republish dance every client
# otherwise reimplements (see hivellm/synap#165):
#
#     await client.stream.get_or_create_room("events")
#
# Or, when you genuinely need to fail on duplicates:
await client.stream.create_room("events")

# Publish event
offset = await client.stream.publish("events", "user.created", {
    "userId": "123",
    "name": "Alice"
})

# Read events
events = await client.stream.read("events", offset=0, limit=100)
for evt in events:
    print(f"Event: {evt.event} at offset {evt.offset}")

# Get stream stats
stats = await client.stream.stats("events")
rooms = await client.stream.list_rooms()

# Delete room
await client.stream.delete_room("events")

Pub/Sub

# Subscribe to topics (supports wildcards)
await client.pubsub.subscribe_topics("subscriber-1", [
    "user.created",
    "notifications.*",
    "events.#"
])

# Publish message
delivered = await client.pubsub.publish("notifications.email", {
    "to": "[e-mail]",
    "subject": "Welcome!"
})
print(f"Delivered to {delivered} subscribers")

# Unsubscribe
await client.pubsub.unsubscribe_topics("subscriber-1", ["notifications.*"])

# Get stats
stats = await client.pubsub.stats()

Async Context Manager

The client supports async context manager for automatic cleanup:

async with SynapClient(config) as client:
    await client.kv.set("key", "value")
    # Client automatically closed when exiting context

Or manual close:

client = SynapClient(config)
try:
    await client.kv.set("key", "value")
finally:
    await client.close()

Error Handling

from synap_sdk.exceptions import SynapException

try:
    await client.kv.set("key", "value")
except SynapException as e:
    print(f"Synap error: {e}")

Custom HTTP Client

You can provide your own httpx.AsyncClient for advanced scenarios:

import httpx
from synap_sdk import SynapClient, SynapConfig

http_client = httpx.AsyncClient(timeout=120)
config = SynapConfig.create("http://localhost:15500")

async with SynapClient(config, http_client) as client:
    # Use client
    pass

Type Hints

The SDK is fully typed and passes mypy strict mode:

mypy synap_sdk

Development

Install Development Dependencies

pip install -e ".[dev]"

Run Tests

pytest

Run Tests with Coverage

pytest --cov=synap_sdk --cov-report=html

Type Checking

mypy synap_sdk

Linting

ruff check synap_sdk

Formatting

ruff format synap_sdk

License

Apache License 2.0 - See LICENSE for details.

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for details.

Links

• Synap Server
• Documentation
• Other SDKs