msgmesh 0.1.2

MsgMesh Python SDK — publish / consume / realtime (SSE / WebSocket) / governance client for the multi-tenant event bus (Python port of sdk-js).

msgmesh (Python SDK)

English | 繁體中文

Python SDK for MsgMesh — a publish / consume / realtime (SSE / WebSocket) / governance client for the multi-tenant event bus. This is the Python port of the TypeScript SDK (@msgmesh/sdk): the public API surface is aligned, method names follow Python conventions (snake_case), and the semantics and coverage match sdk-js. The HTTP contract's source of truth is the platform OpenAPI spec.

Install

pip install msgmesh

Requires Python ≥ 3.9; the only runtime dependency is httpx.

Quick start

Register an account in the panel and issue an API key (shown in plaintext only once), then:

from msgmesh import MsgMesh

mq = MsgMesh(
    api_key="mk_live_...",                  # long-lived key, server-side only
    control_plane_url="https://cp.example.com",
    gateway_url="https://gw.example.com",
    realtime_url="https://rt.example.com",
)

mq.create_topic("orders")
mq.publish("orders", {"hello": 1})
msgs = mq.poll("orders", group="g1")
for m in msgs:
    print(m.value)

MsgMesh is also a context manager and closes the underlying connection pool on exit:

with MsgMesh(api_key="mk_live_...") as mq:
    mq.publish("orders", {"hello": 1})

Not for browsers / untrusted clients; mint short-lived tokens server-side

get_token mirrors the sdk-js token-broker pattern: give it a callable that fetches a short-lived dp token from your backend (returning {"token": ..., "expires_in": ...} or a TokenResponse). The SDK caches it, refetches before expiry, and rotates it on SSE reconnect. It is mutually exclusive with api_key; at least one is required.

import httpx
from msgmesh import MsgMesh

def fetch_token():
    return httpx.get("https://my-backend/mm-token").json()  # {"token": ..., "expires_in": 300}

mq = MsgMesh(
    get_token=fetch_token,
    gateway_url="https://gw.example.com",
    realtime_url="https://rt.example.com",
)

Realtime receive

Consistent interface — each returns a "stop" callable and runs on a background daemon thread:

• subscribe(topic, handler, *, group=None, max=None, on_error=None): long-polling (a poll loop). Terminal vs recoverable — only 401 (key invalid/missing = terminal) stops permanently; 403 (resolvable by self-service top-up) and other transient errors are reported then retried with backoff (self-healing). In get_token mode a 401 is usually just an expired token: the cache is invalidated and it retries with a fresh token, only declaring permanent revocation after several consecutive rejections. On permanent stop, if no on_error is given, a logging warning is emitted.

  stop = mq.subscribe("room.42", lambda m: print(m.value), group="g1", on_error=print)
  # ...
  stop()  # stop polling
  

• stream(topic, on_message, on_error=None, room=None): realtime receive over SSE (connection auth via query key). on_message receives the text body of each event; on_error receives an Exception (transport / non-2xx) or a StreamClose (a named server-side close event whose .data is the reason string).

  stop = mq.stream("room.42", print)
  # ...
  stop()
  

  Difference from sdk-js: sdk-js stream relies on the browser-native EventSource (which auto-reconnects). Python has no EventSource, so this SDK manages reconnection itself (in the spirit of sdk-js streamWs): it reconnects with backoff after each stream end / connection error, rotates the token in get_token mode, resets the failure counter on a successful connect, and stops after a bounded number of consecutive failures (avoiding infinite reconnect to a dead key/endpoint); "authorization revoked" is terminal and stops proactively.

• stream_ws(topic, on_message, on_error=None, room=None): realtime receive over WebSocket (connection auth via query key). Same interface as stream; needs the optional dependency websocket-client:

  pip install "msgmesh[ws]"
  

  from msgmesh import WsClose
  
  def on_err(e):
      if isinstance(e, WsClose):
          print("closed", e.code, e.reason)   # e.g. 1008 / "authorization revoked"
      else:
          print("error", e)                    # transport / handshake failure
  
  stop = mq.stream_ws("room.42", print, on_error=on_err)
  # ...
  stop()
  

  This is a pure-Python server-side capability, unrelated to browsers / Node (sdk-js's streamWs is what involves the browser / Node ≥ 22 global WebSocket). Calling stream_ws without websocket-client installed raises an immediate ImportError with an install hint.

  Difference from stream (SSE): WebSocket has no native reconnect, so reconnection is entirely SDK-managed — backoff reconnect, reset the failure counter on a successful connect, and both modes bound consecutive failures (stop at the limit, warn if no on_error); get_token also rotates the token before reconnecting. Revocation: mid-connection it is CLOSE 1008 + "authorization revoked" (terminal, stops immediately); during handshake (HTTP 401) it is close code 1006, absorbed by bounded reconnection. on_error receives an Exception (transport / handshake failure) or a WsClose (a close event with readable .code/.reason); stop() closes proactively without triggering on_error. If you'd rather not add the dependency, stream (SSE) or subscribe (long-poll) cover realtime receive server-side.

Rooms

A single topic can be split into multiple rooms (room = Kafka record key), decoupling "number of rooms" from "number of topics". Two layers:

① Routing — publish with publish(topic, body, key=room_id) to target a room, and subscribe with the optional room argument (same for stream / stream_ws) to receive only that room:

stop = mq.stream("chat", print, room="room-42")      # only room-42 messages
stop = mq.stream_ws("chat", print, room="room-42")   # WebSocket, same idea
mq.publish("chat", {"text": "hi"}, key="room-42")    # publish to room-42

Omitting room = receive all messages on the topic (backward compatible). Routing only filters — it does not enforce isolation; a malicious client can switch to someone else's room and eavesdrop on other rooms in the same topic. For real isolation see ②.

② Isolation (platform-enforced) — add the optional rooms to a credential's capabilities and the platform enforces that the credential can only send/receive the named rooms (403 on overreach). rooms omitted/empty = all rooms (backward compatible); non-empty = only these. The typical approach: the backend holds an all-rooms key and downscopes it via POST /v1/tokens to mint a short-lived "room-42 only" token for the frontend (downscope may only narrow, must be a subset of the key's capabilities, 403 on overreach):

import httpx

# Backend token-broker: downscope an all-rooms key to a short-lived "chat / room-42 only"
# token, returned to the frontend as get_token
def mint_room_token():
    r = httpx.post(
        "https://cp.example.com/v1/tokens",
        headers={"Authorization": "Bearer mk_live_..."},   # backend-held all-rooms long-lived key
        json={
            "ttl_seconds": 600,
            "capabilities": [{"ops": ["subscribe", "publish"], "topics": ["chat"], "rooms": ["room-42"]}],
        },
    )
    return r.json()   # {"token": ..., "expires_in": 600}

You can also mint a persistent room-scoped key with create_key("key", capabilities=[{"ops": [...], "topics": [...], "rooms": [...]}]). Platform enforcement points: subscribe (SSE/WS) must carry a ?room within the allowed set (omitting it = wanting all rooms, also 403); publish ?key must be within the allowed set.

⚠️ A room-scoped credential can only use realtime (SSE/WS) + publish to its rooms; it cannot poll / consume / DLQ. Those are a whole-topic firehose (the consumer-group offset would consume other rooms; one group per room = read amplification) and can't be cleanly per-room filtered, so a room-restricted credential always gets 403 (use realtime SSE/WS ?room=). Use an unrestricted credential when you need poll/consume.

Room isolation security notes (must read)

• Isolation strength = the scope of the token you issue. Isolation only exists when the backend downscopes an all-rooms key into a room-scoped token for the frontend. Never put an unrestricted credential (a full key, or a token without rooms) into the frontend / untrusted clients — that lets anyone change room and see all rooms, so isolation is meaningless.
• The platform does not verify "who the sender is." Room isolation governs "which rooms you can send/receive," not "who you are in the room." Within a room, anyone holding that room's token can impersonate any sender in the payload. To prevent in-room impersonation: mint a token per user on the backend and stamp / verify the sender there, don't let untrusted clients self-report identity.
• Note: presence (online count) is currently per-topic, not per-room (only leaks an aggregate number); short-lived tokens are bearer tokens — leaking one = usable for that room until TTL expires (so keep the TTL short and don't log it).

Error handling

Non-2xx responses raise a typed error by status code (all inherit MsgMeshError and carry status/code/path/request_id):

from msgmesh import ValidationError, RateLimitError

try:
    mq.create_topic("Bad Name!")
except ValidationError as e:
    print("invalid argument:", e)
except RateLimitError:
    print("rate limited, retry later")

Status	Type	code
400 / 422	ValidationError	validation
401 / 403	AuthError	auth
404	NotFoundError	not_found
429	RateLimitError	rate_limit
other	MsgMeshError	server
transport failure	MsgMeshConnectError	connect

API overview (mapped to sdk-js)

Method names are sdk-js camelCase → Python snake_case; return types are dataclasses (attribute access, e.g. topic.name).

• Topics: create_topic / list_topics / delete_topic
• Send/receive: publish / poll / subscribe (polling) / stream (SSE) / stream_ws (WebSocket, optional msgmesh[ws]) / get_presence
• Keys: list_keys / create_key (with optional scope + capabilities) / delete_key
• Webhooks: list_webhooks / create_webhook / delete_webhook / reactivate_webhook
• Schemas: register_schema / list_schemas / get_latest_schema / delete_schema
• Functions: register_function / get_function / delete_function (JavaScript / WASM)
• Plan: get_plan / set_plan; usage: get_usage
• Settings: get_settings / set_strict_topics (data-plane topic gate toggle)
• Billing (crypto PAYG prepaid): get_billing / get_deposit_addresses / get_deposits / get_ledger / get_usage_debits / get_deposit_status
• Misc: get_snippet / get_docs / get_audit, DLQ dlq_peek / dlq_replay

Optional capability: stream_ws (WebSocket) is provided via the optional dependency websocket-client (pip install msgmesh[ws]); the base install doesn't include it and other features are unaffected. Server-side, SSE (stream) + long-poll (subscribe) are also available. Registration and admin go through panel sessions, not this SDK.

Design notes

• HTTP client uses httpx: it supports both regular requests and streaming (SSE), fitting this SDK's realtime needs; it's more modern than requests (typing, streaming context managers) and nicer than urllib. You can pass a transport (httpx.BaseTransport) into the constructor to stub it for tests, mirroring sdk-js's fetchImpl.
• Synchronous API: no async for now; background loops (subscribe/stream) run on daemon threads and return a stop callable.

Development

pip install -e ".[dev]"
pytest

Publishing (maintainers)

Publishing runs via GitHub Actions (.github/workflows/release-sdk-py.yml), authenticated with a PyPI API token stored as the repo secret PYPI_API_TOKEN.

1. Bump the version: change both [project].version in pyproject.toml and __version__ in src/msgmesh/__init__.py to the new version (the two must match).
2. Merge to master.
3. Tag and push: git tag sdk-py-v<version> (e.g. sdk-py-v0.1.0) → git push origin sdk-py-v<version>. The workflow then builds (python -m build) + tests (pytest) + twine check + publishes to PyPI.

An already-published version fails (never clobbers a released version). The sdk-py-v* tag prefix is exclusive to this package and doesn't collide with the npm packages' sdk-v* / mcp-v* / cli-v*.

---

msgmesh (Python SDK) · 繁體中文

English | 繁體中文

MsgMesh 的 Python SDK — 多租戶事件總線的收發 / 即時(SSE / WebSocket)/ 治理 client。 這是 TypeScript SDK(@msgmesh/sdk)的 Python 移植版: 對外 API 面對齊,方法名採 Python 慣例(snake_case),語意/涵蓋範圍與 sdk-js 一致。HTTP 契約 以平台 OpenAPI(packages/shared-go/openapi/openapi.yaml)為權威來源。

安裝

pip install msgmesh

需要 Python ≥ 3.9;唯一執行期依賴為 httpx。

快速開始

先在面板註冊帳號、簽發一把 API key(明文僅顯示一次),再:

from msgmesh import MsgMesh

mq = MsgMesh(
    api_key="mk_live_...",                  # 伺服器端用長期 key
    control_plane_url="https://cp.example.com",
    gateway_url="https://gw.example.com",
    realtime_url="https://rt.example.com",
)

mq.create_topic("orders")
mq.publish("orders", {"hello": 1})
msgs = mq.poll("orders", group="g1")
for m in msgs:
    print(m.value)

MsgMesh 也是 context manager,離開時關閉底層連線池:

with MsgMesh(api_key="mk_live_...") as mq:
    mq.publish("orders", {"hello": 1})

瀏覽器/不可信端不適用;伺服器代拿短期 token

get_token 對應 sdk-js 的 token-broker 模式:給一個「去後端拿短期 dp token」的 callable (回傳 {"token": ..., "expires_in": ...} 或 TokenResponse),SDK 會自動快取、將過期前重取, SSE 重連時亦換新。與 api_key 二擇一,至少需其一。

import httpx
from msgmesh import MsgMesh

def fetch_token():
    return httpx.get("https://my-backend/mm-token").json()  # {"token": ..., "expires_in": 300}

mq = MsgMesh(
    get_token=fetch_token,
    gateway_url="https://gw.example.com",
    realtime_url="https://rt.example.com",
)

即時接收

介面一致、皆回傳「停止用 callable」,並在背景 daemon thread 執行:

• subscribe(topic, handler, *, group=None, max=None, on_error=None):長輪詢(poll 迴圈)。 終態 vs 可恢復——只有 401(金鑰失效/不存在=終態)才永久停止;403(可自助充值解封)與其他 暫時性錯誤回報後退避續試(自癒)。get_token 模式的 401 多半只是 token 過期,失效快取後以新 token 續試,連續多次仍被拒才判定永久撤權。永久停止時若未提供 on_error,會記一則 logging warning。

  stop = mq.subscribe("room.42", lambda m: print(m.value), group="g1", on_error=print)
  # ...
  stop()  # 停止輪詢
  

• stream(topic, on_message, on_error=None, room=None):透過 SSE 即時接收(連線鑑權走 query key)。 on_message 收到每則事件的文字內容;on_error 收到 Exception(連線層/非 2xx)或 StreamClose(伺服器具名關閉事件,.data 為原因字串)。

  stop = mq.stream("room.42", print)
  # ...
  stop()
  

  與 sdk-js 的差異:sdk-js 的 stream 依賴瀏覽器原生 EventSource(有原生自動重連)。 Python 無 EventSource,故本 SDK 自管重連(對齊 sdk-js streamWs 的精神):每次串流結束/ 連線錯誤退避後重連,get_token 模式換新 token,成功連上重置失敗計數,連續失敗達上限即停 (避免對死 key/端點無限重連);"authorization revoked" 為終態,主動停止。

• stream_ws(topic, on_message, on_error=None, room=None):透過 WebSocket 即時接收(連線鑑權走 query key)。 介面與 stream 一致;需選用依賴 websocket-client:

  pip install "msgmesh[ws]"
  

  from msgmesh import WsClose
  
  def on_err(e):
      if isinstance(e, WsClose):
          print("closed", e.code, e.reason)   # 例如 1008 / "authorization revoked"
      else:
          print("error", e)                    # 連線層 / 握手失敗
  
  stop = mq.stream_ws("room.42", print, on_error=on_err)
  # ...
  stop()
  

  這是純 Python 伺服器端能力,與瀏覽器 / Node 無關(sdk-js 的 streamWs 才涉及瀏覽器/Node ≥ 22 的全域 WebSocket)。未安裝 websocket-client 時呼叫 stream_ws 會立即 ImportError 附安裝提示。

  與 stream(SSE)的差異:WebSocket 無原生重連,故重連一律由 SDK 接管——退避重連、成功連上 重置失敗計數、兩模式皆對連續失敗設上限(達上限停止,未提供 on_error 時記 warning),get_token 另在重連前換新 token。撤權:連線中為 CLOSE 1008 + "authorization revoked"(終態,立即停); 握手期(HTTP 401)為關閉碼 1006,由有界重連收口。on_error 收到 Exception(連線層/握手失敗) 或 WsClose(關閉事件,可讀 .code/.reason);stop() 主動關閉不觸發 on_error。 不想加依賴時,伺服器端用 stream(SSE)或 subscribe(長輪詢)即可覆蓋即時接收。

多房間(rooms)

一個 topic 內可再切多個房間(room = Kafka record key),脫鉤「房間數」與「topic 數」。分兩層:

① 路由——發佈時用 publish(topic, body, key=room_id) 指定房間,訂閱時傳選用 room(stream / stream_ws 皆同)只收該房間:

stop = mq.stream("chat", print, room="room-42")      # 只收 room-42 的訊息
stop = mq.stream_ws("chat", print, room="room-42")   # WebSocket 同理
mq.publish("chat", {"text": "hi"}, key="room-42")    # 發到 room-42

省略 room=收該 topic 全部訊息(向後相容)。路由本身只做過濾、無強制隔離——惡意 client 可改成別人的 room 偷聽同 topic 其他房間。要真隔離看 ②。

② 隔離(平台強制)——把憑證的 capabilities 加上選用 rooms,平台即強制該憑證只能收發指定房間 (逾越 403)。rooms 省略/空 = 所有房間(向後相容);非空 = 僅限這些。典型作法是後端持一把全房間金鑰, 向 POST /v1/tokens 降權簽出「只准某房間」的短期 token 給前端(降權只准更窄、須為金鑰能力子集,逾越 403):

import httpx

# 後端 token-broker:用全房間 key 降權鑄「只准 chat / room-42」的短期 token,回給前端當 get_token
def mint_room_token():
    r = httpx.post(
        "https://cp.example.com/v1/tokens",
        headers={"Authorization": "Bearer mk_live_..."},   # 後端持有的全房間長期 key
        json={
            "ttl_seconds": 600,
            "capabilities": [{"ops": ["subscribe", "publish"], "topics": ["chat"], "rooms": ["room-42"]}],
        },
    )
    return r.json()   # {"token": ..., "expires_in": 600}

也可用 create_key("key", capabilities=[{"ops": [...], "topics": [...], "rooms": [...]}]) 簽一把常駐 room-scoped 鍵。平台強制點:訂閱(SSE/WS)必須帶允許集內的 ?room(不帶=想收全部房間,一樣 403); 發佈的 ?key 必須 ∈ 允許集。

⚠️ room-scoped 憑證只能走即時(SSE/WS)+ 對其房間 publish;不能 poll / consume / DLQ。 後者是整個 topic 的 firehose(consumer-group offset 會吃掉別房間、每房一 group = 讀取放大),無法乾淨 per-room 過濾,受限房間憑證一律 403(use realtime SSE/WS ?room=)。需要 poll/consume 時請改用不限房間的憑證。

房間隔離的安全須知(必讀)

• 隔離強度 = 你發的 token 範圍。 只有在「後端用全房間金鑰降權鑄 room-scoped token 給前端」時才有隔離。別把不限房間的憑證(全權 key、或沒有 rooms 的 token)放進前端 / 不可信端——那樣對方改個 room 就能看到所有房間,隔離形同虛設。
• 平台不驗「發訊者是誰」。 房間隔離管的是「能收發哪些房間」,不是「你是房裡的誰」。同一房內,任何持該房 token 的人都能在 payload 裡冒充任何 sender。要防房內冒名:後端為每個使用者各自鑄 token、並由後端戳上 / 驗證 sender,別讓不可信端自報身分。
• 附帶:presence(在線數)目前是 per-topic 非 per-room(只洩漏聚合數字);短期 token 為 bearer,洩漏 = 該房 ≤TTL 可用(故 TTL 短、勿記進 log)。

錯誤處理

非 2xx 回應會依狀態碼拋型別化錯誤(都繼承 MsgMeshError,帶 status/code/path/request_id):

from msgmesh import ValidationError, RateLimitError

try:
    mq.create_topic("Bad Name!")
except ValidationError as e:
    print("參數不合法:", e)
except RateLimitError:
    print("被限流,稍後重試")

狀態碼	型別	code
400 / 422	ValidationError	validation
401 / 403	AuthError	auth
404	NotFoundError	not_found
429	RateLimitError	rate_limit
其他	MsgMeshError	server
連線層失敗	MsgMeshConnectError	connect

API 一覽(對照 sdk-js)

方法名為 sdk-js 的 camelCase → Python snake_case,回傳型別為 dataclass(屬性存取,如 topic.name)。

• Topics:create_topic / list_topics / delete_topic
• 收發:publish / poll / subscribe(輪詢)/ stream(SSE)/ stream_ws(WebSocket,選用 msgmesh[ws])/ get_presence
• Keys:list_keys / create_key(可帶 scope + capabilities)/ delete_key
• Webhooks:list_webhooks / create_webhook / delete_webhook / reactivate_webhook
• Schemas:register_schema / list_schemas / get_latest_schema / delete_schema
• Functions:register_function / get_function / delete_function(JavaScript / WASM)
• 方案:get_plan / set_plan;用量:get_usage
• 設定:get_settings / set_strict_topics(資料面 topic 閘門開關)
• 帳務(加密貨幣 PAYG 預付):get_billing / get_deposit_addresses / get_deposits / get_ledger / get_usage_debits / get_deposit_status
• 其他:get_snippet / get_docs / get_audit、DLQ dlq_peek / dlq_replay

選用能力:stream_ws(WebSocket)透過選用依賴 websocket-client(pip install msgmesh[ws])提供; base 安裝不含此依賴、也不影響其他功能。伺服器端另有 SSE(stream)+ 長輪詢(subscribe)可用。 註冊、超管走面板 session,不在本 SDK。

設計說明

• HTTP client 用 httpx:同時支援一般請求與串流(SSE),契合本 SDK 的即時需求;比 requests 更現代(型別、streaming context manager),且較 urllib 好用。可在建構式傳入 transport (httpx.BaseTransport)以打樁測試,對應 sdk-js 的 fetchImpl。
• 同步 API:先不做 async;背景迴圈(subscribe/stream)以 daemon thread 執行並回傳停止 callable。

開發

pip install -e ".[dev]"
pytest

發布(維護者)

發布走 GitHub Actions(.github/workflows/release-sdk-py.yml),認證用存於 repo secret PYPI_API_TOKEN 的 PyPI API token。

1. bump 版本:pyproject.toml 的 [project].version 與 src/msgmesh/__init__.py 的 __version__ 一起改成新版號(兩處需一致)。
2. 合併 master。
3. 打 tag 並 push:git tag sdk-py-v<版本>(如 sdk-py-v0.1.0)→ git push origin sdk-py-v<版本>。 workflow 隨即 build(python -m build)+ test(pytest)+ twine check + 發布到 PyPI。

版本已存在於 PyPI 會 fail(不覆蓋已發版本)。tag 前綴 sdk-py-v* 為本套件專屬,不與 npm 套件的 sdk-v* / mcp-v* / cli-v* 撞車。