The Socket Server

AgentSocketServer is the server half of the agent socket protocol — it drives every WebSocket an agent holds. Construct one per actor instance and delegate the actor’s webSocket(ws, req) to attach(). The server sends each admitted connection a state snapshot followed by hello, dispatches inbound RPC frames to the actor’s @rpc()-decorated methods, and pushes incremental updates through the broadcast* methods as the agent’s data changes.

import { Agent, type Env } from "@telnyx/edge-runtime";
import type { MergePatch } from "@telnyx/edge-runtime/internal";
import { AgentSocketServer, type AgentServerSocket } from "@telnyx/edge-runtime/agent-socket";

interface ConvState extends Record<string, unknown> {
  status: string;
}

export class Conversation extends Agent<Env, ConvState> {
  private desk = new AgentSocketServer<ConvState>(this, {
    getState: () => this.getState(),
    getMessages: () => this.messages.all(),
    // token → claims; throw to reject; `undefined` = anonymous client
    authorize: (token) => (token === undefined ? ["read"] : ["read", "rpc"]),
  });

  async webSocket(ws: AgentServerSocket, req: Request): Promise<void> {
    await this.desk.attach(ws, req);
  }

  protected override async setState(patch: MergePatch<ConvState>): Promise<ConvState> {
    const next = await super.setState(patch);
    this.desk.broadcastPatch(patch); // incremental push to every client
    return next;
  }
}

With an authorize hook configured, every connection becomes an authorized session: a client that opens with an attach frame presents its token, authorize(token) maps it to claims, and per-stream subscriptions are negotiated; a client that never attaches is authorized as anonymous — authorize(undefined) — so admit it with limited claims, or throw to reject. RPC call frames require the "rpc" claim, which is what lets read-only watchers and fully privileged operators share one socket endpoint.

Attaching a Socket

attach() registers its listeners on the actor’s socket before its first await.

// agent-client/server.ts
ws.on("close", gone);
ws.on("error", gone);
ws.on("message", (data, isBinary) => {
  if (admitted) {
    return ordered ? enqueue(data, isBinary) : this.handleMessage(ws, data, isBinary, req);
  }
  // …
  if (buffered.length >= MAX_PENDING_INBOUND) {
    this.sendError(ws, "protocol_error", "too many frames before connect");
    this.reject(ws, 1013, "inbound queue overflow");
    return;
  }
  buffered.push({ data, isBinary });
  // …
});

webSocket(ws, req) hands attach() the accepted socket and the handshake request. The listeners go on before the entry returns, as the entry requires.
ws.on("message") delivers each frame as one turn of the instance, serialized with its calls and its alarm. Each turn runs under the 30-second method budget.
ws.close(1013) ends a connection that sends more than 64 frames, the value of MAX_PENDING_INBOUND, before it is admitted. The platform uses the same code when its own event queue overflows.

Broadcasts

Every broadcast* method ends in the same loop over the sockets the server admitted.

// agent-client/server.ts
broadcastTo(frame, deliver) {
  const text = encodeFrame(frame);
  for (const ws of [...this.sockets]) {
    if (!deliver(this.sessions.get(ws)))
      continue;
    if (ws.readyState !== ws.OPEN)
      continue;
    try {
      ws.send(text);
    }
    catch {
      this.sockets.delete(ws);
      this.sessions.delete(ws);
    }
  }
}

ws.send goes to each admitted socket subscribed to the frame’s stream. The send is immediate, and it is not held until the turn’s writes commit.
ctx.broadcast() and ctx.count() are not used. Both cover every socket on the instance. The server keeps its own set instead, as Sending to Some Sockets, Not All describes.

AgentSocketServerOptions

Options for AgentSocketServer — the callbacks that connect the server to one agent’s state, message log, and event log, plus the connection-authorization policy. Type Parameters

Type Parameter	Description
`TState`	The agent’s persistent state shape, as returned by `getState` and pushed to clients.

Properties attachGraceMs?

optional attachGraceMs?: number

How long (ms) an attach-capable bootstrap waits for the client’s first frame before authorizing the connection as ANONYMOUS and running the v1 bootstrap — clients that never attach send nothing until hello, so without this fallback they would wait forever. Default 300. The wait is a timer armed when attach returns, not a pause inside it: on the Telnyx actor host, which delivers frames only after webSocket() has returned, an on-time attach frame is negotiated in its own turn and the timer never fires (see attach). The fallback is never a permanent downgrade: an attach arriving AFTER the window elapsed (a slow link losing the race) still upgrades the session with the token’s grants. But if authorize REJECTS anonymous connections, an expired window closes the socket before a late attach can arrive — clients of such a server must attach within this window, so give slow links a generous grace, or grant anonymous a minimal (e.g. read-only) claim set instead of rejecting. Only used when authorize is configured.

authorize?

optional authorize?: (token, req?) => readonly Claim[] | Promise<readonly Claim[]>

Resolves a credential into this connection’s claims (grants). Throw to reject: the socket gets error { code: "unauthorized" } and is closed before any snapshot. Claims are application-defined strings echoed back verbatim in attached.grants; the server itself only interprets "rpc" (required for call frames once this hook is configured). The second argument is the connection’s upgrade request (when the caller of attach provided one) — front-door identity headers ride here, so a policy can authorize from headers as well as from the token. Providing this hook makes EVERY connection an authorized session:

A connection opening with a v2 attach is authorized with the frame’s token and negotiates grants + per-stream subscriptions.
Any other connection (a v1 first frame, or silence until the attachGraceMs window expires) is authorized as ANONYMOUS — this hook is called with undefined — before the v1 bootstrap (snapshot(s) + hello). Grant claims to admit anonymous (e.g. read-only) clients; throw to reject them.
An attach arriving later upgrades an anonymous session in place, replacing its grants and subscriptions with the token’s.

When this hook is absent, the server behaves exactly as before (no sessions, no gating). Parameters

Parameter	Type
`token`	`string` \| `undefined`
`req?`	`Request`

Returns readonly Claim[] | Promise<readonly Claim[]>

getEvents?

optional getEvents?: (afterSeq) => StoredEvent[] | Promise<StoredEvent[]>

Optional: returns the actor’s persistent events after an exclusive cursor (afterSeq), in seq order. Wire it to the actor’s event log (e.g. (after) => this.events.read(after)). When provided, attached sessions can subscribe to the events stream: the bootstrap replays events past the client’s cursor (or all retained events without one), and broadcastEvent pushes live events to those subscribers. Parameters

Parameter	Type
`afterSeq`	`number`

Returns StoredEvent[] | Promise<StoredEvent[]>

getMessages?

optional getMessages?: () => unknown[] | Promise<unknown[]>

Optional: returns the actor’s conversation log (this.messages.all()). When provided, a messages snapshot is sent on connect and broadcastMessages pushes appends to every watcher. Returns unknown[] | Promise<unknown[]>

getState

getState: () => TState | Promise<TState>

Returns the actor’s current persistent state (snapshot on connect). Returns TState | Promise<TState>

onAttach?

optional onAttach?: (att) => void | Promise<void>

Called once per authorized session — after authorize resolves the connection’s claims and BEFORE any frame is sent for that negotiation — with the connection’s Attachment: its claims and a close to veto it. Call att.close(code?, reason?) (or throw) to reject: the socket is closed, never admitted as a watcher, and receives no frames. Resolve normally to admit. Also runs when a late attach re-authorizes an already-admitted connection; a veto then closes that connection. Only runs on sessions, so it requires authorize to be configured. Parameters

Parameter	Type
`att`	`Attachment`

Returns void | Promise<void>

onConnect?

optional onConnect?: (ctx) => void | Promise<void>

Called once per new connection BEFORE the socket is admitted as a watcher or sent hello/state. Receives the upgrade request (the front door’s headers — auth, identity — ride here) and a close to reject the connection. Throw, or call ctx.close(), to reject: the socket is closed and never joins the watcher set. Resolve normally to accept. Inbound frames arriving while this gate is pending are buffered and replayed in order once the connection is admitted (a rejection discards them with the socket). Wire it from the actor’s onConnect via a closure: new AgentSocketServer(this, { getState: () => this.getState(), onConnect: (c) => this.onConnect(c) }) Parameters

Parameter	Type
`ctx`	`AgentConnectContext`

Returns void | Promise<void>

AgentConnectContext

Context passed to onConnect for per-connection auth/gating. Properties req?

optional req?: Request

The upgrade request (front-door headers, e.g. x-user, ride here). Methods close()

close(code?, reason?): void

Close this connection now to reject it (it won’t receive hello/state). Parameters

Parameter	Type
`code?`	`number`
`reason?`	`string`

Returns void

Get Started

Inside the Managed Harness

The Agent Harness SDK

The Agent Runtime SDK

The Agent Client SDK

For AI Agents

The Socket Server

Attaching a Socket

Broadcasts

AgentSocketServerOptions

AgentConnectContext

​Attaching a Socket

​Broadcasts

​AgentSocketServerOptions

​AgentConnectContext

Attaching a Socket

Broadcasts

AgentSocketServerOptions

AgentConnectContext