Receive events

Receive messages, delivery updates and reactions from one ordered event log. Choose signed webhooks, a live stream or polling to fit how your agent runs.

TypeScript (Next.js route handler)

// app/api/flow/route.ts
import { FlowMessaging, contentText } from "@flow-engineer/messaging";

const flow = new FlowMessaging();

// Verifies Flow-Signature with FLOW_MESSAGING_WEBHOOK_SECRET, then calls onEvent.
// For message.received, what onEvent returns is sent as the reply.
export const POST = flow.webhooks.handler({
  onEvent: async (event) => {
    if (event.type !== "message.received") return;
    return `You said: ${contentText(event.data.message.content)}`;
  },
});

Three ways to receive events

Way	Use it when	How
Webhook	Production; serverless	Register an HTTPS URL with POST /v0/webhook_endpoints. Flow POSTs each event, signed.
Live stream	Long-running workers, local scripts	GET /v0/stream (WebSocket) or flow.events.stream() in TypeScript. Resumable with after.
Polling	Catch-up, replay, debugging	GET /v0/events?after=evt_..., oldest first.

All three read the same log, so you can mix them: for example a webhook in production and GET /v0/events to recover after downtime.

Polling: GET /v0/events parameters

Parameter	Meaning
after	An event ID (evt_...): only the events after it. Omit to start at the oldest.
before	An event ID: only the events before it.
limit	1 to 100 per page.
type	Only these event types; repeat it for several (type=message.received&type=message.failed, at most 20).
conversation	Only events in this conversation (conv_...).

The answer is {"data": [...], "has_more": true}, oldest first: page on with after set to the last id while has_more is true. Events are kept for your plan’s retention (30 days today; see Retention).

The event shape

Example message.received payload

{
  "id": "evt_01JB8ZC3K5M7P9R1T3V5X7Z9B1",
  "type": "message.received",
  "created_at": "2026-11-03T10:15:00Z",
  "app": "app_01JB8ZC3K5M7P9R1T3V5X7Z9B1",
  "livemode": true,
  "conversation": {
    "id": "conv_01JB8ZC3K5M7P9R1T3V5X7Z9B1",
    "channel": "telegram",
    "sender": "snd_01JB8ZC3K5M7P9R1T3V5X7Z9B1",
    "contact": "ct_01JB8ZC3K5M7P9R1T3V5X7Z9B1"
  },
  "data": {
    "message": {
      "id": "msg_01JB8ZC3K5M7P9R1T3V5X7Z9B1",
      "direction": "in",
      "status": "received",
      "content": { "kind": "basic", "text": "Do you deliver to 560103?" }
    }
  },
  "timing": { "received_at": "2026-11-03T10:15:00.012Z", "stored_at": "2026-11-03T10:15:00.020Z", "delivered_at": "2026-11-03T10:15:00.041Z" }
}

timing shows Flow’s own overhead: delivered_at minus received_at.

Event types

Event types and when to subscribe

type	When
message.received	The contact sent something; data.message.content.kind says what (basic, voice, button_reply, location, …). A tap on a button or list row arrives as button_reply with button_id and label. data.message.from is who wrote, also in an iMessage group chat.
message.sent, message.delivered, message.read	Your message’s progress, where the channel reports it. On iMessage, data.message.network in message.delivered (and message.failed) says whether it went by iMessage, SMS or RCS.
message.failed	Your message could not be sent; data.message.error says why, with an error type.
reaction.added, reaction.removed	The contact reacted to a message.
typing.started, typing.stopped	The contact is typing, where the channel reports it.
conversation.started	A new contact wrote first, joined the sandbox, or you started a conversation (via). A sandbox join is then also delivered as the person’s first message.received.
conversation.window_closing	WhatsApp only, opt-in: the 24-hour window closes in 1 hour.
sender.status_changed	A sender was throttled, flagged, banned or restored (or, on WhatsApp, its quality changed).
template.status_changed	A WhatsApp template was approved, rejected or paused.
contact.deleted	A contact was deleted with DELETE /v0/contacts/{contact_id}; carries only the contact’s ID. See Retention and deleting contacts.
call.received	iMessage only: a contact tried to FaceTime the line. Nothing answers it, it is not a message, and it does not let a reply-only line write first. Delivered only to endpoints that list it in events.
contact.left	A person on the sandbox joined another app (or your app’s other mode) by sending its join code; data.reason is joined_other_app. Nothing more reaches you from them, and sends to them fail with sandbox_contact_left, until they send your join code again.

Status events are high volume. Each webhook endpoint subscribes to the types it lists, so subscribe only to what you use.

Register a webhook endpoint

curl

curl https://api.flow.engineer/v0/webhook_endpoints \
  -H "Authorization: Bearer $FLOW_MESSAGING_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/api/flow", "events": ["message.received", "message.failed"]}'

The answer includes the endpoint’s signing secret (whsec_...), shown only this once. Store it as FLOW_MESSAGING_WEBHOOK_SECRET. Endpoints and their secrets belong to the mode of the key that created them: a test key’s endpoint receives test events, so going live means registering the endpoint again with the live key and storing its new secret.

The URL must be public HTTPS. To run your handler on localhost with no tunnel, npx @flow-engineer/messaging listen --forward-to http://localhost:3000/api/flow forwards each event from the live stream to it, signed like a real delivery (Local development).

Verify the signature

Every delivery carries these headers:

Flow-Signature: t=1791763200,v1=5257a869e7ecebeda32affa62cdca3fa51cad7e77a0e56ff536d0ce8e108d8bd
Flow-Event-Id: evt_01JB8ZC3K5M7P9R1T3V5X7Z9B1
Flow-Event-Type: message.received
Flow-Version: 2026-11-01

v1 is the lowercase hex HMAC-SHA256, keyed with the endpoint secret, of the string t.{t}.{body}: the letter t, a dot, the timestamp from the header, a dot, and the raw request body exactly as received. Compare in constant time and reject timestamps more than 5 minutes from your clock. While a secret is being rotated the header carries one v1 per active secret; accept the request if any matches.

Rotate the signing secret

POST /v0/webhook_endpoints/{webhook_endpoint_id}/rotate_secret makes a new secret and returns it in secret, shown only this once. The old secret keeps signing for overlap_seconds (default 86400, one day; 0 retires it at once), so every delivery during the overlap carries two v1 values and verifies with either secret while you deploy the new one. previous_secret_expires_at says when the old one stops. Rotating again during an overlap retires the older secret at once: at most two are ever active. So send an Idempotency-Key and retry with the same key; a retry without one is a second rotation, which retires the secret you still have deployed. A repeat of a rotation that went through answers 409 idempotency_conflict (the new secret is shown only once).

const { secret } = await flow.webhookEndpoints.rotateSecret("we_...", { overlap_seconds: 3600 });
// Store secret as FLOW_MESSAGING_WEBHOOK_SECRET and deploy within the hour.

TypeScript (Express)

import express from "express";
import { FlowMessaging, contentText } from "@flow-engineer/messaging";

const flow = new FlowMessaging();
const app = express();

// The raw body is required: a parsed and re-serialized body will not match.
app.post("/flow", express.raw({ type: "application/json" }), async (req, res) => {
  let event;
  try {
    event = await flow.webhooks.constructEvent(req.body, req.header("Flow-Signature"), process.env.FLOW_MESSAGING_WEBHOOK_SECRET!);
  } catch {
    return res.status(400).end();
  }
  if (event.type === "message.received") {
    return res.json(flow.webhooks.reply(`You said: ${contentText(event.data.message.content)}`));
  }
  res.json({});
});

app.listen(3000);

Python (FastAPI)

# pip install flow-messaging fastapi
from fastapi import FastAPI, Request, Response
from flow_messaging import WebhookSignatureError, construct_event, content_text, webhook_reply

app = FastAPI()

@app.post("/flow")
async def flow_webhook(request: Request):
    body = await request.body()  # the raw body: a parsed and re-serialized body will not match
    try:
        # The secret defaults to FLOW_MESSAGING_WEBHOOK_SECRET.
        event = construct_event(body, request.headers.get("Flow-Signature"))
    except WebhookSignatureError:
        return Response(status_code=400)
    if event.type == "message.received":
        return webhook_reply(f"You said: {content_text(event.data.message.content)}")
    return {}

Go (net/http)

// The Go SDK is not published yet, so this checks the signature by hand.
package main

import (
  "crypto/hmac"
  "crypto/sha256"
  "encoding/hex"
  "encoding/json"
  "io"
  "net/http"
  "os"
  "strconv"
  "strings"
  "time"
)

func verify(body []byte, header, secret string) bool {
  var ts string
  var sigs []string
  for _, part := range strings.Split(header, ",") {
    k, v, _ := strings.Cut(strings.TrimSpace(part), "=")
    switch k {
    case "t":
      ts = v
    case "v1":
      sigs = append(sigs, v)
    }
  }
  t, err := strconv.ParseInt(ts, 10, 64)
  if err != nil {
    return false
  }
  if d := time.Now().Unix() - t; d > 300 || d < -300 {
    return false
  }
  mac := hmac.New(sha256.New, []byte(secret))
  mac.Write([]byte("t." + ts + "."))
  mac.Write(body)
  want := hex.EncodeToString(mac.Sum(nil))
  for _, s := range sigs {
    if hmac.Equal([]byte(want), []byte(s)) {
      return true
    }
  }
  return false
}

func main() {
  secret := os.Getenv("FLOW_MESSAGING_WEBHOOK_SECRET")
  http.HandleFunc("/flow", func(w http.ResponseWriter, r *http.Request) {
    body, _ := io.ReadAll(r.Body)
    if !verify(body, r.Header.Get("Flow-Signature"), secret) {
      w.WriteHeader(http.StatusBadRequest)
      return
    }
    var ev struct {
      Type string `json:"type"`
      Data struct {
        Message struct {
          Content struct {
            Text string `json:"text"`
          } `json:"content"`
        } `json:"message"`
      } `json:"data"`
    }
    _ = json.Unmarshal(body, &ev)
    w.Header().Set("Content-Type", "application/json")
    if ev.Type != "message.received" {
      w.Write([]byte("{}"))
      return
    }
    json.NewEncoder(w).Encode(map[string]any{
      "reply": map[string]any{"kind": "basic", "text": "You said: " + ev.Data.Message.Content.Text},
    })
  })
  http.ListenAndServe(":3000", nil)
}

Reply in the webhook response

To answer a message.received at once, respond 200 with a reply body. Flow sends it into the event’s conversation through the send gate, piece by piece, as if you had called each piece’s kind’s endpoint (POST /v0/conversations/{conversation_id}/basic, /buttons, …) with the event’s id as the idempotency key. This saves a whole round trip.

{ "reply": { "kind": "basic", "text": "Yes, we deliver to 560103." } }

Each piece is a kind object with kind, the same fields as on its endpoint. A string is a basic message: {"reply": "Hi"} sends {"kind": "basic", "text": "Hi"}. reply may also be a list of up to 11 pieces, sent in order, so a text plus 10 cards sent one by one fits:

{ "reply": ["Here are the sizes we have.", { "kind": "buttons", "text": "Which size?", "buttons": [{ "id": "s", "label": "Small" }, { "id": "m", "label": "Medium" }] }] }

• An empty body, {}, {"reply": null} or a body that is not a JSON object (plain text such as OK) sends nothing, and is not an error.
• A JSON object with a reply that is not valid, or a body that starts with { but is not valid JSON (a piece that is not a kind object or a string, an unknown kind, an empty list, more than 11 pieces) sends nothing at all, not even the valid pieces. It is recorded on the delivery as an invalid_request error with the reason, which the MCP tool get_webhook_deliveries shows. The delivery counts as delivered and is not retried.
• A piece the send gate refuses (a carousel on Telegram answers 422 unsupported_content) arrives as a message.failed event. Nothing is converted: send what the channel takes instead.

Slow agents answer 200 {} at once and send later with the kinds’ endpoints (POST /v0/conversations/{conversation_id}/basic, …). Keep your handler under 10 seconds; anything that takes longer belongs after the response.

Delivery rules

• Answer 2xx within 10 seconds. Anything else, or no answer, is a failed delivery.
• Retries: failed deliveries are retried with backoff for 3 days. After that the event stays in the log, marked failed, and you can read it again with GET /v0/events.
• At least once: a delivery can repeat. Deduplicate on the event’s id (also in Flow-Event-Id).
• Ordered per conversation: events in one conversation arrive in order, one at a time. Later events in that conversation wait behind a failing one, so a slow endpoint holds back only its own conversations. See Order per person.
• Ordering in the log: events are ordered by commit, not by created_at. Reading forward with after never skips an event.
• Disabled endpoints keep their place: nothing is lost from the log.

Order per person

Order is kept per conversation, and a conversation is one person on one of your senders. The same person on two channels, or on two of your senders, is two conversations, with no order between them.

• What you send goes out one message at a time per conversation, in the order Flow accepted it. A send the channel refuses for now is retried in its place, and later messages to that person wait for it. A send that fails (message.failed) holds nothing back: the next one goes at once.
• As the person sees it: a message you send as several (a text, then one message per card) arrives in the order you sent the calls, across every endpoint, webhook replies and the stream. On WhatsApp, a message with media holds the next one briefly until WhatsApp reports on it, so the person sees them in the order you sent them.
• What the person sends becomes events in the channel’s order, one at a time per conversation. On Telegram that is Telegram’s own update order: a later message waits briefly for an earlier one still on its way. On WhatsApp it is WhatsApp’s timestamp: each message is held briefly so one whose webhook comes late can still go first. A message that arrives after that wait is delivered when it arrives.
• On iMessage, inbound messages carry no order from the channel, so they become events in the order Flow receives them. Two messages the person sends within about a second of each other can arrive swapped.
• A sandbox join gives conversation.started first, then the join itself as the person’s first message.received, then whatever they send next.
• Webhook retries never reorder a conversation: its next event waits until the one before is delivered, or marked failed after 3 days of retries.
• The stream and GET /v0/events give your app’s events in log order, so one conversation’s events come in order there too.

The live stream

GET /v0/stream upgrades to a WebSocket. Every frame is one JSON object: the server sends event, ack, error and reconnect frames, and you may send send frames ({"type": "send", "ref": ..., "conversation": "conv_...", "message": <kind object>}, the same as the kind’s endpoint) and start frames ({"type": "start", "ref": ..., "message": <a start>}, as POST /v0/messages). ref is your idempotency key, echoed in the ack or error. Without after the stream starts live, at the moment it opens: earlier events are not replayed (read them with GET /v0/events, or see Ask and wait for a reply). Pass after to resume: the stream first replays everything after that event, then goes live. When the server is about to restart it sends reconnect; reconnect with after set to the last event you received. The TypeScript SDK does all of this for you in flow.events.stream(), and polls GET /v0/events where the runtime has no WebSocket.

Authenticate with the Authorization header. Where your WebSocket cannot set headers (a browser’s WebSocket, and Node’s global WebSocket on a server), offer the key as a subprotocol instead: offer both flow and flow.key.<api key>. This is as valid from server-side code as from a browser. The server selects flow and never echoes the key; offering the key protocol without flow is refused with 400 invalid_request, and an Authorization header, when present, takes precedence. Keys are never accepted in the query string, since URLs end up in logs. The TypeScript SDK offers the subprotocol by itself where the WebSocket cannot send headers.

const ws = new WebSocket("wss://api.flow.engineer/v0/stream", ["flow", "flow.key." + key]);

A refused stream (a bad, revoked or expired key, a bad parameter, too many streams for the key) is still upgraded, because most WebSocket clients cannot read an HTTP status: the server sends one error frame with the usual error body, then closes with 4000 plus the HTTP status. Stop reconnecting on 4401 (authentication), 4403 (permission) and 4400 (invalid_request); reconnect after retry_after on 4429 and 4503. An open stream whose key is revoked also gets an authentication error frame and close 4401.

A key used in a browser is visible to whoever uses that page. Do this only for internal tools or with test keys (fk_test_); keep live keys on your server.

Related

• Local development: receive events on your laptop through the live stream, with no public URL.
• API: The webhook event, Create a webhook endpoint, List events, Open the live stream.