Realtime
Realtime tokens, channels, publish and presence.
Overview
Base path:
/v1/realtimePython namespace:
infra.realtimetext
# Python
from infrai import infra
# TypeScript
import { InfraiClient } from "@infrai/sdk";
const client = new InfraiClient({ apiKey: process.env.INFRAI_API_KEY! });
// → client.realtimeMethods
realtime.token.issue
POST /v1/realtime/token
Issue a client token scoped to channels and capabilities.
Python
python
infra.realtime.token.issue(user_id=None, channels=None, ttl_seconds=None)TypeScript
typescript
client.realtime.token(opts?: RealtimeTokenOptions): Promise<RealtimeToken>Parameters
| Name | Type | Required | Description |
|---|---|---|---|
user_id | string | Optional | Id of the connecting user. |
channels | string[] | Optional | Channels the token may access. |
ttl_seconds | number | Optional | Lifetime in seconds before expiry. |
Returns
RealtimeToken { token, client_id, expires_at, endpoint }Example
一次性前置(每个范例都假定已完成):
bash
pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...python
from infrai import infra
tok = infra.realtime.token.issue(user_id="u_123", channels=["orders"], ttl_seconds=300)
print("token:", (tok.get("token") or "")[:12], "...")realtime.channel.create
POST /v1/realtime/channels
Create a realtime channel.
Python
python
infra.realtime.channel.create(name=..., type=None)TypeScript
typescript
client.realtime.channel.create(opts): Promise<{ channel_id, name, type }>Parameters
| Name | Type | Required | Description |
|---|---|---|---|
name | string | Required | Channel name. |
type | "public" | "private" | "presence" | Optional | Channel visibility type. |
Returns
{ channel_id, name, type }Example
一次性前置(每个范例都假定已完成):
bash
pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...python
from infrai import infra
ch = infra.realtime.channel.create(name="orders", type="presence")
print("channel_id:", ch.get("channel_id"))realtime.publish
POST /v1/realtime/publish
Publish an event to a channel.
Python
python
infra.realtime.publish(channel=..., event=..., data=..., idempotency_key=None)TypeScript
typescript
client.realtime.publish(opts): Promise<{ message_id }>Parameters
| Name | Type | Required | Description |
|---|---|---|---|
channel | string | Required | Channel name. |
event | string | Required | Event name. |
data | unknown | Required | Event payload to publish. |
idempotency_key | string | Optional | Optional dedup key; identical retries return the same result. |
Returns
{ message_id }Example
一次性前置(每个范例都假定已完成):
bash
pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...python
from infrai import infra
published = infra.realtime.publish(
channel="orders",
event="order.created",
data={"order_id": "ord_42", "total_usd": 19.99},
)
print("delivered:", published.get("delivered") or published.get("status"))realtime.presence.get
GET /v1/realtime/channels/{channel}/presence
Get the members currently present on a channel.
Python
python
infra.realtime.presence.get(channel=...)TypeScript
typescript
// see ROUTES["realtime.presence.get"]Parameters
| Name | Type | Required | Description |
|---|---|---|---|
channel | string | Required | Channel name. |
Returns
{ members: Array<{ client_id, user_id? }> }Example
一次性前置(每个范例都假定已完成):
bash
pip install infrai
# one-time auth (no secret needed): anonymous account + trial, writes ~/.infrai/credentials
python -c "from infrai import infra; infra.activate()"
# returning user instead: export INFRAI_API_KEY=ifr_pk_proj_...python
from infrai import infra
p = infra.realtime.presence.get(channel="orders")
print("members:", len(p.get("members", [])))