Skip to content

Build your own harness

Last updated View as MarkdownAgent setup

Write your own harness when you need full control over the model call, the message format, the tool loop, or the protocol your clients speak. A harness is a Lifecycle capability like any other: it gets the Durable Object's SQLite storage and a durable job queue, and the job queue wakes it after an eviction so it can finish its work.

What you build on depends on where the agent loop runs:

  • Inside the Durable Object. Combine the Sessions, Streams, and Tasks capabilities. Sessions stores the conversation, Streams stores model output as it arrives so clients can reconnect and resume, and Tasks runs each turn as replayable steps.
  • Outside the Durable Object, such as in a container or a remote process. Keep the agent's durable record in HarnessStore, and let the agent speak its own format.

Run the loop inside the Durable Object

This sketch stores the user's message, runs the turn as a task, and streams the reply to the client over Server-Sent Events (SSE). Replace generateReply() with your model call.

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { Sessions } from "agents/sessions";
import { sseResponse, Streams } from "agents/streams";
import { Tasks } from "agents/tasks";

// Replace with your model call. It yields text deltas.

export class MyHarness extends DurableObject {
	sessions = new Sessions();
	streams = new Streams();
	tasks = new Tasks({
		definitions: {
			"turn@v1": async ({ streamId }, step) => {
				const text = await step.do("generate", async () => {
					const history = await this.session.getHistory();
					const stream = await this.streams.open(streamId);
					let reply = "";
					for await (const delta of generateReply(history)) {
						reply += delta;
						stream.append({ delta });
					}
					stream.close();
					return reply;
				});
				await step.do("save", async () => {
					await this.session.appendMessage({
						id: streamId,
						role: "assistant",
						parts: [{ type: "text", text }],
					});
				});
			},
		},
	});
	lifecycle = Lifecycle.install(this)
		.use(this.sessions)
		.use(this.streams)
		.use(this.tasks);
	session = this.sessions.session();

	async onRequest(request) {
		const url = new URL(request.url);
		if (request.method === "POST") {
			await this.session.appendMessage({
				id: crypto.randomUUID(),
				role: "user",
				parts: [{ type: "text", text: await request.text() }],
			});
			const streamId = crypto.randomUUID();
			await this.tasks.run("turn@v1", { streamId });
			return Response.json({ streamId }, { status: 202 });
		}
		const streamId = url.searchParams.get("stream");
		if (!streamId) return new Response("Missing stream", { status: 400 });
		return sseResponse(this.streams, streamId, { request });
	}
}
import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { Sessions, type SessionMessage } from "agents/sessions";
import { sseResponse, Streams } from "agents/streams";
import { Tasks, type TaskStep } from "agents/tasks";

// Replace with your model call. It yields text deltas.
declare function generateReply(
	history: SessionMessage[],
): AsyncIterable<string>;

type TurnInput = { streamId: string };

export class MyHarness extends DurableObject<Env> {
	readonly sessions = new Sessions();
	readonly streams = new Streams();
	readonly tasks = new Tasks({
		definitions: {
			"turn@v1": async ({ streamId }: TurnInput, step: TaskStep) => {
				const text = await step.do("generate", async () => {
					const history = await this.session.getHistory();
					const stream = await this.streams.open(streamId);
					let reply = "";
					for await (const delta of generateReply(history)) {
						reply += delta;
						stream.append({ delta });
					}
					stream.close();
					return reply;
				});
				await step.do("save", async () => {
					await this.session.appendMessage({
						id: streamId,
						role: "assistant",
						parts: [{ type: "text", text }],
					});
				});
			},
		},
	});
	readonly lifecycle = Lifecycle.install(this)
		.use(this.sessions)
		.use(this.streams)
		.use(this.tasks);
	readonly session = this.sessions.session();

	async onRequest(request: Request): Promise<Response> {
		const url = new URL(request.url);
		if (request.method === "POST") {
			await this.session.appendMessage({
				id: crypto.randomUUID(),
				role: "user",
				parts: [{ type: "text", text: await request.text() }],
			});
			const streamId = crypto.randomUUID();
			await this.tasks.run("turn@v1", { streamId });
			return Response.json({ streamId }, { status: 202 });
		}
		const streamId = url.searchParams.get("stream");
		if (!streamId) return new Response("Missing stream", { status: 400 });
		return sseResponse(this.streams, streamId, { request });
	}
}

How the pieces work together:

  • Sessions keeps the transcript in the object's SQLite database. getHistory() reads the active path, and appendMessage() writes a message.
  • Tasks journals each step.do(). After an eviction, a finished step returns its stored result, and the run continues from the first unfinished step. A model call cannot resume halfway, so an interrupted generate step runs again from the start. Check step.interrupted in the handler to decide what to do with a partial reply.
  • Streams stores each delta as it is appended. sseResponse() replays what a client missed, then follows the live stream, so a client that reconnects mid-turn catches up.

To turn the sketch into a reusable harness, move the loop into a class that extends LifecycleCapability and takes the three capabilities as constructor options. The host object installs all four on its Lifecycle. Refer to Build a capability.

For a complete harness built this way, refer to the Codex harness. It runs a turn loop derived from Codex, compiled to WebAssembly, as a capability on a plain Durable Object. The transcript lives in Sessions, operation events go to Streams, model and tool steps run as Tasks, files live in a Cloudflare Computer workspace, and the WebSockets capability serves the events to the browser.

The Agents SDK also includes a small experimental AI SDK harness, AiSdkHarness from agents/harness/ai-sdk. It runs one AI SDK streamText call per operation and keeps its sessions in the Durable Object's storage. It is the smallest complete harness in the SDK, and a useful reference when you write your own.

Keep the record of an agent that runs elsewhere

When the agent loop runs outside the Durable Object, the object still needs a durable record: which sessions exist, what was submitted, and what the agent produced. HarnessStore from agents/harness/store keeps that record in the object's SQLite database. It is a helper over ctx.storage.sql, not a capability, so you open it on the storage and do not install it on the Lifecycle.

It stores three things:

  • Sessions, each with an optional parent and a JSON state that your harness owns.
  • Operations, the submissions to a session. Each has an idempotency id that is unique within its session, an input, and a status: queued, running, done, or unanswered.
  • Logs, named lists of JSON entries per session, such as messages or events. An entry with an id replaces the earlier entry with the same id, in place.

The store does not know your agent's format. Every input, result, and log entry is plain JSON.

import { DurableObject } from "cloudflare:workers";
import { openHarnessStore } from "agents/harness/store";

export class RemoteAgent extends DurableObject {
	store = openHarnessStore(this.ctx.storage, { prefix: "my_agent_" });

	submit(session, operationId, text) {
		this.store.createSession({ id: session });
		const { record, accepted } = this.store.enqueue({
			session,
			id: operationId,
			input: { text },
		});
		return { operationId: record.id, accepted };
	}

	// Called by the process that runs the agent, as it makes progress.
	report(session, operationId, messages) {
		this.store.start(session, operationId);
		this.store.append(
			session,
			"messages",
			messages.map((message) => ({ id: message.id, data: message })),
		);
	}

	finish(session, operationId, text) {
		this.store.settle(session, operationId, {
			status: "done",
			result: { text },
		});
	}

	transcript(session) {
		return this.store.read(session, "messages").map((entry) => entry.data);
	}
}
import { DurableObject } from "cloudflare:workers";
import { openHarnessStore } from "agents/harness/store";

export class RemoteAgent extends DurableObject<Env> {
	readonly store = openHarnessStore(this.ctx.storage, { prefix: "my_agent_" });

	submit(session: string, operationId: string, text: string) {
		this.store.createSession({ id: session });
		const { record, accepted } = this.store.enqueue({
			session,
			id: operationId,
			input: { text },
		});
		return { operationId: record.id, accepted };
	}

	// Called by the process that runs the agent, as it makes progress.
	report(
		session: string,
		operationId: string,
		messages: { id: string; text: string }[],
	) {
		this.store.start(session, operationId);
		this.store.append(
			session,
			"messages",
			messages.map((message) => ({ id: message.id, data: message })),
		);
	}

	finish(session: string, operationId: string, text: string) {
		this.store.settle(session, operationId, {
			status: "done",
			result: { text },
		});
	}

	transcript(session: string) {
		return this.store.read(session, "messages").map((entry) => entry.data);
	}
}

The store's tables share one prefix, harness_ by default. Give each harness in the same object its own prefix. A prefix cannot start with _cf_, which Durable Objects reserve.

Pair the store with the Lifecycle's job queue to wake the object while the remote agent works, and with the WebSockets capability to serve the logs to clients.

Was this helpful?