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 — plus its StreamMessageMeta when the callback declares a second parameter (see below); 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.

  Resume on reconnect (at-least-once, no gaps). Both stream (SSE) and stream_ws (WebSocket) resume across reconnects: each message carries a <partition>-<offset> cursor, the SDK tracks the last one seen, and on reconnect asks the server to replay from there — messages dropped during a disconnect are backfilled, not lost. Delivery is at-least-once: the SDK dedupes per-partition by cursor, so a rare overlap is suppressed rather than delivered twice. If the server can't cover the gap (older than the replay window) it emits a resync signal — pass on_resync to be told to re-fetch a snapshot. Transparent: on_message still receives the raw value string. (Requires the platform's realtime resume tier; against an older server it degrades to live-tail.)

  on_message also hands you the message's coordinates. Declare a second parameter and you get on_message(value, meta), where meta is a StreamMessageMeta — id / partition / offset. meta.id is character-for-character the same <partition>-<offset> that history() puts on every message, so "replay history, then join the live stream" dedupes with a single set instead of comparing message bodies — body comparison quietly swallows one of two genuinely identical messages ("ok" sent twice in a chat room). Backward compatible: the SDK inspects your callback and passes meta only when it declares a second positional parameter, so one-parameter callbacks (including print and list.append) are called exactly as before. meta is None only for a frame that carried no parseable cursor (a non-msgmesh server); it is still delivered.

  There is deliberately no ts in meta: the live wire carries no timestamp (an SSE frame has room for an id and the payload, nothing else), and reading the local clock would be an arrival time wearing a produce time's name. Read HistoryMessage.ts when you need the real timestamp.

  mq.stream("room.42", lambda value, meta: print(meta.id if meta else "-", value))
  

Rooms

A single topic can be split into multiple rooms (a room is one partition key under the hood), decoupling "number of rooms" from "number of topics". Two layers:

① Routing — publish with publish(topic, body, room=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"}, room="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); the publish room 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).

Safe retries: idempotency_key

A publish that times out leaves you guessing: did it land? Pass a unique idempotency_key (a UUID per message is the usual choice) and you can just retry. The platform remembers the outcome for 24 hours and returns the original partition/offset instead of writing a second copy — and it does not meter the duplicate either.

import uuid

key = str(uuid.uuid4())
mq.publish("orders", {"order_id": 1001}, idempotency_key=key)
# Timed out? Same call, same key — at most one message exists either way.
mq.publish("orders", {"order_id": 1001}, idempotency_key=key)

• One key per message. Reusing a key with a different body or room raises ValidationError (422) rather than silently handing back the old result — otherwise the second message would vanish behind a 200.
• Concurrent resend → 409. While the first attempt is still in flight, a duplicate is rejected with no side effects at all; retry a moment later to read its result.
• Best effort, not a guarantee. Delivery is at-least-once by design, and if the dedup store is briefly unavailable the platform publishes anyway rather than failing your write. Keep consumers idempotent regardless.

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

History: don't leave late joiners staring at a blank screen

A live stream starts at "now". Someone who opens your chat room, support thread, or event feed after the fact sees nothing. history() fetches the most recent messages and hands you a cursor to resume the live stream from — no gap, no duplicates.

seen = set()

def render(msg_id: str, value: str) -> None:
    if msg_id in seen:      # at-least-once: history and the stream deliberately overlap
        return
    seen.add(msg_id)
    print(value)

# 1. Fetch the last 50 messages of this room (oldest → newest).
page = mm.history("chat", room="room-42", limit=50)
for m in page.messages:
    render(m.id, m.value)

# 2. Resume live from where history ended, deduping by the same id. `resume_from` was
#    replayable when it was issued.
mm.stream("chat", lambda value, meta: render(meta.id, value) if meta else print(value),
          room="room-42", resume_from=page.resume_from or None)

Two cursors, two meanings — this is the part people get wrong:

field	what it is for	guarantee
resume_from	pass to stream / stream_ws as resume_from	it was inside the replayable window when the response was produced
before	pass back to history for the previous page	none — never hand it to stream

Two details about resume_from that save debugging time later:

• It is not necessarily the id of the last message you received. The scan runs newest → oldest, so the server already knows the range above your newest message holds nothing for this room, and it advances the cursor there on purpose — that leaves the largest possible margin for however long your app takes to open the stream.
• resume_gap=True means resuming would skip messages: you paged further back than the replayable window, retention deleted the middle, or this was a whole-topic (no room) query on a multi-partition topic. When it is False, nothing is skipped — the server only reports a gap it actually has.

Paging further back:

cursor = page.before
while cursor:
    older = mm.history("chat", room="room-42", limit=50, before=cursor)
    for m in older.messages:
        render(m.id, m.value)
    cursor = older.before        # "" = nothing older

That loop terminates: before enumerates every partition the scan covered, so each round strictly moves back and no page is ever repeated.

complete=True means the scan stopped because it satisfied your request — it either filled the page to limit, or it reached the bottom of what exists. It is not "the page was filled to limit", and it is not "there is nothing older". A room with only 3 messages answers a limit=50 query with complete=True and 3 messages; so does a busy room answering with a full page of 50 while thousands more sit further back.

To find out whether there is another page, read before, never complete. A non-empty before means more is reachable; "" means you have reached the bottom. (limit is silently capped server-side, so len(messages) < limit is not a reliable end-of-history test either.)

complete=False means the scan was cut short by something other than your request, and incomplete_reason says by what: budget (the older messages are still in Kafka, this scan just did not reach them — page back with before; it covers the bounded record scan, the per-page byte cap, and a stalled broker) or retention (you passed a since bound reaching further back than the platform still keeps; that stretch is deleted for good — you only ever see this when you asked for a specific older range, a plain "latest N" query is always complete=True). The difference that matters: budget is recoverable by paging, retention is not.

Timestamps in before / since are epoch milliseconds (or RFC3339). Passing epoch seconds is rejected with 400 rather than silently answering about 1970 — pass a datetime and the SDK gets it right for you (naive datetimes are treated as UTC).

What this is, and is not. Under the hood this is a bounded scan over the event log, not a separate history database. Two consequences worth designing around: how far back you can reach is capped by your plan's retention period, and passing room is dramatically cheaper than scanning the whole topic (a room lives on a single partition). If you need audit-grade access to messages older than your retention window, keep your own copy.

The SDK stores nothing. Where a cursor belongs — memory, a file, your own backend — is your app's decision. Keep resume_from wherever fits and pass it back next time.

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 (optional room + idempotency_key for safe retries) / poll / subscribe (polling) / stream (SSE) / stream_ws (WebSocket, optional msgmesh[ws]) / get_presence / history (recent messages + a cursor to resume the live stream from)
• 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 收到每則事件的文字內容——回呼若宣告第二個參數,還會收到該則的 StreamMessageMeta (見下);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(長輪詢)即可覆蓋即時接收。

on_message 同時給你這則訊息的座標。 回呼多宣告一個參數就會收到 on_message(value, meta), meta 為 StreamMessageMeta(id / partition / offset)。meta.id 與 history() 每則訊息上的 <partition>-<offset> 逐字相同,這正是「先回放歷史、再接即時流」能只用一個 set 去重的原因—— 否則只能拿內容硬比,而兩則內容一模一樣的訊息(聊天室裡連送兩次「ok」)會被誤吞一則。 向後相容:SDK 會內省你的回呼,只有宣告了第二個位置參數才帶 meta,故既有只吃一個參數的回呼 (含 print、list.append)呼叫方式完全不變。只有當該則訊息沒帶得出可解析的游標時(非 msgmesh 伺服器) meta 才是 None,而訊息照樣投遞。

meta 刻意沒有 ts:即時線上根本不帶時間戳(SSE 一則事件只放得下 id 與 payload), 在客戶端補一個當下時鐘,等於把「到達時間」冠上「產生時間」的名字。需要真時間戳請讀 HistoryMessage.ts。

mq.stream("room.42", lambda value, meta: print(meta.id if meta else "-", value))

多房間(rooms)

一個 topic 內可再切多個房間(底層就是一個分割鍵),脫鉤「房間數」與「topic 數」。分兩層:

① 路由——發佈時用 publish(topic, body, room=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"}, room="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); 發佈的 room 必須 ∈ 允許集。

⚠️ 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)。

安全重送:idempotency_key

publish 逾時的當下你是沒有答案的:到底送出去了沒?帶一把唯一的 idempotency_key(慣例是每則訊息一個 UUID)就可以直接重送——平台會記住這把鍵 24 小時,同鍵重送直接回原本的 partition/offset,不會寫進第二份,也不會重複計量。

import uuid

key = str(uuid.uuid4())
mq.publish("orders", {"order_id": 1001}, idempotency_key=key)
# 逾時了?同一個呼叫、同一把鍵再送一次——不論如何最多只會存在一則。
mq.publish("orders", {"order_id": 1001}, idempotency_key=key)

• 一則訊息一把鍵。 同鍵配不同 body/room 會拋 ValidationError(422),而不是靜默把舊結果還給你——否則第二則訊息會躲在 200 後面憑空消失。
• 併發重送回 409。 前一次還在進行中時,重送直接被擋,沒有任何副作用;稍後再試就能讀到它的結果。
• 是輔助,不是保證。 平台的投遞語意本來就是 at-least-once;去重服務短暫不可用時會選擇照常發布,而不是讓你的寫入失敗。消費端仍要能容忍重複。

錯誤處理

非 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

歷史訊息:別讓晚到的人面對一片空白

即時串流是從「現在」開始的。晚一步打開聊天室、工單、事件流的人,看到的是空的。 history() 取回最近的訊息,並交給你一個游標,讓你無縫接上即時串流——不漏、不重複。

seen = set()

def render(msg_id: str, value: str) -> None:
    if msg_id in seen:      # at-least-once:歷史與串流刻意有重疊,依 id 去重
        return
    seen.add(msg_id)
    print(value)

# 1. 取這個房間最近 50 則(由舊到新)。
page = mm.history("chat", room="room-42", limit=50)
for m in page.messages:
    render(m.id, m.value)

# 2. 從歷史的結尾接上即時,並用同一個 id 去重。resume_from 在發出的當下必定可續傳。
mm.stream("chat", lambda value, meta: render(meta.id, value) if meta else print(value),
          room="room-42", resume_from=page.resume_from or None)

兩個游標語意不同——這是最容易搞錯的地方:

欄位	用途	保證
resume_from	交給 stream / stream_ws 的 resume_from	在回應產生的當下必定落在可續傳窗內
before	交回 history 取更舊的一頁	沒有——絕對不要拿去餵 stream

關於 resume_from,兩個之後會省你除錯時間的細節:

• 它不一定等於你收到的最後一則的 id。掃描是由新到舊的,伺服器已經知道「你最新那一則之上」 那段沒有這個房間的訊息,於是刻意把游標推到那裡——讓你的 app 從「拿到歷史」到「連上串流」 之間的容忍時間最大化。
• resume_gap=True 代表續傳真的會漏掉一段:你往回翻得比可續傳窗還舊、中間被保留期刪掉, 或這是多 partition 的全 topic(不帶 room)查詢。為 False 時就是沒有漏。

往前翻頁:

cursor = page.before
while cursor:
    older = mm.history("chat", room="room-42", limit=50, before=cursor)
    for m in older.messages:
        render(m.id, m.value)
    cursor = older.before        # "" = 沒有更舊的了

這個迴圈保證會結束:before 會列出本次掃描的每一個 partition,故每一圈都嚴格往舊推進、 不會重複回同一頁。

complete=True 代表「你要的範圍內拿得到的都在這一頁了」,不是「裝滿了 limit 則」。 一個總共只有 3 則的房間查 limit=50,回的就是 complete=True + 3 則。 只有 complete=False 才代表有東西沒給到,incomplete_reason 說明是哪一種: budget(更舊的還在 Kafka 裡,只是這次沒掃到,帶 before 再翻一頁;涵蓋則數掃描、單頁位元組 上限、broker 一時卡住三種)、retention(已被方案保留期刪掉,再問也沒有)。 差別在可否補救:budget 翻頁就有,retention 沒了就是沒了。

before / since 的時間是 epoch 毫秒(或 RFC3339)。傳成 epoch 秒會回 400, 而不是默默回答 1970 年的事——直接傳 datetime(naive 視為 UTC),SDK 會幫你轉對。

它是什麼、不是什麼:底下是對事件日誌的有界回掃,不是獨立的歷史資料庫。兩個要納入設計的 結果:能回溯多遠受方案保留期約束;帶 room 遠比掃整個 topic 便宜(一個 room 落在單一 partition)。 需要稽核等級地存取保留期以外的訊息,請自行留存副本。

SDK 不替你保存任何東西:游標該放哪裡(記憶體、檔案、你自己的後端)是你的 app 的決定。

API 一覽(對照 sdk-js)

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

• Topics:create_topic / list_topics / delete_topic
• 收發:publish(可帶 room 與 idempotency_key 安全重送)/ poll / subscribe(輪詢)/ stream(SSE)/ stream_ws(WebSocket,選用 msgmesh[ws])/ get_presence / history(最近訊息 + 接回即時串流的游標)
• 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* 撞車。