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": "user@example.com",
            "subject": "Welcome"
        })

asyncio.run(main())

Transports

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

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.

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 (`delta` applies identically on every transport;
# zero and negative steps apply as given, overflow fails without a change)
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": "user@example.com",
    "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