Metadata-Version: 2.5
Name: msgmesh
Version: 0.7.0
Summary: MsgMesh Python SDK — publish / consume / realtime (SSE / WebSocket) / governance client for the multi-tenant event bus (Python port of sdk-js).
Project-URL: Homepage, https://msgmesh.alderflux.com
Project-URL: Issues, https://github.com/LukeLogix/msgmesh-examples/issues
Author: LukeLogix
License: MIT License
        
        Copyright (c) 2026 LukeLogix
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: event-bus,kafka,message-queue,msgmesh,pubsub,realtime,sdk,sse
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.9
Requires-Dist: httpx<1,>=0.27
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: websocket-client>=1.7; extra == 'dev'
Provides-Extra: ws
Requires-Dist: websocket-client>=1.7; extra == 'ws'
Description-Content-Type: text/markdown

# msgmesh (Python SDK)

**English** | [繁體中文](#msgmesh-python-sdk--繁體中文)

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`](https://www.npmjs.com/package/@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.

**Upgrading from 0.6.x?** No code change is required, but some behaviour differs — see
[Upgrading from 0.6.x](#upgrading-from-06x).

## Install

```sh
pip install msgmesh
```

Requires Python ≥ 3.9; the only runtime dependency is [`httpx`](https://www.python-httpx.org/)
(0.27 or later, below 1.0 — httpx 1.0 is not supported yet).

## Quick start

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

```python
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:

```python
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.

```python
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), 429 and other transient errors are
  reported then retried with backoff (self-healing; see "Reconnect backoff" below). 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 five consecutive
  rejections. The one other thing that stops it is a `get_token` that raises
  `TokenUnavailableError` (see "Permanent token failure" below). On permanent stop, if no
  `on_error` is given, a `logging` warning is emitted.

  ```python
  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).

  ```python
  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, and resets the backoff once a
  > connection has lasted 5 seconds or delivered a message. **It does not give up after a number
  > of attempts** (see "Reconnect backoff" below); `"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`:

  ```sh
  pip install "msgmesh[ws]"
  ```

  ```python
  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 in both modes — backoff reconnect, reset once a connection has lasted 5
  > seconds or delivered a message, and
  > **no giving up after a number of attempts** (see "Reconnect backoff" below); `get_token`
  > also rotates the token before reconnecting. **Revocation**: mid-connection it is CLOSE
  > `1008` + `"authorization revoked"` (terminal, stops immediately); a refused handshake is
  > retried with backoff like any other failure. `on_error` receives an
  > `Exception` (transport / handshake failure) or a `WsClose` (a close event with readable
  > `.code`/`.reason`); `stop()` ends it without triggering `on_error`, and returns at once even
  > when the peer has gone silent (it shuts the socket down instead of waiting for the peer to
  > answer a closing handshake). If you'd
  > rather not add the dependency, `stream` (SSE) or `subscribe` (long-poll) cover realtime
  > receive server-side.

  **Reconnect backoff: it keeps trying.** `subscribe`, `stream` and `stream_ws` all manage their own
  retries. The wait between attempts starts at up to 1s and doubles with each further consecutive
  failure, up to 30s; the actual wait is a random value in the upper half of that ceiling, so
  clients that failed together do not all come back together. A success resets it to 1s: for
  `subscribe`, a poll that returned; for `stream` and `stream_ws`, a connection that stayed open
  for 5 seconds or delivered a message. Merely connecting does not count — otherwise something
  that accepts a connection and drops it at once would be reconnected to every second,
  indefinitely.

  `stream` and `stream_ws` **do not give up after a number of attempts**: a refused connection may
  be a rate limit that lifts on its own (HTTP 429 — a connection limit, for instance), and stopping
  would leave the stream dead after the limit is gone. They end in three cases only: on the
  server's explicit revocation (`"authorization revoked"`); when you call the stop callable, which
  also cuts a pending wait short (and, for `stream`, interrupts a read that is waiting for the next
  event); and when your `get_token` raises `TokenUnavailableError` (below).
  `on_error` is called on every failure, with one exception: when the server ends a `stream`
  (SSE) response normally — the body simply finishes, no error — the SDK backs off and reconnects
  without calling `on_error` (`stream_ws` has no such silent case: it reports every connection
  that closes). A refused connection arrives as a typed error for
  `stream` (`AuthError`, `RateLimitError`, …) and as websocket-client's handshake exception for
  `stream_ws`, so you can stop yourself on a status you consider final. Nothing is reported once
  you have called the stop callable — not even a token fetch, a connection attempt or a handshake
  that was in flight at that moment and fails afterwards. The SDK does not do that
  for you, not even on a 401: on a realtime connection the platform also answers 401 while the
  tenant is suspended or while it cannot check the key at that moment, both of which pass. If you
  pass no `on_error`, the fifth consecutive failure logs a warning (logger `msgmesh`) saying it is
  still retrying — once per run of consecutive failures, so a later run of five logs it again.

  **In `get_token` mode, end the stream yourself when the session ends.** Every retry calls your
  `get_token` for a fresh token — settling at one call every 15–30 seconds for as long as nothing
  recovers — and the stream does not stop on its own count of failures. Once you know the
  credential is gone for good, do one of two things:

  - **Call the stop callable** the method returned. Nothing further is reported.
  - **Raise `TokenUnavailableError` from `get_token`.** The SDK stops that subscription by itself
    and tells its `on_error`, once.

  **Permanent token failure: `TokenUnavailableError`.** It is the one exception your `get_token`
  can raise to say "no token can be obtained any more":

  ```python
  import httpx
  from msgmesh import MsgMesh, TokenUnavailableError

  def get_token():
      r = httpx.post("https://my-backend.example/api/mm-token", headers=session_headers())
      if r.status_code == 401:  # your backend's way of saying the session is over
          raise TokenUnavailableError("signed out")
      r.raise_for_status()      # anything else is temporary: the SDK retries
      return r.json()

  def on_error(err):
      if isinstance(err, TokenUnavailableError):
          ask_the_user_to_sign_in()  # this stream has stopped and will not retry

  mq = MsgMesh(get_token=get_token, realtime_url="https://rt.example.com")
  mq.stream("room.42", print, on_error)
  ```

  - **Raise it when** asking again cannot help: your backend answers that the user has signed out,
    the refresh token was rejected, the account is closed.
  - **Do not raise it for** anything that may pass — a network error on the way to your backend, a
    5xx or a 429 from it. Raise any other exception (or let the HTTP client's own error through):
    the SDK treats it as temporary and retries with the backoff above, exactly as it did before
    this exception existed.
  - **What the SDK does.** `subscribe`, `stream` and `stream_ws` stop that subscription — no
    further retry, no further call to `get_token`, and the background thread ends — and hand the
    exception to `on_error` once; without an `on_error` they log one warning (logger `msgmesh`). A
    subscription whose stop callable you already called reports nothing. One-off calls (`publish`,
    `history`, …) raise the exception, unchanged.
  - **It is not broadcast, and nothing is remembered.** Other subscriptions on the same client
    carry on until they next need a token themselves — at their own reconnect or their own next
    poll — and stop then, when `get_token` raises again. Every later need for a token calls
    `get_token` again.
  - A subclass counts. It is a `MsgMeshError` (so `except MsgMeshError` still catches everything
    the SDK can raise) with `status` 0 and `code` `"token_unavailable"`; `str(err)` is your message
    as given. Use `raise TokenUnavailableError("…") from err` to keep the cause. (In sdk-js the
    equivalent class extends `Error` and is also recognised by its `name`, because a bundler can
    end up with two copies of that SDK; Python has no such case.)

  `subscribe` keeps its two terminal cases (401 with an API key; five consecutive 401s in
  `get_token` mode) and retries everything else with the same backoff.

  When a 429 response says how long to wait (`retry_after` in the body first, then the
  `Retry-After` header), all three wait that long if it is longer than the backoff, capped at 300s.
  Not every 429 says so; without it the backoff applies. For a refused `stream_ws` handshake the
  body can be read only when the response states its `Content-Length`; otherwise only the header
  counts. For HTTP responses the value is also on the error as
  `RateLimitError.retry_after` (`None` when the response did not say; never more than 86400).
  This is one place where the two SDKs behave differently: sdk-js's `stream` / `streamWs` sit on
  `EventSource` / `WebSocket`, which reveal neither the status code nor the headers of a refused
  connection, so there only `subscribe` can follow a `Retry-After`.

  Known limits of the reconnect logic:

  - **A peer that delivers one new message on every connection and then closes it** resets the
    backoff each round, so that loop is reconnected to about once a second — and in `get_token`
    mode your `get_token` is called just as often. A message the SDK has already seen (a resend it
    dedupes) does not reset it.
  - **`stream`: a server that accepts the connection but never sends response headers.** The stop
    callable returns at once and nothing is reported afterwards, but the background thread stays
    blocked until the server answers or drops the connection (waiting for a response has no
    timeout, so that a quiet stream is never cut). It is a daemon thread and does not keep the
    process alive.

  **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.

  ```python
  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:

```python
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):

```python
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.

```python
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.

## Client identification (`User-Agent`)

The SDK labels everything it sends, so the platform can tell its traffic apart from other clients':

```
User-Agent: msgmesh-sdk-py/<SDK version>
```

Pass `app_info` to put your own application in front of it:

```python
mq = MsgMesh(api_key=key, app_info={"name": "order-worker", "version": "1.4.2"})
# User-Agent: order-worker/1.4.2 msgmesh-sdk-py/<SDK version>     ("version" is optional)
```

- `name` and `version` must be User-Agent product tokens: letters, digits and ``!#$%&'*+-.^_`|~``
  only — no spaces, `/` or `@` — and at most 128 characters each. An invalid value, a missing
  `name`, or an `app_info` that is not a mapping raises `ValueError` from the constructor.
- The SDK's own label is `msgmesh-sdk-py`; the JavaScript SDK's is `msgmesh-sdk-js`. Treat them as
  fixed strings if you filter on them.
- The header goes on every request the SDK makes: every HTTP call, the SSE connection opened by
  `stream`, and the WebSocket handshake of `stream_ws`. All of it is new in 0.7.0: up to 0.6.x,
  HTTP and SSE requests carried httpx's default `User-Agent` (`python-httpx/<version>`) and the
  WebSocket handshake carried none.
- **Names starting with `msgmesh-` are reserved** for MsgMesh's own clients (`msgmesh-panel`,
  `msgmesh-mcp`, `msgmesh-cli`, and any added later) — please do not use that prefix for your
  application. The constructor does not reject it, but a name that is exactly one of those makes
  the platform count your traffic as that client's (the platform compares these names without
  regard to letter case).
- The label is self-reported and informational. It plays no part in authentication or routing.
- `msgmesh.__version__` is the SDK version, should you want to log it.

(sdk-js sends the same kind of label, `msgmesh-sdk-js/<version>`, three ways: as `User-Agent`
outside browsers; as an `X-MsgMesh-Client` header in browsers, where the `User-Agent` is the
browser's own; and as a `client` query parameter on its EventSource / WebSocket connections, which
cannot carry a header there — without the caller's `appInfo`, because a URL ends up in access logs.
This SDK runs server-side, where `User-Agent` is always available, so it uses that one header
throughout, sends neither of the other two, and never puts your `app_info` into a URL.)

## Upgrading from 0.6.x

No code change is required, but these behave differently:

- **`stream` / `stream_ws` no longer stop after five consecutive failures.** They used to give up;
  now they keep retrying with backoff (see "Reconnect backoff" above) until the server revokes
  the credential, you call the stop callable, or `get_token` raises `TokenUnavailableError`. In
  `get_token` mode every retry calls your `get_token`.
- **Retries wait longer.** The wait after a failure used to be a fixed 1s; it now doubles with
  each consecutive failure, up to 30s — and all three methods (`subscribe`, `stream`,
  `stream_ws`) wait up to 300s when a 429 response says how long. After an outage, messages can
  resume that much later.
- **The `User-Agent` changes** from httpx's default (`python-httpx/<version>`) to
  `msgmesh-sdk-py/<SDK version>` on every HTTP and SSE request; the WebSocket handshake of
  `stream_ws`, which carried none, now carries it too. See "Client identification" above.
- **The stop callable acts at once.** It interrupts a `stream` read that is waiting for the next
  event, and for `stream_ws` it no longer waits for the peer to answer the closing handshake.
  Nothing is reported after it was called.
- **The `httpx` requirement now has an upper bound** (`httpx>=0.27,<1`). httpx 1.0 changes how a
  client is constructed and this SDK does not work on it yet; up to 0.6.x the requirement had no
  upper bound, so a fresh install could pick up a 1.0 release and fail the moment a client is
  created. If something else you install needs httpx 1.x, the two cannot coexist: asked for both
  in one install, pip refuses with a dependency conflict; installed into an environment that
  already has httpx 1.x, pip replaces it with a 0.x release.

New in 0.7.0: `TokenUnavailableError`, the `app_info` option and `RateLimitError.retry_after`.

## Error handling

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

```python
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` |
| 409 (`code: operation_in_progress`) | `OperationInProgressError` | `operation_in_progress` |
| other | `MsgMeshError` | `server` |
| transport failure | `MsgMeshConnectError` | `connect` |
| raised by your own `get_token` | `TokenUnavailableError` | `token_unavailable` |

`TokenUnavailableError` never comes from the platform: you raise it from your own `get_token` to say no token
can be obtained any more — see "Permanent token failure" under Realtime receive. It carries `status` 0 —
and so does `MsgMeshConnectError` (the platform could not be reached), so do not read `status == 0` as
"retry": tell them apart by class or by `code` (`"token_unavailable"` / `"connect"`).

`OperationInProgressError` comes from `dlq_replay`, `delete_webhook(id, dlq="purge")` and `delete_topic`:
replaying a topic's dead letters, purging them and deleting the topic are mutually exclusive, and another one of
them is running on that topic (`err.operation` says which). Nothing was done — wait `err.retry_after` seconds and
make the same call again; do not switch to a different operation to get around it.

## 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.

```python
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:

```python
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 `dataclass`es
(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` (new webhooks only receive messages published after
  creation unless you pass `from_="start"`) / `update_webhook` (change the URL in place — no resend;
  use it instead of delete + create) / `delete_webhook` (deleting a webhook that leaves no other active
  webhook on the topic while its DLQ holds messages raises `DLQNotEmptyError`; retry with `dlq="keep"` or
  `dlq="purge"` — purge raises `DLQPurgeBlockedError` while the topic still has suspended webhooks, see
  `err.suspended_siblings`) / `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 `get_dlq_depth` (read-only count) / `dlq_peek` /
  `dlq_replay` (fix or recreate the webhook before replaying)

> **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

```sh
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) | **繁體中文**

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

**從 0.6.x 升級?** 程式不必改,但有些行為不同了——見[從 0.6.x 升級](#從-06x-升級)。

## 安裝

```sh
pip install msgmesh
```

需要 Python ≥ 3.9;唯一執行期依賴為 [`httpx`](https://www.python-httpx.org/)(0.27 以上、1.0 以下——還不支援 httpx 1.0)。

## 快速開始

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

```python
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,離開時關閉底層連線池:

```python
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` 二擇一,至少需其一。

```python
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(可自助充值解封)、429 與其他
  暫時性錯誤回報後退避續試(自癒;見下方「重連退避」)。`get_token` 模式的 401 多半只是 token 過期,失效快取後以新 token
  續試,連續 5 次仍被拒才判定永久撤權。另一個會讓它停止的情況是 `get_token` 丟出 `TokenUnavailableError`
  (見下方「永久拿不到 token」)。永久停止時若未提供 `on_error`,會記一則 `logging` warning。

  ```python
  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` 為原因字串)。

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

  > **與 sdk-js 的差異**:sdk-js 的 `stream` 依賴瀏覽器原生 `EventSource`(有原生自動重連)。
  > Python 無 `EventSource`,故本 SDK **自管重連**(對齊 sdk-js `streamWs` 的精神):每次串流結束/
  > 連線錯誤退避後重連,`get_token` 模式換新 token,連線撐過 5 秒或收到訊息後退避歸零。**不會因為試了幾次就放棄**
  > (見下方「重連退避」);`"authorization revoked"` 為終態,主動停止。

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

  ```sh
  pip install "msgmesh[ws]"
  ```

  ```python
  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 接管(兩種模式皆然)——退避重連、
  > 連線撐過 5 秒或收到訊息後歸零、**不會因為試了幾次就放棄**(見下方「重連退避」),`get_token`
  > 另在重連前換新 token。**撤權**:連線中為 CLOSE `1008` + `"authorization revoked"`(終態,立即停);
  > 握手被拒則與其他失敗一樣退避重試。`on_error` 收到 `Exception`(連線層/握手失敗)
  > 或 `WsClose`(關閉事件,可讀 `.code`/`.reason`);`stop()` 主動結束不觸發 `on_error`,
  > 對端已經沒有回應時也會立刻返回(直接關掉 socket,不等對端回應關閉握手)。
  > 不想加依賴時,伺服器端用 `stream`(SSE)或 `subscribe`(長輪詢)即可覆蓋即時接收。

**重連退避:會一直試下去。** `subscribe`、`stream`、`stream_ws` 的重試都由 SDK 自己管。等待時間從最多 1 秒起,
每多連續失敗一次就加倍,上限 30 秒;實際等待是該級距上半段裡的隨機值,所以同時失敗的一批客戶端不會同時回來。
成功一次就歸零回到 1 秒:`subscribe` 是一次輪詢有回應;`stream` 與 `stream_ws` 是連線撐過 5 秒、或收到一則訊息。
光是「連上」不算——否則遇到一接受連線就馬上關掉的對象,會變成每秒重連一次、沒完沒了。

`stream` 與 `stream_ws` **不會因為試了幾次就放棄**:連線被拒可能只是過一陣子會自己解除的限流(HTTP 429,
例如連線數上限),這時停下來,限制解除後串流也不會回來。它們只在三種情況結束:伺服器明確撤權
(`"authorization revoked"`);你呼叫停止用 callable(它也會立刻中斷等待中的重試,`stream` 正在等下一則事件的讀取
也會被打斷);以及你的 `get_token` 丟出 `TokenUnavailableError`(見下)。
每次失敗都會呼叫 `on_error`,只有一種情況例外:伺服器把 `stream`(SSE)的回應正常結束時——內文單純收完、
沒有錯誤——SDK 會退避後重連,但不呼叫 `on_error`(`stream_ws` 沒有這種安靜的情況:連線每次被關都會回報)。
連線被拒時,`stream` 收到的是有型別的錯誤(`AuthError`、`RateLimitError`…),
`stream_ws` 收到的是 websocket-client 的握手例外,你認定某個狀態碼是終態時可以自己停。
呼叫停止用 callable 之後就不再回報任何東西——當下還在進行中的取 token、連線或握手,事後才失敗也一樣。SDK 不會替你停,
401 也不會:在即時連線上,租戶停權中、或平台當下無法查核這把 key 時,回的同樣是 401,而這兩種情況都會過去。
沒給 `on_error` 時,連續第 5 次失敗會記一則 warning(logger `msgmesh`)說明它還在重試——每一輪連續失敗各記一則,
所以之後若又連續失敗 5 次會再記。

**`get_token` 模式下,工作階段結束時請自己把串流結束掉。** 每次重試都會向你的 `get_token` 要一張新 token
——一直沒恢復的話,最後穩定在每 15–30 秒呼叫一次——而串流不會因為失敗了幾次就自己停。確定憑證已經失效時,
請用下面兩種做法之一:

- **呼叫停止用 callable**(該方法的回傳值)。之後不再回報任何東西。
- **讓 `get_token` 丟出 `TokenUnavailableError`。** SDK 會自己把那個訂閱停掉,並通知它的 `on_error` 一次。

**永久拿不到 token:`TokenUnavailableError`。** 你的 `get_token` 用它表示「之後也換不到 token 了」:

```python
import httpx
from msgmesh import MsgMesh, TokenUnavailableError

def get_token():
    r = httpx.post("https://my-backend.example/api/mm-token", headers=session_headers())
    if r.status_code == 401:  # 你的後端用這個回應表示工作階段已經結束
        raise TokenUnavailableError("signed out")
    r.raise_for_status()      # 其他狀況都是暫時性的:SDK 會重試
    return r.json()

def on_error(err):
    if isinstance(err, TokenUnavailableError):
        ask_the_user_to_sign_in()  # 這條串流已經停止,不會再重試

mq = MsgMesh(get_token=get_token, realtime_url="https://rt.example.com")
mq.stream("room.42", print, on_error)
```

- **什麼時候該丟:** 再問一次也沒有用的時候——你的後端回覆使用者已登出、refresh token 被拒、帳號已關閉。
- **什麼時候不該丟:** 任何可能過去的狀況——連不上你的後端的網路錯誤、後端回的 5xx 或 429。這時丟其他任何例外
  (或讓 HTTP 用戶端自己的錯誤直接往外丟):SDK 會當成暫時性的,照上面的退避重試,與加入這個例外之前完全相同。
- **SDK 會做什麼:** `subscribe`、`stream`、`stream_ws` 會把那個訂閱停掉——不再重試、不再呼叫 `get_token`、
  背景執行緒結束——並把這個例外交給 `on_error` 一次;沒給 `on_error` 時記一則 warning(logger `msgmesh`)。
  已經呼叫過停止用 callable 的訂閱不會回報。一次性的呼叫(`publish`、`history`…)則是原樣丟出這個例外。
- **不廣播,也不記住。** 同一個用戶端上的其他訂閱會繼續運作,直到它們自己下一次需要 token——各自重連、
  或各自下一次輪詢——那時 `get_token` 再丟一次,它們才停。之後每一次需要 token,SDK 都會再呼叫 `get_token`。
- 子類別也算。它是 `MsgMeshError`(所以 `except MsgMeshError` 仍然接得到 SDK 會丟的所有例外),`status` 為 0、
  `code` 為 `"token_unavailable"`;`str(err)` 就是你給的訊息原文。要保留原因請用
  `raise TokenUnavailableError("…") from err`。(sdk-js 的對應類別繼承的是 `Error`,而且另外認 `name`,
  因為打包工具可能把那個 SDK 打包成兩份;Python 沒有這種情況。)

`subscribe` 保留兩個終態(API key 的 401;`get_token` 模式連續 5 次 401),其餘錯誤都以同一套退避重試。

429 的回應有說要等多久時(先看內文的 `retry_after`,再看 `Retry-After` 標頭),若比退避值長,三者都會等那麼久
(上限 300 秒)。不是每個 429 都會帶;沒帶就照退避走。`stream_ws` 握手被拒時,回應要有 `Content-Length`
才讀得到內文,否則只看標頭。HTTP 回應的這個值也掛在錯誤上:`RateLimitError.retry_after`
(回應沒給時為 `None`;最多 86400)。這是兩個 SDK 行為不同的一處:sdk-js 的 `stream` / `streamWs` 架在
`EventSource` / `WebSocket` 上,連線被拒時拿不到狀態碼與標頭,所以它只有 `subscribe` 能照 `Retry-After` 等。

重連邏輯的已知限制:

- **對方每條連線都先送一則新訊息、再把連線關掉**時,每一輪都會歸零,所以這種迴圈大約每秒重連一次;
  `get_token` 模式下你的 `get_token` 也會被呼叫得一樣頻繁。SDK 已經看過的訊息(被去重的重送)不會讓它歸零。
- **`stream`:伺服器收下連線卻一直不回應標頭。** 停止用 callable 會立刻返回、之後也不會回報任何東西,
  但背景執行緒會卡到伺服器回應或斷線為止(等回應這一段沒有逾時,安靜的串流才不會被切斷)。
  它是 daemon 執行緒,不會撐住行程。

**`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`。

```python
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` 皆同)只收該房間:

```python
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):

```python
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,不會寫進第二份,也不會重複計量。

```python
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;去重服務短暫不可用時會選擇照常發布,而不是讓你的寫入失敗。消費端仍要能容忍重複。

## 用戶端標記(`User-Agent`)

SDK 會替自己送出的東西加上標記,讓平台分得出它與其他用戶端的流量:

```
User-Agent: msgmesh-sdk-py/<SDK 版號>
```

傳入 `app_info`,就會把你的應用程式放在前面:

```python
mq = MsgMesh(api_key=key, app_info={"name": "order-worker", "version": "1.4.2"})
# User-Agent: order-worker/1.4.2 msgmesh-sdk-py/<SDK 版號>     ("version" 可省略)
```

- `name` 與 `version` 必須是 User-Agent 的 product token:只能有英數字與 ``!#$%&'*+-.^_`|~``,
  不能有空白、`/` 或 `@`;長度各自最多 128 個字元。不合法、沒給 `name`、或 `app_info` 不是 mapping 時,
  建構子會拋 `ValueError`。
- 這個 SDK 自報的名稱是 `msgmesh-sdk-py`;JavaScript SDK 是 `msgmesh-sdk-js`。要依它們篩選時,請當成固定字串。
- SDK 發出的每個請求都帶:每個 HTTP 呼叫、`stream` 開的 SSE 連線、`stream_ws` 的 WebSocket 握手。
  這些都是 0.7.0 新增的:0.6.x 以前,HTTP 與 SSE 請求帶的是 httpx 預設的 `User-Agent`(`python-httpx/<版號>`),
  WebSocket 握手不帶 `User-Agent`。
- **`msgmesh-` 開頭的名稱是保留的**,只給 MsgMesh 自己的用戶端用(`msgmesh-panel`、`msgmesh-mcp`、`msgmesh-cli`,
  以及日後新增的)——請不要拿這個前綴當你的應用程式名稱。建構子不會擋,但名稱剛好是其中之一時,
  平台會把你的流量算成那個用戶端的(平台比對這些名稱時不分大小寫)。
- 標記是用戶端自己報的、只供參考,不參與鑑權,也不影響路由。
- 需要記錄 SDK 版號時用 `msgmesh.__version__`。

(sdk-js 的同一種標記 `msgmesh-sdk-js/<版號>` 走三條路:瀏覽器以外用 `User-Agent`;瀏覽器裡的 `User-Agent`
是瀏覽器自己的,所以改用 `X-MsgMesh-Client` 標頭;EventSource / WebSocket 連線在瀏覽器裡帶不了標頭,
所以放在網址的 `client` 參數——那裡不含呼叫端的 `appInfo`,因為網址會進存取紀錄。本 SDK 只跑在伺服器端,
`User-Agent` 一律可用,因此全程只用這一個標頭,另外兩種都不送,你的 `app_info` 也不會出現在任何網址上。)

## 從 0.6.x 升級

程式不必改,但下列行為不同了:

- **`stream` / `stream_ws` 不再因為連續失敗 5 次而停止。** 以前會放棄;現在會照退避一直重試(見上面的「重連退避」),
  直到伺服器撤權、你呼叫停止用 callable,或 `get_token` 丟出 `TokenUnavailableError`。
  `get_token` 模式下每次重試都會呼叫你的 `get_token`。
- **重試的間隔變長了。** 失敗後的等待以前固定 1 秒;現在每多連續失敗一次就加倍,上限 30 秒
  ——三個方法(`subscribe`、`stream`、`stream_ws`)遇到有說要等多久的 429 時最長等 300 秒。
  服務恢復之後,訊息可能晚這麼久才回來。
- **`User-Agent` 變了。** 每個 HTTP 與 SSE 請求從 httpx 的預設值(`python-httpx/<版號>`)變成
  `msgmesh-sdk-py/<SDK 版號>`;`stream_ws` 的 WebSocket 握手以前不帶,現在也帶。見上面的「用戶端標記」。
- **停止用 callable 立刻生效。** 它會打斷 `stream` 正在等下一則事件的讀取;對 `stream_ws`,
  不再等對端回應關閉握手。呼叫之後不再回報任何東西。
- **`httpx` 的版本範圍加了上限**(`httpx>=0.27,<1`)。httpx 1.0 改了 client 的建立方式,這個 SDK 還不能在上面跑;
  0.6.x 以前沒有設上限,全新安裝可能裝到 1.0、一建立 client 就出錯。你的環境若另外需要 httpx 1.x,兩者不能並存:
  同一次安裝同時要求兩者時,pip 會回報依賴衝突而拒絕;裝進已經有 httpx 1.x 的環境時,pip 會把它換成 0.x 的版本。

0.7.0 新增:`TokenUnavailableError`、`app_info` 選項、`RateLimitError.retry_after`。

## 錯誤處理

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

```python
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` |
| 409(`code: operation_in_progress`) | `OperationInProgressError` | `operation_in_progress` |
| 其他 | `MsgMeshError` | `server` |
| 連線層失敗 | `MsgMeshConnectError` | `connect` |
| 你自己的 `get_token` 丟的 | `TokenUnavailableError` | `token_unavailable` |

`TokenUnavailableError` 不會來自平台:它是由你自己的 `get_token` 丟出來、表示「之後也換不到 token 了」的例外
——見「即時接收」裡的「永久拿不到 token」。它的 `status` 是 0;連不上平台時的 `MsgMeshConnectError` 也是 0,
所以不要看到 `status == 0` 就重試——用類別或 `code`(`"token_unavailable"` / `"connect"`)分辨。

`OperationInProgressError` 來自 `dlq_replay`、`delete_webhook(id, dlq="purge")` 與 `delete_topic`:同一個 topic 的
重放死信、清除死信、刪 topic 三者互斥,另一個正在進行(`err.operation` 說明是哪一種)。什麼都沒做——等
`err.retry_after` 秒後再呼叫同一個方法即可;不要改用其他操作繞過。

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

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

```python
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` 時就是沒有漏。

往前翻頁:

```python
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` 則,或已經掃到底。
它**不是**「這一頁裝滿了 `limit` 則」,也**不是**「沒有更舊的了」。一個總共只有 3 則的房間查 `limit=50`,
回的是 `complete=True` 加 3 則;一個很忙的房間回了滿滿 50 則、後面還有幾千則,同樣是 `complete=True`。

**要知道還有沒有下一頁,看 `before`,不要看 `complete`。** `before` 非空代表還拿得到更舊的,
`""` 代表已經到底了。(`limit` 在伺服器端有上限、超過時不會報錯,所以 `len(messages) < limit`
同樣不能拿來判斷歷史是否到底。)

`complete=False` 代表掃描是被你的請求以外的原因打斷的,`incomplete_reason` 說明是哪一種:
`budget`(更舊的還在 Kafka 裡,只是這次沒掃到,帶 `before` 再翻一頁;涵蓋則數掃描、單頁位元組
上限、broker 一時卡住三種)、`retention`(你帶的 `since` 範圍比平台還留著的更舊,那一段已經被刪掉、
拿不回來;只有指定了某段較舊的範圍才會遇到,單純「最近 N 則」的查詢一律是 `complete=True`)。
差別在可否補救:`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`(新 webhook 預設只收建立之後的訊息,要補送歷史帶 `from_="start"`)/ `update_webhook`(就地改網址,不重送;改網址用它,不要刪除重建)/ `delete_webhook`(刪掉之後 topic 上沒有其他 active webhook、而 DLQ 還有死信時拋 `DLQNotEmptyError`,帶 `dlq="keep"` 或 `dlq="purge"` 重送;topic 上還有停權中的 webhook 時(`err.suspended_siblings` > 0)purge 會拋 `DLQPurgeBlockedError`,只能 keep)/ `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 `get_dlq_depth`(唯讀筆數)/ `dlq_peek` / `dlq_replay`(先修好或重建 webhook 再重放)

> **選用能力**:`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。

## 開發

```sh
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*` 撞車。
