@msgmesh/sdk
Advanced tools
MsgMesh TypeScript SDK — publish / consume / realtime (SSE / WebSocket) / governance client for the multi-tenant event bus, universal across Node and the browser.
English | 繁體中文
The TypeScript SDK for MsgMesh — a publish / consume / realtime (SSE / WebSocket) /
governance client for the multi-tenant event bus, universal across Node and the browser
(built on fetch). It is the single source of truth for the entire frontend ecosystem's HTTP
contract: both the MCP server and the panel build on it.
Also available for Python: msgmesh — the same API surface in snake_case.
Register an account in the panel and issue an API key (shown in plaintext only once), then:
import { MsgMesh } from "@msgmesh/sdk";
const mq = new MsgMesh({
apiKey: process.env.MSGMESH_KEY, // long-lived key, server-side only
controlPlaneUrl: "https://cp.example.com",
gatewayUrl: "https://gw.example.com",
realtimeUrl: "https://rt.example.com",
});
await mq.createTopic("orders");
await mq.publish("orders", { hello: 1 });
const msgs = await mq.poll("orders", { group: "g1" });
getToken, never embed an API keyAnything in the browser leaks — never put a long-lived API key in the frontend. Use a
token-broker instead: your backend (which holds the key) calls POST /v1/tokens to exchange it
for a short-lived dp token, and the frontend only ever holds the token. The SDK caches it,
refetches before expiry, and rotates it on SSE reconnect.
const mq = new MsgMesh({
// no apiKey; provide a function that fetches a short-lived token from your backend
getToken: async () => (await fetch("/api/mm-token")).then((r) => r.json()), // { token, expires_in }
gatewayUrl: "https://gw.example.com",
realtimeUrl: "https://rt.example.com",
});
mq.stream("room.42", (data) => console.log(data)); // SSE; auto-rotates the token on expiry and reconnects
mq.streamWs("room.42", (data) => console.log(data)); // WebSocket; same interface, SDK-managed reconnect
Two options for realtime receive, with the same interface, each returning a stop function:
stream (SSE): backed by the browser-native EventSource (which auto-reconnects), over
/…/sse. Browser only.streamWs (WebSocket): the global WebSocket (browser-native; built into Node ≥ 22),
over /…/ws. WebSocket has no native reconnect, so the SDK takes it over: reconnect with a
1s backoff after each drop, reset the failure counter on a successful connect, and stop once
consecutive failures (never having connected) reach the limit — avoiding infinite
reconnect to a revoked credential or a persistently unavailable endpoint; in getToken mode
it also rotates the token before reconnecting. Revocation mid-connection is CLOSE 1008
authorization revoked (stops immediately); during the handshake (HTTP 401 → CloseEvent 1006)
it is bounded by the failure limit. Good when SSE is blocked by a middlebox, or when you
already have WebSocket infrastructure. Node < 22 has no global WebSocket and will throw — use
subscribe (long-polling) instead.Resume on reconnect (at-least-once, no gaps). Both stream (SSE) and streamWs (WebSocket)
resume across reconnects: each message carries a <partition>-<offset> cursor, the SDK tracks the
last one seen, and on reconnect it asks the server to replay from there — so 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
onResync to be told to re-fetch a snapshot. All of this is transparent: onMessage still
receives the raw value string, no API change. (Resume requires the platform's realtime resume
tier; against an older server the stream degrades gracefully to live-tail.)
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: roomId }) to target a room, and
subscribe with the optional room (the fourth argument opts, same for stream / streamWs)
to receive only that room:
mq.stream("chat", (data) => console.log(data), undefined, { room: "room-42" }); // only room-42
mq.streamWs("chat", (data) => console.log(data), undefined, { room: "room-42" });
await 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 "this room only" token for the frontend (a downscope may only narrow, must
be a subset of the key's capabilities, 403 on overreach):
// Backend token-broker: downscope an all-rooms key to a short-lived "chat / room-42 only"
// token, returned to the frontend as getToken
const r = await fetch(`${controlPlaneUrl}/v1/tokens`, {
method: "POST",
headers: { Authorization: `Bearer ${process.env.MSGMESH_KEY}`, "Content-Type": "application/json" },
body: JSON.stringify({
ttl_seconds: 600,
capabilities: [{ ops: ["subscribe", "publish"], topics: ["chat"], rooms: ["room-42"] }],
}),
});
const { token, expires_in } = await r.json(); // return to the frontend; it connects SSE/WS via getToken
You can also mint a persistent room-scoped key with
createKey("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) +
publishto its rooms; it cannotpoll/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.
rooms) into the frontend /
untrusted clients — that lets anyone change room and see all rooms, so isolation is
meaningless.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).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.
const seen = new Set<string>();
const render = (id: string, value: string) => {
if (seen.has(id)) return; // at-least-once: history and the stream deliberately overlap
seen.add(id);
console.log(value);
};
// 1. Fetch the last 50 messages of this room (oldest → newest).
const page = await mm.history("chat", { room: "room-42", limit: 50 });
for (const m of page.messages) render(m.id, m.value);
// 2. Resume live from where history ended. `resume_from` was replayable when it was issued.
mm.stream("chat", (data) => console.log(data), undefined, {
room: "room-42",
from: page.resume_from || undefined,
});
Two cursors, two meanings — this is the part people get wrong:
| field | what it is for | guarantee |
|---|---|---|
resume_from | hand to stream() / streamWs() as from | it was inside the replayable window when the response was produced |
before | hand back to history() for the previous page | none — never pass it to stream() |
Two details about resume_from that save debugging time later:
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, it does not raise the flag "just to be safe".Paging further back:
let cursor = page.before;
while (cursor) {
const older = await mm.history("chat", { room: "room-42", limit: 50, before: cursor });
older.messages.forEach((m) => 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 messages.length < 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. (Three server-side limits share this reason because the fix is the same: the
bounded record scan, the per-page byte cap, and a broker that stalled part-way.)retention — you passed a since bound that reaches further back than the platform still
keeps; that stretch has been deleted by your plan's retention period and paging will not bring
it back. 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 Date and the SDK gets
it right for you.
What this is, and is not. Under the hood this is a bounded scan over the event log, not a
separate history database. That has 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 — a bounded scan cannot invent
data the log no longer holds.
The SDK stores nothing. Where a cursor belongs — memory, localStorage, your own backend — is
your app's decision, not the SDK's. Keep resume_from wherever fits and pass it back as from.
controlPlaneUrl (governance API), gatewayUrl
(send/receive), realtimeUrl (SSE/WS/presence). When unset, the SDK falls back to local-dev
defaults (http://localhost:8080/8081/8082), which are for local use only; when a production
call can't connect, the error message appends a "set controlPlaneUrl/gatewayUrl/realtimeUrl"
hint.getToken
instead.producer, consuming needs consumer, administration needs
admin (which covers everything); a data-plane key that both publishes and consumes, or needs
fine-grained access, can use the neutral key scope (which requires capabilities). getToken
mode is data-plane only (send/receive); calling governance endpoints returns 401/403.Non-2xx responses throw a typed error by status code (all inherit MsgMeshError and carry
status/code/path):
import { ValidationError, AuthError, NotFoundError, RateLimitError } from "@msgmesh/sdk";
try {
await mq.createTopic("Bad Name!");
} catch (e) {
if (e instanceof ValidationError) console.error("invalid argument:", e.message);
else if (e instanceof RateLimitError) console.error("rate limited, retry later");
else throw e;
}
| Status | Type | code |
|---|---|---|
| 400 / 422 | ValidationError | validation |
| 401 / 403 | AuthError | auth |
| 404 | NotFoundError | not_found |
| 429 | RateLimitError | rate_limit |
| other | MsgMeshError | server |
createTopic / listTopics / deleteTopicpublish / poll / subscribe (polling) / stream (SSE, browser) / streamWs
(WebSocket, browser + Node ≥ 22) / getPresence / history (recent messages + a cursor to
resume the live stream from)listKeys (returns capabilities / name) / createKey (accepts scope +
capabilities) / deleteKeylistWebhooks / createWebhook / deleteWebhook / reactivateWebhookregisterSchema / listSchemas / getLatestSchema / deleteSchemaregisterFunction / getFunction / deleteFunction (JavaScript / WASM)getPlan / setPlan; usage: getUsagegetSettings / setStrictTopics (data-plane topic gate toggle)getBilling / getDepositAddresses / getDeposits /
getLedger / getUsageDebits / getDepositStatusgetSnippet / getDocs / getAudit, DLQ dlqPeek / dlqReplayRegistration and admin (finance / tenant governance) go through panel sessions, not this SDK.
npm test -w @msgmesh/sdk && npm run build -w @msgmesh/sdk
English | 繁體中文
MsgMesh 的 TypeScript SDK(Node / 瀏覽器通用,基於 fetch)。是整個前端生態的唯一 HTTP 真相源——MCP server 與面板都複用它。
也有 Python 版:msgmesh(相同 API 面、snake_case)。
先在面板註冊帳號、簽發一把 API key(明文僅顯示一次),再:
import { MsgMesh } from "@msgmesh/sdk";
const mq = new MsgMesh({
apiKey: process.env.MSGMESH_KEY, // 伺服器端用長期 key
controlPlaneUrl: "https://cp.example.com",
gatewayUrl: "https://gw.example.com",
realtimeUrl: "https://rt.example.com",
});
await mq.createTopic("orders");
await mq.publish("orders", { hello: 1 });
const msgs = await mq.poll("orders", { group: "g1" });
getToken,不要放 API key瀏覽器裡的東西都會外洩,不要把長期 API key 放進前端。改採 token-broker:後端(持 key)呼叫
POST /v1/tokens 換一張短期 dp token,前端只拿 token;SDK 會自動快取、將過期前重取,SSE 重連時亦換新。
const mq = new MsgMesh({
// 不放 apiKey;給一個「去我後端拿短期 token」的函式
getToken: async () => (await fetch("/api/mm-token")).then((r) => r.json()), // { token, expires_in }
gatewayUrl: "https://gw.example.com",
realtimeUrl: "https://rt.example.com",
});
mq.stream("room.42", (data) => console.log(data)); // SSE;token 過期自動換新後重連
mq.streamWs("room.42", (data) => console.log(data)); // WebSocket;同介面,SDK 自管重連
即時接收兩種選擇,介面一致、皆回傳停止函式:
stream(SSE):靠瀏覽器原生 EventSource(有原生自動重連),走 /…/sse。僅瀏覽器。streamWs(WebSocket):全域 WebSocket(瀏覽器原生;Node ≥ 22 內建),走 /…/ws。WebSocket 無原生重連,故由 SDK 接管:每次斷線退避 1s 重連,成功連上即重置失敗計數;連續失敗達上限(未曾連上)即停止(避免對已撤銷的憑證或持續不可用的端點無限重連),getToken 模式另在重連前換新 token。撤權若在連線中發生為 CLOSE 1008 authorization revoked(立即停);若在握手期(HTTP 401→CloseEvent 1006)則由上限收口。適合 SSE 被中間層擋掉、或已有 WS 基礎設施的場景。Node < 22 無全域 WebSocket 會拋錯,改用 subscribe(長輪詢)。一個 topic 內可再切多個房間(底層就是一個分割鍵),脫鉤「房間數」與「topic 數」。分兩層:
① 路由——發佈時用 publish(topic, body, { room: roomId }) 指定房間,訂閱時傳選用 room(第四參數 opts,stream/streamWs 皆同)只收該房間:
mq.stream("chat", (data) => console.log(data), undefined, { room: "room-42" }); // 只收 room-42
mq.streamWs("chat", (data) => console.log(data), undefined, { room: "room-42" });
await mq.publish("chat", { text: "hi" }, { room: "room-42" }); // 發到 room-42
省略 room=收該 topic 全部訊息(向後相容)。路由本身只做過濾、無強制隔離——惡意 client 可改成別人的 room 偷聽同 topic 其他房間。要真隔離看 ②。
② 隔離(平台強制)——把憑證的 capabilities 加上選用 rooms,平台即強制該憑證只能收發指定房間(逾越 403)。rooms 省略/空 = 所有房間(向後相容);非空 = 僅限這些。典型作法是後端持一把全房間金鑰,向 POST /v1/tokens 降權簽出「只准某房間」的短期 token 給前端(降權只准更窄、須為金鑰能力子集,逾越 403):
// 後端 token-broker:用全房間 key 降權鑄「只准 chat / room-42」的短期 token,回給前端當 getToken
const r = await fetch(`${controlPlaneUrl}/v1/tokens`, {
method: "POST",
headers: { Authorization: `Bearer ${process.env.MSGMESH_KEY}`, "Content-Type": "application/json" },
body: JSON.stringify({
ttl_seconds: 600,
capabilities: [{ ops: ["subscribe", "publish"], topics: ["chat"], rooms: ["room-42"] }],
}),
});
const { token, expires_in } = await r.json(); // 回給前端;前端以 getToken 用它連 SSE/WS
也可用 createKey("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 時請改用不限房間的憑證。
rooms 的 token)放進前端——那樣 client 改個 room 就能看到所有房間,隔離形同虛設。presence(在線數)目前是 per-topic 非 per-room(只洩漏聚合數字);短期 token 為 bearer,洩漏 = 該房 ≤TTL 可用(故 TTL 短、勿記進 log)。即時串流是從「現在」開始的。晚一步打開聊天室、工單、事件流的人,看到的是空的。
history() 取回最近的訊息,並交給你一個游標,讓你無縫接上即時串流——不漏、不重複。
const seen = new Set<string>();
const render = (id: string, value: string) => {
if (seen.has(id)) return; // at-least-once:歷史與串流刻意有重疊,依 id 去重
seen.add(id);
console.log(value);
};
// 1. 取這個房間最近 50 則(由舊到新)。
const page = await mm.history("chat", { room: "room-42", limit: 50 });
for (const m of page.messages) render(m.id, m.value);
// 2. 從歷史的結尾接上即時。resume_from 在發出的當下必定可續傳。
mm.stream("chat", (data) => console.log(data), undefined, {
room: "room-42",
from: page.resume_from || undefined,
});
兩個游標語意不同——這是最容易搞錯的地方:
| 欄位 | 用途 | 保證 |
|---|---|---|
resume_from | 交給 stream() / streamWs() 的 from | 在回應產生的當下必定落在可續傳窗內 |
before | 交回 history() 取更舊的一頁 | 沒有——絕對不要拿去餵 stream() |
關於 resume_from,兩個之後會省你除錯時間的細節:
id。掃描是由新到舊的,伺服器已經知道「你最新那一則之上」
那段沒有這個房間的訊息,於是刻意把游標推到那裡——讓你的 app 從「拿到歷史」到「連上串流」
之間的容忍時間最大化。resume_gap: true 代表續傳真的會漏掉一段:你往回翻得比可續傳窗還舊、中間被保留期刪掉,
或這是多 partition 的全 topic(不帶 room)查詢。為 false 時就是沒有漏——
伺服器只回報真的存在的缺口,不會為了保守而亂舉旗。往前翻頁:
let cursor = page.before;
while (cursor) {
const older = await mm.history("chat", { room: "room-42", limit: 50, before: cursor });
older.messages.forEach((m) => 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 再翻一頁即可。
(三種伺服器端上限共用這個 reason,因為處置完全一樣:有界的則數掃描、單頁位元組上限、
以及 broker 一時卡住。)retention — 更舊的訊息已被你方案的保留期刪掉;再問也沒有了。差別在可否補救:budget 翻頁就有,retention 沒了就是沒了。
before / since 的時間是 epoch 毫秒(或 RFC3339)。傳成 epoch 秒會回 400,
而不是默默回答 1970 年的事——直接傳 Date,SDK 會幫你轉對。
它是什麼、不是什麼:底下是對事件日誌的有界回掃,不是一個獨立的歷史資料庫。有兩個
設計上要納入考量的結果:能回溯多遠受方案保留期約束;帶 room 遠比掃整個 topic 便宜
(一個 room 的訊息落在單一 partition)。若你需要稽核等級地存取保留期以外的訊息,請自行留存副本
——有界回掃變不出日誌裡已經沒有的資料。
SDK 不替你保存任何東西:游標該放哪裡(記憶體、localStorage、你自己的後端)是你的 app
的決定,不是 SDK 的。把 resume_from 放在合適的地方,下次當 from 傳回來即可。
controlPlaneUrl(治理 API)、gatewayUrl(收發)、realtimeUrl(SSE/WS/presence)。
未指定時 SDK 退回本機開發預設值(http://localhost:8080/8081/8082),只適合本機;
生產環境連不上時錯誤訊息會附「請指定 controlPlaneUrl/gatewayUrl/realtimeUrl」提示。getToken。producer、收訊需 consumer、管理需 admin(通吃);又推又收/細粒度的資料面鍵可用中性
key scope(強制附 capabilities)。getToken 模式僅供資料面(收發),呼叫治理端點會 401/403。非 2xx 回應會依狀態碼拋型別化錯誤(都繼承 MsgMeshError,帶 status/code/path):
import { ValidationError, AuthError, NotFoundError, RateLimitError } from "@msgmesh/sdk";
try {
await mq.createTopic("Bad Name!");
} catch (e) {
if (e instanceof ValidationError) console.error("參數不合法:", e.message);
else if (e instanceof RateLimitError) console.error("被限流,稍後重試");
else throw e;
}
| 狀態碼 | 型別 | code |
|---|---|---|
| 400 / 422 | ValidationError | validation |
| 401 / 403 | AuthError | auth |
| 404 | NotFoundError | not_found |
| 429 | RateLimitError | rate_limit |
| 其他 | MsgMeshError | server |
createTopic / listTopics / deleteTopicpublish / poll / subscribe(輪詢)/ stream(SSE,瀏覽器)/ streamWs(WebSocket,瀏覽器 + Node ≥ 22)/ getPresencelistKeys(回傳含 capabilities/name)/ createKey(可帶 scope+capabilities)/ deleteKeylistWebhooks / createWebhook / deleteWebhook / reactivateWebhookregisterSchema / listSchemas / getLatestSchema / deleteSchemaregisterFunction / getFunction / deleteFunction(JavaScript / WASM)getPlan / setPlan;用量:getUsagegetSettings / setStrictTopics(資料面 topic 閘門開關)getBilling / getDepositAddresses / getDeposits / getLedger / getUsageDebits / getDepositStatusgetSnippet / getDocs / getAudit、DLQ dlqPeek / dlqReplay註冊、超管(finance/租戶治理)走面板 session,不在本 SDK。
npm test -w @msgmesh/sdk && npm run build -w @msgmesh/sdk
FAQs
MsgMesh TypeScript SDK — publish / consume / realtime (SSE / WebSocket) / governance client for the multi-tenant event bus, universal across Node and the browser.
We found that @msgmesh/sdk demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
A new federal program will let vetted U.S. cybersecurity firms help investigate and disrupt foreign cybercrime groups under government direction.
Research
/Security News
The campaign amassed more than 75,000 installs by targeting Russian-speaking users seeking access to blocked services.
Company News
Open source maintainers are under more pressure than ever. We're raising our open source program from the Team plan to the Business plan, free.