The webhook inbox for apps and agents

Poll for events

curl -G \
  "https://app.dotpipe.io/api/sources/$SOURCE_ID/events" \
  -H "Authorization: Bearer $DOTPIPE_API_KEY" \
  --data-urlencode "after=42" \
  --data-urlencode "limit=100"

Stream in realtime

curl -N -G \
  "https://app.dotpipe.io/api/sources/$SOURCE_ID/events/sse" \
  -H "Authorization: Bearer $DOTPIPE_API_KEY" \
  -H "Accept: text/event-stream" \
  --data-urlencode "after=42"

Each SSE message carries one complete event. -N turns off buffering, so events print as they arrive.

Poll for events

async function pollEvents(after: number) {
  const response = await fetch(
    `https://app.dotpipe.io/api/sources/${sourceId}/events?after=${after}&limit=100`,
    { headers: { Authorization: `Bearer ${apiKey}` } },
  )
  if (!response.ok) throw new Error(`HTTP ${response.status}`)
  return response.json()
}

Stream in realtime

import { EventSource } from "eventsource"

const stream = new EventSource(
  `https://app.dotpipe.io/api/sources/${sourceId}/events/sse?after=42`,
  {
    fetch: (url, init) =>
      fetch(url, {
        ...init,
        headers: { ...init?.headers, Authorization: `Bearer ${apiKey}` },
      }),
  },
)

stream.addEventListener("source-event", (message) => {
  const event = JSON.parse(message.data)
  console.log(event.cursor, event.payload)
})

Each message id is the event cursor, so the standardLast-Event-ID reconnect resumes where you stopped.

Poll for events

import json
import os
from urllib.parse import urlencode
from urllib.request import Request, urlopen

def poll_events(after):
    source_id = os.environ["SOURCE_ID"]
    query = urlencode({"after": after, "limit": 100})
    url = f"https://app.dotpipe.io/api/sources/{source_id}/events?{query}"
    request = Request(url, headers={
        "Authorization": f"Bearer {os.environ['DOTPIPE_API_KEY']}"
    })

    with urlopen(request) as response:
        return json.load(response)

Stream in realtime

source_id = os.environ["SOURCE_ID"]
request = Request(
    f"https://app.dotpipe.io/api/sources/{source_id}/events/sse?after=42",
    headers={
        "Authorization": f"Bearer {os.environ['DOTPIPE_API_KEY']}",
        "Accept": "text/event-stream",
    },
)

with urlopen(request) as stream:
    for line in stream:
        line = line.decode()
        if line.startswith("data:"):
            event = json.loads(line[len("data:"):])
            print(event["cursor"], event["payload"])

Catch up and stream in realtime

import * as DotpipeClient from "@dotpipe/client"
import { Effect, Redacted, Stream } from "effect"
import { FetchHttpClient } from "effect/unstable/http"

const program = Effect.gen(function* () {
  const client = yield* DotpipeClient.make({
    apiKey: Redacted.make(process.env.DOTPIPE_API_KEY!),
    sourceId: process.env.SOURCE_ID!,
  })

  return yield* client.events(DotpipeClient.Cursor.make(42)).pipe(
    Stream.tap((event) => Effect.log(event.cursor)),
    Stream.runDrain,
  )
}).pipe(Effect.provide(FetchHttpClient.layer))

Effect.runPromise(program)