Local development

Receive messages on your laptop through the live event stream, without a public URL or tunnel. Save the last handled event id and pass it as after when restarting to replay missed events.

TypeScript

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

const flow = new FlowMessaging(); // reads FLOW_MESSAGING_KEY

for await (const event of flow.events.stream({ types: ["message.received"] })) {
  await event.conversation.reply(`You said: ${contentText(event.data.message.content)}`, {
    idempotencyKey: event.id,
  });
}

How it works

1. Connect to wss://api.flow.engineer/v0/stream with Authorization: Bearer $FLOW_MESSAGING_KEY. Filter with type (repeat it for several types).
2. Every text frame is one JSON object. event frames carry the same event a webhook would receive.
3. Reply over the API (POST /v0/conversations/{conversation_id}/basic and the other kinds’ endpoints), or send a send frame on the same socket, whose message is a kind object ({"type": "send", "ref": "evt_123", "conversation": "conv_...", "message": {"kind": "basic", "text": "Hi"}}); each send gets one ack or error frame back, matched by ref (your idempotency key).
4. Keep the id of the last event you handled. When you reconnect, or when the server sends a reconnect frame before it restarts, connect again with after set to that ID: the stream replays everything after it, then goes live.

The TypeScript SDK’s flow.events.stream() does the reconnecting and resuming for you. In other languages use any WebSocket client that can set a header:

Python

# pip install websockets
import asyncio, json, os, websockets

async def main():
    url = "wss://api.flow.engineer/v0/stream?type=message.received"
    headers = {"Authorization": f"Bearer {os.environ['FLOW_MESSAGING_KEY']}"}
    async with websockets.connect(url, additional_headers=headers) as ws:
        async for raw in ws:
            frame = json.loads(raw)
            print(frame)  # an "event" frame; reply with POST /v0/conversations/{id}/basic

asyncio.run(main())

See The live stream for the frame types and browser authentication.

Testing your webhook handler locally

If your production code is a webhook handler, you can still develop it on localhost: read the stream and pass each event to your handler function directly, then deploy the same handler behind a real webhook endpoint. The stream and your webhook endpoints read the same log and do not interfere: both receive events.

The CLI wraps the stream for webhook handlers: listen --forward-to reads GET /v0/stream and POSTs each event to your local URL, signed like a real delivery.

npx @flow-engineer/messaging listen --forward-to http://localhost:3000/api/flow

• Each event is POSTed with the same headers and signature as a real webhook delivery (Flow-Signature, Flow-Event-Id, Flow-Event-Type, Flow-Version), signed with FLOW_MESSAGING_WEBHOOK_SECRET (made and saved to .env if it is not set).
• If your handler answers a message.received with {"reply": ...}, listen sends the reply into the conversation, with the event’s ID as the idempotency key, as the API does for real webhooks.
• Flags: --events a,b (only these event types), --after evt_... (replay from this event first, then go live), --secret whsec_... (the signing secret to use). Without --forward-to, listen only prints events.

Tips

• Use a test key while developing: only people who joined your sandbox can reach your agent.
• To run past events through new code, find the event just before them with GET /v0/events and pass its ID as after.
• The stream is for long-running processes. In production, serverless apps usually use webhooks.

Related

• Events and webhooks
• For AI coding agents: the MCP server can wait for events and replay them too.
• API: Open the live stream