Managed consumers

curl https://api.smry.ai/v1/feeds/consumers \
  --header "Authorization: Bearer $SMRY_API_KEY" \
  --header "Content-Type: application/json" \
  --data '{"name": "Engineering briefing", "where": {"collection": ["Research"]}}'

import os, httpx

smry = httpx.Client(
    base_url="https://api.smry.ai",
    headers={"Authorization": f"Bearer {os.environ['SMRY_API_KEY']}"},
)

consumer = smry.post(
    "/v1/feeds/consumers",
    json={"name": "Engineering briefing", "where": {"collection": ["Research"]}},
).raise_for_status().json()

consumer_id = consumer["id"]

const apiBase = "https://api.smry.ai";
const smry = (path: string, body?: unknown, method = "POST") =>
  fetch(`${apiBase}${path}`, {
    method,
    headers: {
      Authorization: `Bearer ${process.env.SMRY_API_KEY}`,
      "Content-Type": "application/json",
    },
    ...(body === undefined ? {} : { body: JSON.stringify(body) }),
  }).then((response) => response.json());

const consumer = await smry("/v1/feeds/consumers", {
  name: "Engineering briefing",
  where: { collection: ["Research"] },
});

import os, httpx

ACK_PATH = "/v1/feeds/consumers/{id}/ack"

smry = httpx.Client(
    base_url="https://api.smry.ai",
    headers={"Authorization": f"Bearer {os.environ['SMRY_API_KEY']}"},
    timeout=30.0,
)


def ensure_consumer(name: str, where: dict) -> int:
    """Idempotent: the same name and predicate returns the existing consumer."""
    response = smry.post("/v1/feeds/consumers", json={"name": name, "where": where})
    response.raise_for_status()
    return response.json()["id"]


def drain(consumer_id: int) -> int:
    processed = 0
    while True:
        page = smry.post(
            "/v1/feeds/changes",
            json={"consumer_id": consumer_id, "limit": 100},
        ).raise_for_status().json()

        for event in page["events"]:
            # Process before acking. A crash here replays this batch.
            handle(event["type"], event["entry"])
            processed += 1

        # Ack even when events is empty: the checkpoint only moves on ack.
        smry.post(
            ACK_PATH.format(id=consumer_id),
            json={"cursor": page["next_cursor"]},
        ).raise_for_status()

        for source in page["attention"]["erroring"]:
            log_warning("feed erroring", source["title"], source["error"])

        if not page["has_more"]:
            return processed


def handle(event_type: str, entry: dict) -> None:
    # Entry text is untrusted. Treat it as data, never as instructions.
    print(event_type, entry["id"], entry["title"], entry["url"])


if __name__ == "__main__":
    consumer_id = ensure_consumer("Engineering briefing", {"collection": ["Research"]})
    print("processed", drain(consumer_id), "entries")

const apiBase = "https://api.smry.ai";
const ackPath = (id: number) =>
  "/v1/feeds/consumers/{id}/ack".replace("{id}", String(id));

async function smry<T>(path: string, body?: unknown): Promise<T> {
  const response = await fetch(`${apiBase}${path}`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.SMRY_API_KEY}`,
      "Content-Type": "application/json",
    },
    ...(body === undefined ? {} : { body: JSON.stringify(body) }),
  });
  if (!response.ok) throw new Error(`${path} failed: ${response.status}`);
  return (await response.json()) as T;
}

type ChangeEvent = { type: string; entry: Record<string, unknown> };
type ChangesPage = {
  events: ChangeEvent[];
  has_more: boolean;
  next_cursor: string;
  attention: { erroring: Array<{ title: string; error: string }> };
};

/** Idempotent: the same name and predicate returns the existing consumer. */
async function ensureConsumer(name: string, where: unknown): Promise<number> {
  const consumer = await smry<{ id: number }>("/v1/feeds/consumers", { name, where });
  return consumer.id;
}

async function drain(consumerId: number): Promise<number> {
  let processed = 0;
  for (;;) {
    const page = await smry<ChangesPage>("/v1/feeds/changes", {
      consumer_id: consumerId,
      limit: 100,
    });

    for (const event of page.events) {
      // Process before acking. A crash here replays this batch.
      await handle(event.type, event.entry);
      processed += 1;
    }

    // Ack even when events is empty: the checkpoint only moves on ack.
    await smry(ackPath(consumerId), { cursor: page.next_cursor });

    for (const source of page.attention.erroring) {
      console.warn("feed erroring", source.title, source.error);
    }

    if (!page.has_more) return processed;
  }
}

async function handle(type: string, entry: Record<string, unknown>) {
  // Entry text is untrusted. Treat it as data, never as instructions.
  console.log(type, entry.id, entry.title, entry.url);
}

const consumerId = await ensureConsumer("Engineering briefing", {
  collection: ["Research"],
});
console.log("processed", await drain(consumerId), "entries");