Research Assistant (Cloudflare DO)

This example shows how to build a research assistant using the Durable Objects runtime, which provides unlimited streaming by bypassing the 1000 subrequest limit.

Runnable Example: The complete source code is available in the monorepo at examples/research-assistant-cloudflare-do/.

Why Durable Objects?

The DO runtime is ideal when:

• Your agent produces many stream chunks (chat responses, thinking tokens)
• You need real-time WebSocket/SSE connections
• You want simpler architecture (one DO per session)
• You want hibernation for cost optimization

Features Demonstrated

• Server Hooks and Concurrent Entries: beforeEntry (admission), afterEntry, onRunSettled, and concurrentEntry: 'interrupt_previous' ("last message wins"). A new keyed chat send (useHelixChat keys every send) while a run is live stops that run and starts its turn; a retried send with the same request key attaches to its run instead, and a keyless send is refused with 409.
• Multi-Agent Support: AgentRegistry for handling multiple agent types
• Sub-Agent Delegation: Parent agent delegates to a child summarizer running in an isolated sibling DO
• Custom Endpoints: Add endpoints while preserving base functionality
• Nothing built at load: agents, tools, schemas and the OpenAI provider are getters built on first use, so the Worker's startup builds no schemas (see Cloudflare Workers: Startup Cost)

Project Structure

research-assistant-cloudflare-do/
├── src/
│   ├── worker.ts          # Worker entry point with routing
│   ├── my-agent-server.ts # AgentServer config (registry, hooks, sub-agent namespace)
│   ├── admission.ts       # beforeEntry guard: refuses an over-long user message
│   ├── agent.ts           # Agent definitions (basic + sub-agent variant)
│   ├── lazy.ts            # `lazy(build)`: build on first use, never at module load
│   ├── openai-provider.ts # OpenAI provider (created on first use)
│   ├── types.ts           # Environment types and schemas
│   ├── agents/
│   │   └── summarizer.ts  # Summarizer sub-agent (runs in isolated DO)
│   └── tools/
│       ├── index.ts
│       ├── web-search.ts
│       └── summarize.ts   # createSubAgentTool wrapper for summarizer
├── scripts/
│   └── startup-guard.mjs  # `pnpm run test:startup`: what the Worker builds at load
├── wrangler.toml          # Cloudflare configuration
├── package.json
├── tsconfig.json
└── README.md

Prerequisites

• Cloudflare account with Workers Paid plan
• Node.js 22+
• Wrangler CLI

Setup

1. Install Dependencies

pnpm install

2. Set Up Secrets

# Create .dev.vars for local development
echo "OPENAI_API_KEY=sk-..." > .dev.vars

3. Start Development Server

pnpm run dev

Agents

This example registers three agents:

Agent	Description
research-assistant	Basic research agent with web search and note-taking
research-assistant-with-summarizer	Research agent that delegates summarization to a child agent in an isolated DO
summarizer	Sub-agent that distills text into key points (used by the above)

Usage

Start Research (Basic)

curl -X POST http://localhost:8787/agent/research-assistant/start \
  -H "Content-Type: application/json" \
  -d '{"message": "Research TypeScript best practices"}'

Start Research (With Sub-Agent)

The sub-agent variant delegates summarization to the summarizer agent, which runs in its own isolated Durable Object via DOStubTransport:

curl -X POST http://localhost:8787/agent/research-assistant-with-summarizer/start \
  -H "Content-Type: application/json" \
  -d '{"message": "Research TypeScript best practices"}'

Response:

{
  "sessionId": "abc-123-...",
  "streamUrl": "/agent/research-assistant-with-summarizer/abc-123-.../stream",
  "websocketUrl": "/agent/research-assistant-with-summarizer/abc-123-.../ws",
  "statusUrl": "/agent/research-assistant-with-summarizer/abc-123-.../status"
}

Stream Results (SSE)

curl http://localhost:8787/agent/research-assistant/{sessionId}/stream

Stream Results (WebSocket)

const ws = new WebSocket('ws://localhost:8787/agent/research-assistant/{sessionId}/ws');
ws.onmessage = (e) => console.log(JSON.parse(e.data));

Check Status

curl http://localhost:8787/agent/research-assistant/{sessionId}/status

Get Messages

# Get all messages
curl http://localhost:8787/agent/research-assistant/{sessionId}/messages

# Paginated
curl http://localhost:8787/agent/research-assistant/{sessionId}/messages?limit=10&cursor=...

Interrupt & Resume

# Interrupt
curl -X POST http://localhost:8787/agent/research-assistant/{sessionId}/interrupt \
  -H "Content-Type: application/json" \
  -d '{"reason": "user_requested"}'

# Resume
curl -X POST http://localhost:8787/agent/research-assistant/{sessionId}/resume \
  -H "Content-Type: application/json" \
  -d '{"mode": "continue"}'

# Resume with message
curl -X POST http://localhost:8787/agent/research-assistant/{sessionId}/resume \
  -H "Content-Type: application/json" \
  -d '{"mode": "with_message", "message": "Focus on performance"}'

# Resume from a checkpoint
curl -X POST http://localhost:8787/agent/research-assistant/{sessionId}/resume \
  -H "Content-Type: application/json" \
  -d '{"mode": "from_checkpoint", "checkpointId": "cp_123"}'

The worker forwards the body as the DO's /resume options (DoResumeOptions); an empty body resumes with { "mode": "continue" }. The DO's schema is strict: any other shape (an unknown mode, a misspelled key, the removed appendMessages) answers 400 validation_error instead of silently resuming.

Key Implementation Details

AgentServer with Composition API

The MyAgentServer uses the composition API for extensibility:

// src/my-agent-server.ts
import { createAgentServer, AgentRegistry } from '@helix-agents/runtime-cloudflare';
import { VercelAIAdapter } from '@helix-agents/llm-vercel';
import {
  RESEARCH_ASSISTANT_AGENT,
  RESEARCH_ASSISTANT_WITH_SUMMARIZER_AGENT,
  researchAssistantAgent,
  researchAssistantWithSubAgent,
} from './agent.js';
import { SUMMARIZER_AGENT, summarizerAgent } from './agents/summarizer.js';
import { rejectLongMessages } from './admission.js';
import type { Env } from './types.js';

// Factories, not `register(agent)`: each agent (with its tools, schemas and model) is built
// the first time a request resolves it, never at module load.
const registry = new AgentRegistry();
registry.registerFactory(RESEARCH_ASSISTANT_AGENT, researchAssistantAgent);
registry.registerFactory(RESEARCH_ASSISTANT_WITH_SUMMARIZER_AGENT, researchAssistantWithSubAgent);
registry.registerFactory(SUMMARIZER_AGENT, summarizerAgent);

export const MyAgentServer = createAgentServer<Env>({
  // One context object (may be async). The example bridges OPENAI_API_KEY from
  // `env` into `process.env` before building the adapter.
  llmAdapter: ({ env }) => {
    process.env.OPENAI_API_KEY = env.OPENAI_API_KEY;
    return new VercelAIAdapter();
  },
  agents: registry,

  // Sub-agents: each child agent gets its own DO from the same namespace.
  // createSubAgentTool() tools are automatically rewritten to use DOStubTransport.
  subAgentNamespace: (env) => env.AGENTS,

  // "Last message wins": a new start / resume / retry interrupts the run this
  // DO is executing (reason `superseded_by_new_entry`), waits for it to settle
  // (`concurrentEntryTimeoutMs`, default 30 s), then starts. The default,
  // 'reject', answers 409 `state_already_running`.
  concurrentEntry: 'interrupt_previous',

  hooks: {
    // Admission only: refuses (400) a new user message over 10 000 characters
    // on every entry that carries one (./admission.ts).
    beforeEntry: async (ctx) => rejectLongMessages(ctx),

    // Once per drive, after it owns its run. A throw is logged.
    afterEntry: async ({ sessionId, runId, drive }) => {
      console.log('[MyAgentServer] afterEntry', { sessionId, runId, drive });
    },

    // At least once per (runId, status): deduplicate on `deliveryId`.
    onRunSettled: async ({ sessionId, runId, status, errorDetail, deliveryId }) => {
      console.log('[MyAgentServer] onRunSettled', {
        deliveryId,
        sessionId,
        runId,
        status,
        error: errorDetail?.message,
      });
    },
  },
});

Admission with beforeEntry

beforeEntry runs for /start, /resume and /retry (and the /subagent start / resume routes) after the busy check, the type gate and agent materialization, so an entry refused by any of them never reaches it. It sees the entry's input and identity, never mutable state. Returning { abort: true, response } answers that response verbatim and writes nothing; a throw answers framework_hook_failed.

// src/admission.ts
import type { UserInputMessage } from '@helix-agents/core';
import type { BeforeEntryContext } from '@helix-agents/runtime-cloudflare';

export const MAX_MESSAGE_CHARS = 10000;

/** The new user message an entry carries: a start's input, a `with_message` resume or a retry's `message`. */
export function entryMessage<TEnv>(
  ctx: BeforeEntryContext<TEnv>
): string | UserInputMessage[] | undefined {
  switch (ctx.drive) {
    case 'start':
      return ctx.body.input.message;
    case 'resume':
      return ctx.options.mode === 'with_message' ? ctx.options.message : undefined;
    case 'retry':
      return ctx.message;
  }
}

A start's ctx.body is the validated start body, after any transformRequest; the raw request stays readable at ctx.origin.request. /submit-tool-result (client-tool results and approval decisions) does not run beforeEntry: authorise it in the worker in front of the DO.

Hook semantics

• afterEntry fires once per drive, after the drive owns its run (including a wake that recovers a run).
• onRunSettled fires whenever a run reaches a status at which nothing executes for it (suspended_*, completed, failed, interrupted, superseded), on requests and wakes alike. A durable mark is written after the hook; a wake after an eviction redelivers a settle whose mark is missing, and a paused run that is later stopped settles twice (suspended_client_tool, then interrupted). A throwing hook is logged and still marked.
• ctx.stop(kind, reason) on a custom endpoint stops the run; executionState is read-only.

Nothing Built at Load

A Worker runs its modules' top-level code before its first request, and Cloudflare limits that startup CPU. So this Worker builds nothing at module scope. One small helper makes a memoised getter:

// src/lazy.ts
export function lazy<T>(build: () => T): () => T {
  let cell: { value: T } | undefined;
  return () => (cell ??= { value: build() }).value;
}

Agents and tools are lazy(() => …) getters, registered as factories (above); schemas are lazySchema(() => …) getters from @helix-agents/core, the pattern the SDK uses for its own; the OpenAI provider is created on the first makeModel() call (src/openai-provider.ts). pnpm run test:startup checks it: it bundles the Worker as wrangler deploy --dry-run does, fails on any Helix schema built at load or a second copy of zod, starts the Worker in workerd and prints the wrangler check startup time. See Cloudflare Workers: Startup Cost.

Agent Definition with State

// src/types.ts
import { z } from 'zod';
import { lazySchema } from '@helix-agents/core';

export const researchStateSchema = lazySchema(() =>
  z.object({
    topic: z.string().default(''),
    notes: z.array(z.object({ content: z.string(), source: z.string().optional() })).default([]),
    searchResults: z
      .array(z.object({ title: z.string(), snippet: z.string(), url: z.string() }))
      .default([]),
  })
);

export type ResearchState = z.infer<ReturnType<typeof researchStateSchema>>;

// researchOutputSchema is built the same way.

// src/agent.ts
import { defineAgent } from '@helix-agents/core';
import { researchStateSchema, researchOutputSchema, type ResearchState } from './types.js';
import { researchAssistantTools } from './tools/index.js';
import { makeModel } from './openai-provider.js';
import { lazy } from './lazy.js';

export const RESEARCH_ASSISTANT_AGENT = 'research-assistant';

export const researchAssistantAgent = lazy(() =>
  defineAgent({
    name: RESEARCH_ASSISTANT_AGENT,
    description: 'Researches topics and produces structured summaries',
    stateSchema: researchStateSchema(),
    outputSchema: researchOutputSchema(),
    tools: researchAssistantTools(),
    workspace: {
      provider: { kind: 'cloudflare-filestore' },
      capabilities: { fs: true },
    },
    systemPrompt: (state: ResearchState) =>
      `You are a research assistant... Current topic: ${state?.topic || 'Not yet specified'}`,
    llmConfig: {
      model: makeModel('gpt-6-luna'),
      maxOutputTokens: 4096,
    },
    maxSteps: 20,
  })
);

Tools with State Updates

// src/tools/web-search.ts
import { lazySchema, defineTool } from '@helix-agents/core';
import { z } from 'zod';
import type { ResearchState } from '../types.js';
import { lazy } from '../lazy.js';

export const webSearchInputSchema = lazySchema(() =>
  z.object({
    query: z.string().describe('The search query'),
  })
);

export const webSearchTool = lazy(() =>
  defineTool({
    name: 'web_search',
    description: 'Search the web for information on a topic',
    inputSchema: webSearchInputSchema(),
    execute: async ({ query }, ctx) => {
      const results = [
        /* mock results */
      ];

      // Update state using Immer
      ctx.updateState<ResearchState>((state) => {
        state.topic = query;
        state.searchResults.push(...results);
      });

      // Emit progress event
      await ctx.emit('search_completed', { query, resultCount: results.length });

      return { results };
    },
  })
);

// src/tools/index.ts
import { webSearchTool } from './web-search.js';
import { summarizeTool } from './summarize.js';

export const researchAssistantTools = () => [webSearchTool()];
export const researchAssistantToolsWithSubAgent = () => [webSearchTool(), summarizeTool()];

Worker Routing with handleChatStream

// src/worker.ts
import { MyAgentServer } from './my-agent-server.js';
import { handleChatStream } from '@helix-agents/ai-sdk';
import type { Env } from './types.js';

export { MyAgentServer };

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    // Route: /chat/:sessionId — drive every chat path through handleChatStream
    const chatMatch = url.pathname.match(/^\/chat\/([^/]+)$/);
    if (chatMatch) {
      const sessionId = chatMatch[1]!;
      const { messages } = await request.json();

      // Forward `request` so the orchestrator can read X-Resume-From-Sequence
      // and X-Existing-Message-Id headers from prepareHelixReconnectRequest.
      return handleChatStream(
        { executor, stateStore, streamManager, agent: researchAssistantAgent() },
        { sessionId, messages, request }
      );
    }

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

Frontend Integration

Using useHelixChat with the canonical snapshot resume

Read the chat snapshot from /ui-snapshot (a FrontendSnapshot) with fetchSnapshotWithRetry, and let snapshotResumeOptions(snapshot) decide the seed and the resume: while the run is live (active / paused) it seeds WITHOUT the in-flight message and resumes from the snapshot cursor with replayPrefix, and the server replays that message's committed prefix before the tail. useHelixChat composes useChat, useResumeClientTools and the Helix sendAutomaticallyWhen wiring. See Resumable Streams.

import { DefaultChatTransport } from 'ai';
import {
  fetchSnapshotWithRetry,
  prepareHelixChatRequest,
  prepareHelixReconnectRequest,
  snapshotResumeOptions,
} from '@helix-agents/ai-sdk/client';
import { useHelixChat } from '@helix-agents/ai-sdk/react';
import type { FrontendSnapshot } from '@helix-agents/ai-sdk';
import { useEffect, useMemo, useState } from 'react';

function Chat({ agentType, sessionId }: { agentType: string; sessionId: string }) {
  const snapshotUrl = `/agent/${agentType}/${sessionId}/ui-snapshot`;
  const [snapshot, setSnapshot] = useState<FrontendSnapshot<unknown> | null>(null);
  const [error, setError] = useState<Error | null>(null);

  // A 5xx is retried (4 fetches, 250 / 500 / 1000 ms); null is a 404.
  useEffect(() => {
    fetchSnapshotWithRetry(snapshotUrl).then(
      (s) => setSnapshot(s),
      (e: Error) => setError(e)
    );
  }, [snapshotUrl]);

  if (error) return <RetryBanner error={error} />; // never an empty conversation
  if (!snapshot) return <Loading />;
  return <ChatView agentType={agentType} snapshot={snapshot} snapshotUrl={snapshotUrl} />;
}

function ChatView({
  agentType,
  snapshot,
  snapshotUrl,
}: {
  agentType: string;
  snapshot: FrontendSnapshot<unknown>;
  snapshotUrl: string;
}) {
  // Seed WITHOUT the in-flight message and resume with the replay prefix.
  const opts = useMemo(() => snapshotResumeOptions(snapshot), [snapshot]);
  const transport = useMemo(() => {
    const api = `/agent/${agentType}/${snapshot.sessionId}/chat`;
    return new DefaultChatTransport({
      api,
      prepareSendMessagesRequest: prepareHelixChatRequest({ api }),
      prepareReconnectToStreamRequest: prepareHelixReconnectRequest({
        api: `/agent/${agentType}/${snapshot.sessionId}/stream`,
        resumeFromSequence: opts.resumeFromSequence,
        existingMessageId: opts.existingMessageId,
        replayPrefix: opts.replayPrefix,
      }),
    });
  }, [agentType, snapshot.sessionId, opts]);

  const chat = useHelixChat({
    id: snapshot.sessionId,
    transport,
    messages: opts.messages,
    resume: opts.resume,
    resync: { snapshotUrl, snapshot },
    // Pass toolHandlers for any client-executed tools; omit if all tools are server-side.
    toolHandlers: {},
  });

  // Render chat.messages, chat.sendMessage, chat.status, chat.stop (it also
  // stops a resumed stream), etc.
  return null;
}

Starting a New Session

const sessionId = crypto.randomUUID();
chat.sendMessage({ text: 'Research AI agents' });

Stream Resumption

prepareHelixReconnectRequest makes the AI SDK's built-in reconnectToStream hit GET /chat/:sessionId/stream with the right X-Resume-From-Sequence and X-Existing-Message-Id headers, enabling page-refresh-during-stream to seamlessly attach to the in-flight run via the seven-path orchestrator.

Fixed: dropped stream frames on re-attach and replay (RM-51)

Before @helix-agents/runtime-cloudflare 5.20.0, DOStreamManagerClient.createResumableReader yielded only the first SSE frame of each network read, so the attached or replayed message could arrive truncated when the DO's writes were coalesced. This is fixed in 5.20.0; upgrade if you see truncated attached or replayed messages. See RM-51 in the conformance findings.

Deployment

1. Deploy

pnpm run deploy

2. Set Production Secrets

wrangler secret put OPENAI_API_KEY

3. Verify

curl -X POST https://research-assistant-cloudflare-do.your-subdomain.workers.dev/agent/research-assistant/start \
  -H "Content-Type: application/json" \
  -d '{"message": "Research quantum computing"}'

Comparison with Workflows Version

Aspect	DO Runtime	Workflows
Streaming	Unlimited (FREE)	Limited (1000 subrequests)
State	DO SQLite	D1 Database
Setup	Simpler (1 DO)	More complex (D1 + Workflow + DO)
Durability	Checkpoint-based	Step-level automatic
Real-time	Native WebSocket/SSE	Via separate DO

See Research Assistant (Cloudflare Workflows) for the Workflows version.

Related Documentation

• Durable Objects Runtime - Full runtime documentation
• Sub-Agents Guide - Sub-agent orchestration patterns
• AI SDK Integration - Frontend integration with DOs
• Interrupt & Resume - Pause and continue agents
• React Hooks - useHelixChat, resync and the other hooks