Sockudo Python HTTP Server SDK

High-performance Python server SDK for the Sockudo HTTP API. It publishes events, signs channel and user authentication payloads, validates webhooks, queries channel state, reads durable history, mutates versioned messages, and manages annotations.

Features

• Sync and asyncio clients: Sockudo and AsyncSockudo
• Persistent HTTP connection pooling via httpx, with HTTP/2 enabled by default
• Pusher-compatible signed REST requests
• Single, multi-channel, and batch publishing
• Explicit and automatic idempotency keys for safe publish retries
• Private, presence, user, and webhook authentication helpers
• Channel state, presence users, durable history, and presence history APIs
• Versioned message APIs: get, versions, update, delete, append
• Annotation APIs: publish, list, delete
• Operator controls for history reset/purge and presence-history reset
• End-to-end encrypted channel auth, publish, batch publish, and webhook decrypt support for private-encrypted-* channels

Install

For apps, install the published package:

pip install sockudo-http-python

For contributors working inside this repository:

pip install -e server-sdks/sockudo-http-python[dev]

Quick Start

from sockudo_http import Config, Sockudo

sockudo = Sockudo(
    Config(
        app_id="app-id", key="app-key", secret="app-secret", host="127.0.0.1", port=6001
    )
)

result = sockudo.trigger("orders", "order.created", {"id": "ord_123"})
assert result.ok
sockudo.close()

Async:

from sockudo_http_python import AsyncSockudo, SockudoOptions

async with AsyncSockudo(
    "app-id",
    "app-key",
    "app-secret",
    options=SockudoOptions(host="127.0.0.1", port=6001),
) as sockudo:
    await sockudo.trigger("orders", "order.created", {"id": "ord_123"})

Idempotent Publishing

from sockudo_http_python import TriggerOptions

sockudo.trigger(
    "orders",
    "order.created",
    {"id": "ord_123"},
    TriggerOptions(idempotency_key="order-created-ord_123"),
)

sockudo.trigger(
    "orders", "order.created", {"id": "ord_124"}, TriggerOptions(idempotency_key=True)
)

Set SockudoOptions(auto_idempotency=True) to generate keys for publish and batch publish calls that omit one.

Target a single user channel:

sockudo.send_to_user("user-123", "notice", {"body": "hello"})

Authentication Helpers

from sockudo_http_python import PresenceUser

private_body = sockudo.authenticate("123.456", "private-orders")

presence_body = sockudo.authenticate(
    "123.456",
    "presence-room",
    PresenceUser("user-1", {"name": "Ada"}),
)

user_body = sockudo.authenticate_user("123.456", {"id": "user-1", "name": "Ada"})

Encrypted channel auth responses include shared_secret when encryption_master_key_base64 is configured:

encrypted = Sockudo(
    "app-id",
    "app-key",
    "app-secret",
    encryption_master_key_base64="base64-encoded-32-byte-key",
)

body = encrypted.authenticate("123.456", "private-encrypted-room")

Channel And History APIs

from sockudo_http_python import ChannelsParams, HistoryParams, PresenceHistoryParams

sockudo.list_channels(
    ChannelsParams(
        filter_by_prefix="presence-", info=["subscription_count", "user_count"]
    )
)
sockudo.get_channel_users("presence-room")
sockudo.get_channel_history("orders", HistoryParams(limit=50, direction="newest_first"))
sockudo.get_channel_presence_history("presence-room", PresenceHistoryParams(limit=50))

User Controls

Terminate all active connections for a user:

# Synchronous
sockudo.terminate_user_connections("user-123")

# Asynchronous
await async_sockudo.terminate_user_connections("user-123")

Force Reconnect User

Close all active connections for a user with code 4200, prompting clients to reconnect:

# Synchronous
sockudo.force_reconnect_user("user-123")

# Asynchronous
await async_sockudo.force_reconnect_user("user-123")

Versioned Messages And Annotations

from sockudo_http_python import MessageMutation, PublishAnnotationRequest

sockudo.get_message("orders", "msg:1")
sockudo.get_message_versions("orders", "msg:1")
sockudo.update_message("orders", "msg:1", MessageMutation(data={"status": "paid"}))
sockudo.append_message("orders", "msg:1", " appended text")
sockudo.delete_message("orders", "msg:1", MessageMutation(description="moderated"))

sockudo.publish_annotation(
    "orders",
    "msg:1",
    PublishAnnotationRequest(
        type="reactions:distinct.v1", name="like", client_id="user-1", count=1
    ),
)
sockudo.list_annotations("orders", "msg:1")

Webhooks

validity = sockudo.validate_webhook_signature(
    x_pusher_key, x_pusher_signature, raw_body
)
webhook = sockudo.parse_webhook(x_pusher_key, x_pusher_signature, raw_body)

If a webhook contains encrypted channel events and the client has an encryption master key, parse_webhook decrypts those event payloads.

Signed URIs

uri = sockudo.signed_uri(
    "GET", "/apps/app-id/channels", parameters={"filter_by_prefix": "presence-"}
)

The signing format matches Sockudo/Pusher REST auth: auth_key, auth_timestamp, auth_version, optional body_md5, and auth_signature over {METHOD}\n{PATH}\n{SORTED_QUERY}.

URL Configuration

sockudo = Sockudo.from_url("http://app-key:app-secret@127.0.0.1:6001/apps/app-id")