Package Exports
- @projectaria/cf-agents
Readme
@projectaria/cf-agents
ARIA integration for Cloudflare Workers with Durable Objects. Build stateful AI agents with persistent memory, session management, and multi-turn conversations using structured flows.
Installation
npm install @projectaria/cf-agents agents @cloudflare/ai-chat aiMessage Flow Architecture
┌─────────────────────────────────────────────────────────────────────────────┐
│ MESSAGE FLOW DIAGRAM │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ CLIENT SIDE SERVER SIDE │
│ ──────────── ──────────── │
│ │
│ User Input │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Cloudflare AIChatAgent │ │
│ │ this.messages (UIMessage[]) ←── WebSocket/HTTP │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ onChatMessage() │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ AriaCFChatAgent │ │
│ │ │ │
│ │ 1. extractLastUserInput() → userText │ │
│ │ 2. cloudflareToAriaSession() → AgentSession │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ 3. ariaAgent.streamText({ input, session }) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ (AsyncIterable<AriaStreamPart>) │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ ARIA Agent │ │
│ │ createAgent() │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ streamText() │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌────────────────────────────────────────────────────────────────┐ │ │
│ │ │ PromptBuilder + Conversation Context │ │ │
│ │ └────────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌────────────────────────────────────────────────────────────────┐ │ │
│ │ │ AI SDK LanguageModel │ │ │
│ │ │ streamText(model, messages, tools) │ │ │
│ │ └────────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ (stream of TextStreamPart) │ │
│ │ ┌────────────────────────────────────────────────────────────────┐ │ │
│ │ │ ResponseService / Tool Execution │ │ │
│ │ └────────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ (AriaStreamPart: text-delta, tool-call, tool-result, ...) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ ariaToUIStream() │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ aria-to-ui-stream.ts │ │
│ │ AriaStreamPart → TextUIPart | ToolCallUIPart | ToolResultUIPart │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ createUIMessageStreamResponse() │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Response (SSE/WebSocket) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ CLIENT UI ←───────────────────────────────────────────────────────────────
│ │
├─────────────────────────────────────────────────────────────────────────────┤
│ SESSION & STATE FLOW │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ VoltRuntime │ │
│ │ │ │
│ │ ┌───────────────────────────────────────────────────────────────┐ │ │
│ │ │ FlowSupervisor │ │ │
│ │ │ • Selects active flow based on intent │ │ │
│ │ │ • Handles flow switching mid-conversation │ │ │
│ │ └───────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌───────────────────────────────────────────────────────────────┐ │ │
│ │ │ StreamingFlowManager │ │ │
│ │ │ • Executes current node │ │ │
│ │ │ • Validates extracted data │ │ │
│ │ │ • Evaluates transitions │ │ │
│ │ └───────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌───────────────────────────────────────────────────────────────┐ │ │
│ │ │ CombinedAnalysisService │ │ │
│ │ │ • LLM validation of extracted fields │ │ │
│ │ │ • Structured data extraction │ │ │
│ │ └───────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌───────────────────────────────────────────────────────────────┐ │ │
│ │ │ ProgressionEngine │ │ │
│ │ │ • Evaluates condition expressions │ │ │
│ │ │ • Determines next node │ │ │
│ │ └───────────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌───────────────────────────────────────────────────────────────┐ │ │
│ │ │ SessionStore (SQLite via D1) │ │ │
│ │ │ • Persists session.history │ │ │
│ │ │ • Persists session.workingMemory │ │ │
│ │ │ • Restores sessions on agent restart │ │ │
│ │ └───────────────────────────────────────────────────────────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘Message Handling Sequence
// 1. Client sends message via WebSocket or HTTP
// → Cloudflare AIChatAgent receives it
// → Stores in this.messages
// 2. AriaCFChatAgent.onChatMessage() is called
async onChatMessage(onFinish, options) {
// 2a. Extract user text from last message
const userText = extractLastUserInput(this.messages);
// 2b. Convert Cloudflare messages to ARIA session format
const ariaSession = cloudflareToAriaSession(this.messages, this.state);
// 2c. Start session in VoltRuntime (loads or creates)
await this.voltRuntime.startSession(ariaSession.sessionId);
// 2d. Stream from ARIA agent
const result = await this.ariaAgent.streamText({
input: userText,
session: ariaSession,
});
// 2e. Convert ARIA stream parts to UI format
return createARIAStreamResponse(result.fullStream, onFinish);
}
// 3. ARIA Agent processes the stream
streamText({ input, session }) {
// 3a. Build prompt from flow + session history
const prompt = PromptBuilder.build({
flow: currentFlow,
session,
instructions,
guidelines,
});
// 3b. Call LLM with prompt + tools
return streamText(model, messages, tools);
}
// 4. Stream parts are emitted:
// - text-delta → UI text chunk
// - tool-call → UI tool call (input-available)
// - tool-result → UI tool result (output-available)
// - reasoning-delta → UI reasoning part
// 5. Session is updated with working memory changes
// → Persisted to SQLite via SessionStoreQuick Start
1. Create Your Agent with a Flow
// src/agent.ts
import { createAriaCFChatAgent } from "@projectaria/cf-agents";
import type { FlowDefinition } from "@projectaria/aria-agents";
import { z } from "zod";
interface Env {
AI: Ai;
DB: D1Database;
}
// Define a conversation flow
const greetingFlow: FlowDefinition = {
id: "greeting",
name: "Greeting Assistant",
startNodeId: "greet",
nodes: {
greet: {
id: "greet",
type: "task",
instructions: "Welcome the user and ask how you can help.",
transitions: [
{ nextNodeId: "help", condition: "intent == 'help'" },
],
},
help: {
id: "help",
type: "task",
instructions: "Provide helpful assistance.",
transitions: [
{ nextNodeId: "greet", condition: "always" },
],
},
},
};
// Create agent with any AI SDK model
const MyAgent = createAriaCFChatAgent<Env>({
flows: new Map([[greetingFlow.id, greetingFlow]]),
model: env.AI("@cf/meta/llama-3.1-8b-instruct"), // Any LanguageModel
instructions: "You are a friendly assistant.",
});
export { MyAgent };2. Configure Your Worker
// src/index.ts
import { routeAgentRequest } from "agents";
import { MyAgent } from "./agent";
export { MyAgent };
export default {
async fetch(request: Request, env: Env): Promise<Response> {
return (
(await routeAgentRequest(request, env)) ||
new Response("Not found", { status: 404 })
);
},
};3. Configure Wrangler
# wrangler.toml
name = "my-assistant"
main = "src/index.ts"
compatibility_date = "2025-02-11"
[[d1_databases]]
binding = "DB"
database_name = "aria-sessions"
[ai]
binding = "AI"4. Deploy
npx wrangler deployFeatures
Flow Orchestration
Define structured multi-turn conversations:
const orderFlow: FlowDefinition = {
id: "order",
startNodeId: "collect-items",
nodes: {
"collect-items": {
id: "collect-items",
type: "task",
instructions: "Ask which items they want to order.",
transitions: [
{ nextNodeId: "confirm", condition: "items selected" },
],
},
confirm: {
id: "confirm",
type: "task",
instructions: "Show order summary and ask for confirmation.",
transitions: [
{ nextNodeId: "complete", condition: "confirmed" },
{ nextNodeId: "collect-items", condition: "cancelled" },
],
},
},
};Working Memory
Persist data across conversation turns:
const MyAgent = createAriaCFChatAgent<Env>({
flows: myFlows,
model: env.AI("@cf/meta/llama-3.1-8b-instruct"),
});
class ShoppingAssistant extends MyAgent {
async onChatMessage(onFinish) {
// Working memory is automatically managed by ARIA
// Access via session.workingMemory in tools or flows
}
}Tools
Register custom tools for your agent:
import { tool } from "ai";
import { z } from "zod";
const MyAgent = createAriaCFChatAgent<Env>({
flows: myFlows,
model: env.AI("@cf/meta/llama-3.1-8b-instruct"),
tools: {
searchProducts: tool({
description: "Search for products",
parameters: z.object({
query: z.string(),
category: z.string().optional(),
}),
execute: async ({ query, category }) => {
// Your implementation
return [{ id: "1", name: "Product", price: 9.99 }];
},
}),
},
});Multiple Model Providers
Use any AI SDK model:
import { openai } from "@ai-sdk/openai";
import { anthropic } from "@ai-sdk/anthropic";
import { google } from "@ai-sdk/google";
// OpenAI
const agent1 = createAriaCFChatAgent({
model: openai("gpt-4o-mini"),
flows: myFlows,
});
// Anthropic
const agent2 = createAriaCFChatAgent({
model: anthropic("claude-sonnet-4-20250514"),
flows: myFlows,
});
// Google
const agent3 = createAriaCFChatAgent({
model: google("gemini-2.0-flash"),
flows: myFlows,
});
// Workers AI (Cloudflare)
const agent4 = createAriaCFChatAgent({
model: env.AI("@cf/meta/llama-3.1-8b-instruct"),
flows: myFlows,
});API Reference
createAriaCFChatAgent
function createAriaCFChatAgent<Env>(
config: AriaCFChatAgentConfig
): new (ctx: AgentContext, env: Env) => AriaCFChatAgent<Env>;AriaCFChatAgentConfig
interface AriaCFChatAgentConfig {
// Required
flows: Map<string, FlowDefinition>;
model: LanguageModel; // Any AI SDK model
// Optional
instructions?: string; // Agent instructions
guidelines?: string[]; // System guidelines
tools?: ToolRegistry; // Custom tools
historyLimit?: number; // Max history messages (default: 50)
enablePersistence?: boolean; // Enable session persistence (default: true)
}AriaCFChatAgent
Base class extended by the factory:
abstract class AriaCFChatAgent<Env = any> extends AIChatAgent<Env, AriaCFState> {
// Configuration
protected readonly config: AriaCFChatAgentConfigResolved;
protected readonly ariaAgent: AriaAgent;
protected readonly voltRuntime: VoltRuntime;
// Override these in your subclass
protected abstract getConfig(): AriaCFChatAgentConfig;
// Optional overrides
protected getModel(): LanguageModel;
protected getModelId(): string;
protected getTools(): ToolRegistry;
}Connecting from Client
Using React (Chat)
import { useAgentChat } from "@cloudflare/ai-chat/react";
function Chat() {
const { messages, input, handleInputChange, handleSubmit } = useAgentChat({
agent: "my-assistant",
});
return (
<div>
{messages.map((m) => (
<div key={m.id}>{m.content}</div>
))}
<form onSubmit={handleSubmit}>
<input value={input} onChange={handleInputChange} />
<button type="submit">Send</button>
</form>
</div>
);
}Using WebSocket
const ws = new WebSocket("wss://your-worker.workers.dev/agents/my-assistant/session-123");
ws.onopen = () => {
ws.send(JSON.stringify({
type: "cf.agent.chat.request",
init: {
method: "POST",
body: JSON.stringify({
messages: [{ role: "user", content: "Hello!" }]
})
}
}));
};
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log("Response:", data);
};Using HTTP
const response = await fetch("https://your-worker.workers.dev/agents/my-assistant/session-123", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
messages: [{ role: "user", content: "Hello!" }]
})
});Complete Example
See examples/ShoppingAgent.ts for a full shopping assistant example with flows, tools, and persistence.
License
MIT