Building

Live connections

Socket-only platforms without running a server — AgentParley holds the connection and wakes your plugin only when something matters.

View as Markdown

Some platforms only deliver messages over a connection the receiver keeps open for days: Discord's Gateway, IRC, Matrix sync, Slack Socket Mode, a Home Assistant event stream. A plugin runs for seconds at a time and can't hold a socket. So AgentParley holds it for you.

You describe how to connect and what to do with what arrives. A dedicated service opens the socket, keeps it alive, sends your heartbeats, answers keep-alives, drops the noise you told it to drop, reconnects with backoff, survives AgentParley's own deploys, and wakes your plugin only for frames you care about.

The official Discord plugin is built on this — see the walkthrough.

The handlers

globalThis.connection = {
  open:     (request) => ConnectionOpenResult,     // required
  receive:  (request) => ConnectionStepResult,     // required
  timer?:   (request) => ConnectionStepResult,     // only if you use wakeAfterMs
};

Every call gets your stored state and returns what to do next. Nothing else survives between calls.

open — connect (or reconnect)

request: {
  owner: string;                                   // "channel" or "install" — see "Two kinds of owner"
  reason: "start" | "reconnect" | "handoff";       // handoff = the platform moved you (deploy, rebalance)
  state: unknown | null;                           // your stored state; null on the very first open
  lastClose: { code: number | null; reason: string; by: "remote" | "plugin" | "platform"; link: string | null } | null;
  lastSequence: unknown | null;                    // the last sequence number seen (see `sequence`)
}
returns: ConnectionStepResult & {
  links: LinkSpec[];              // at most 5 sockets
  resumable?: boolean;            // your next open may resume the vendor session after a platform move
  connectsPerMinute?: number;     // 1–600; may only LOWER the platform's connect rate
}

Return { links: [], fail: "…" } to stop for good (for example, a rejected token) — nothing reconnects until the user acts.

Links

type LinkSpec =
  | { transport: "websocket"; url: string; headers?: Record<string, string>; protocols?: string[] }
  | { transport: "tcp"; host: string; port: number; tls: boolean }         // line-based; never put CR/LF inside a frame
  | { transport: "sse"; url: string; headers?: Record<string, string> }
  | { transport: "poll"; everyMs: number; request: HttpRequest };    // everyMs: 30 000 – 86 400 000

type HttpRequest = {
  method?: "GET" | "POST" | "PUT"; url: string; headers?: Record<string, string>; body?: string; timeoutMs?: number;
  oauth?: string;     // a manifest `oauth` key: the service adds `Authorization: Bearer <fresh token>`; you never see it
  account?: string;   // the connection id from host.oauth.accounts(), when more than one is connected
};

// Every link also takes:
{
  key?: string;                  // name it if you have several; unnamed links are "link0", "link1", … in order
  idleTimeoutMs?: number;        // reconnect if nothing arrives for this long
  maxDetachedSeconds?: number;   // 30–300 (default 300): keep the socket while cut off from AgentParley
  filters?: FrameRule[];         // keep/drop rules, applied before any wake (on a poll link: to 2xx replies)
  autoReply?: { textPrefix: string; replacePrefixWith: string }[];   // e.g. IRC PING → PONG, costs no wake
  sequence?: { path: string };   // JSON path of the vendor's sequence number, tracked on every frame
}

Poll links: quiet polling

The service repeats request every everyMs (the first poll comes after a random delay of up to everyMs) and does not wake you for a poll that finds nothing. A poll costs one HTTPS request from the service and nothing else — no call to AgentParley, no wake, no allowance. Requests time out at 120 s (set a lower timeoutMs if you like), redirects are not followed, a reply over 2 MB counts as a network failure, and the request host must be on your manifest.egress list.

You are woken (with an InboundFrame of kind response) for:

• a 2xx reply that your filters keep (rules match the first 256 KB of the body; a reply matching no rule is kept), and
• any 3xx or 4xx except 401, 408 and 429.

After a wake the link pauses until receive returns; polling then resumes on its schedule. Return next to change what it asks for (a new cursor, a new pageToken).

You are not woken for failures the service retries for you: 408, 429 and 5xx replies, and network errors. Retries back off from everyMs upward (doubling, at most 30 minutes between tries) and honour Retry-After (up to one hour). After 5 quiet failures in a row (408/429/5xx) the service posts one note to the connection card so the user can see the outage; the next good reply clears the streak. Five network failures in a row (no HTTP reply at all) close the connection and it reconnects through the normal back-off; any HTTP reply, even a 5xx, resets that counter.

With oauth on the request, a 401 is not delivered to you: the service gets a fresh token and sends the request once more. A second 401 stops the connection with "sign-in required" until the user reconnects the sign-in. A 401 on a request without oauth is an ordinary 4xx and wakes you.

Example: a Gmail watcher that wakes only when the mailbox history moves.

{
  transport: "poll",
  everyMs: 120_000,
  request: { url: "https://gmail.googleapis.com/gmail/v1/users/me/history?startHistoryId=" + cursor
                  + "&historyTypes=messageAdded", oauth: "google", account },
  filters: [
    { when: [{ path: "history", exists: true }], action: "keep" },   // something changed
    { when: [{ textPrefix: "" }], action: "drop" },                   // nothing new: stay asleep
  ],
}

receive — frames arrived

request: {
  state: unknown | null;
  link: string;              // which link
  frames: InboundFrame[];    // up to 50 per call
  lastSequence: unknown | null;
}

type InboundFrame =
  | { id: string; kind: "text"; text: string }
  | { id: string; kind: "binary"; base64: string }
  | { id: string; kind: "sse"; event: string; data: string; lastEventId: string | null }
  | { id: string; kind: "response"; status: number; headers: Record<string, string>; body: string };

A frame's id is the platform's de-duplication key across redelivery (not the vendor's id).

timer — nothing arrived for wakeAfterMs

request: { state: unknown | null; lastSequence: unknown | null }

What you return — ConnectionStepResult

Every field is optional; omit what you don't change.

Filters

The service applies your rules to every frame before deciding whether to wake you. A frame you drop costs nothing.

type FrameRule = { when: FrameCondition[]; action: "keep" | "drop" };

type FrameCondition =
  | { path: string; equals: string | number | boolean | null }   // dot path into JSON; "a[].b" = any element
  | { path: string; exists: boolean }
  | { textPrefix: string }
  | { binaryPrefixBase64: string };

The first rule whose conditions all match decides; a frame matching no rule is kept. Discord's rules keep direct messages, @mentions and replies to the bot, and drop everything else in a server:

// plugins/discord/index.ts
function baseFilters(botId: string | undefined, respondToBots: boolean): FrameRule[] {
  const rules: FrameRule[] = [
    { when: [{ path: "op", equals: 11 }], action: "drop" },            // heartbeat ACKs
    { when: [{ path: "t", equals: "RESUMED" }], action: "drop" },
  ];
  if (!botId) {
    return rules;
  }
  if (!respondToBots) {
    rules.push({ when: [{ path: "d.author.bot", equals: true }], action: "drop" });
  }
  rules.push(
    { when: [{ path: "t", equals: "MESSAGE_CREATE" }, { path: "d.guild_id", exists: false }], action: "keep" },
    { when: [{ path: "t", equals: "MESSAGE_CREATE" }, { path: "d.mentions[].id", equals: botId }], action: "keep" },
    { when: [{ path: "t", equals: "MESSAGE_CREATE" }, { path: "d.referenced_message.author.id", equals: botId }], action: "keep" },
    { when: [{ path: "t", equals: "READY" }], action: "keep" },
    { when: [{ path: "op", equals: 0 }], action: "drop" },
  );
  return rules;
}

Heartbeats, sequences and resume

A typical gateway: on HELLO, identify (or resume) and start a heartbeat; track the sequence number; on READY, store the session id so the next open can resume.

// plugins/discord/index.ts (abridged)
if (payload.op === 10) {                                  // HELLO
  const identify = canResume
    ? { op: 6, d: { token, session_id: state.sessionId, seq: sequence } }
    : { op: 2, d: { token, intents: /* … */, properties: { os: "linux", browser: "agentparley", device: "agentparley" } } };
  send.push({ link: request.link, text: JSON.stringify(identify) });
  heartbeat = {
    link: request.link,
    everyMs: payload.d.heartbeat_interval,
    jitterFirst: true,
    frame: { json: { op: 1, d: null }, sequenceAt: "d" },  // the service writes the latest sequence into d
  };
}

With sequence: { path: "s" } on the link, the service tracks s on every frame — even dropped ones — and hands it back as lastSequence. Declare resumable: true and, after a deploy moves your socket (reason: "handoff"), you resume the vendor session instead of logging in again.

Two kinds of owner

An install-owned connection requires manifest.installOwnedConnection: true. The same code can serve both — hello-live branches on request.owner:

// infrastructure/plugins/hello-live/index.ts
if (state.owner === "install") {
  events.push({ id: dispatch.id, text: dispatch.text });
} else {
  messages.push({
    conversationId: dispatch.conversationId,
    messageId: dispatch.id,
    sender: { id: dispatch.senderId ?? "tester", displayName: dispatch.senderName ?? "Tester" },
    text: dispatch.text,
    isGroup: !!dispatch.guildId,
    mentionsBot: !!dispatch.mention,
  });
}

Each event you emit fires the account's schedules of kind On event that listen on that connection; the schedule's prompt runs on its agent with your event text.

Sending outside a step — host.connection.send

From a channel's send handler (a reply the agent wrote), write straight onto the open socket:

// infrastructure/plugins/hello-live/index.ts
globalThis.channel = {
  send: async (request) => {
    const frame = { op: 99, d: { conversationId: request.conversationId, text: request.text } };
    await host.connection!.send([{ text: JSON.stringify(frame) }], { idempotencyKey: request.deliveryId ?? undefined });
  },
};

At most 20 frames per call, accepted or refused whole; pass the same idempotencyKey on a retry so nothing is written twice. Permission host:connection:send is derived from the call. (Discord sends replies over REST instead, which is why it doesn't need this.)

Where it may connect

Links may only reach hosts in your manifest.egress or a hosts setting the user filled in. The service also refuses private, loopback, link-local and cloud-metadata addresses, checked on the address it actually dials.

Limits