Library usage

You do not need the CLI, a scaffolded project, or a workflow to use Assemble. An agent is a plain TypeScript value with .run() and .stream() methods, so you can call one from any existing server process — a Next.js route handler, a Supabase Edge Function, a queue worker — the same way you would call any async library. Agent runs never write to disk, nothing is contacted over the network except your model provider, and the platform is never involved unless you attach its sink — see Bring your own observability.

Calling an agent with run#

Define the agent next to your server code and await it:

import { anthropic } from "@assemble-dev/providers";
import { agent } from "@assemble-dev/sdk";
import { z } from "zod";

export const summarizer = agent({
  name: "summarizer",
  role: "Summarizes support conversations",
  model: anthropic({ model: "claude-sonnet-4-5" }),
  instructions: "Summarize the conversation in three sentences.",
  outputSchema: z.object({ summary: z.string() }),
});

import { summarizer } from "@/agents/summarizer";

export async function POST(request: Request): Promise<Response> {
  const { conversation } = await request.json();

  const result = await summarizer.run(conversation);

  return Response.json({ summary: result.output.summary });
}

Zod versions#

Every schema you hand the SDK — agent inputSchema and outputSchema, tool schemas, workflow inputSchema, batch structured parsing — is typed structurally as a ValidationSchema: any object with a compatible safeParse method satisfies it. Zod v3, Zod v4, and hand-written validators all work. The SDK depends on zod as a regular dependency, not a peer, so your application's own Zod version never conflicts with the SDK's — there is nothing to align or dedupe.

Input: a value or a messages array#

run and stream accept either a JsonValue — a string, or any JSON-serializable object — or a messages array for multi-turn chat:

const result = await summarizer.run([
  { role: "system", content: "The customer is on the enterprise plan." },
  { role: "user", content: "My export failed twice." },
  { role: "assistant", content: "Which format were you exporting?" },
  { role: "user", content: "CSV, about 2 GB." },
]);

Each message is { role, content } with a role of "user", "assistant", "system", or "tool". This is how you carry a chat history across requests: store the transcript in your own database or session, append the new user message, and pass the whole array.

System messages and instructions#

{ role: "system" } messages in the array do not replace the agent's instructions — both apply. The runtime joins them into a single system block, instructions first, then your system messages in order.

One caution when prompt caching is on: the cache breakpoint sits on that joined block, so a per-request system message changes the prefix and invalidates the cached system prompt on every call. Keep per-request data — the customer's plan, the current time — in user messages and let the system prompt stay byte-stable, the same rule as prompt caching and dynamic instructions.

Run options#

The second argument to run and stream is an options object:

metadata fields:

const result = await summarizer.run(conversation, {
  abortSignal: request.signal,
  metadata: { sessionId: "sess_42", userId: "user_7", tags: ["support"] },
});

The run result#

run resolves to an AgentRunResult:

Cost budgets#

costBudgetUsd on the agent definition is enforced: when a run's estimated spend exceeds the budget, the run aborts with a budget error instead of continuing to bill. Set it on any agent exposed to end-user input, where a prompt you did not write decides how much work the model does:

const summarizer = agent({
  name: "summarizer",
  role: "Summarizes support conversations",
  model: anthropic({ model: "claude-sonnet-4-5" }),
  instructions: "Summarize the conversation in three sentences.",
  costBudgetUsd: 0.5,
});

No disk writes#

Agent runs never touch the filesystem — nothing under .assemble/ is created, which is what you want in serverless and edge runtimes with read-only or ephemeral disks. Workflow runs persist run state by default for replay and resumption; pass persistRunState: false in the run options to make workflows disk-free too.

Streaming responses#

.stream() starts the run and returns an AgentStream: async iterables of text deltas, partial objects, and tool-call events, plus a promise for the same final result run would give you.

const stream = summarizer.stream(conversation);

for await (const textDelta of stream.textStream) {
  process.stdout.write(textDelta);
}

const result = await stream.result;

The usual server shape is piping textStream into a server-sent events response:

import { summarizer } from "@/agents/summarizer";

export async function POST(request: Request): Promise<Response> {
  const { messages } = await request.json();
  const stream = summarizer.stream(messages, { abortSignal: request.signal });

  const encoder = new TextEncoder();
  const body = new ReadableStream({
    async start(controller) {
      for await (const textDelta of stream.textStream) {
        controller.enqueue(encoder.encode(`data: ${JSON.stringify(textDelta)}\n\n`));
      }
      controller.enqueue(encoder.encode("data: [DONE]\n\n"));
      controller.close();
    },
  });

  return new Response(body, {
    headers: {
      "content-type": "text/event-stream",
      "cache-control": "no-cache",
      connection: "keep-alive",
    },
  });
}

Passing request.signal as the abortSignal means a client that closes the tab cancels the model call instead of paying for tokens nobody will read.

Partial structured output#

When the agent has an outputSchema, the stream also exposes objectStream (experimental): an async iterable of progressively-parsed partial objects assembled from the token stream. Each yielded value is the best-effort parse of the JSON produced so far, so a UI can render fields as they complete instead of waiting for the full object:

const stream = summarizer.stream(conversation);

for await (const partialObject of stream.objectStream) {
  render(partialObject);
}

const { output } = await stream.result;

Partial objects are exactly that — fields may be missing or truncated mid-string until the stream ends. Only stream.result carries the fully validated output; treat objectStream as display-only and its API as subject to change.

Streaming tool calls#

For an agent with tools, the silent stretch while a tool runs is where chat UIs go dead. toolCallStream fills it — an async iterable of tool lifecycle events:

The usual pattern is a status line — show it on tool_execution_started, clear it on tool_execution_completed:

const stream = assistant.stream(messages);

void (async () => {
  for await (const event of stream.toolCallStream) {
    if (event.type === "tool_execution_started") {
      showStatus(`Searching candidates with ${event.toolName}…`);
    }
    if (event.type === "tool_execution_completed") {
      clearStatus();
    }
  }
})();

for await (const textDelta of stream.textStream) {
  appendText(textDelta);
}

const result = await stream.result;

Events for one toolCallId always arrive in order: tool_call_requested, then any tool_call_input_deltas, then tool_execution_started, then tool_execution_completed. Adapters without native streaming synthesize the same tool_call_requested and execution events (minus the input deltas), so UI code written against toolCallStream works unchanged across adapters, including mockModel in tests. Like textStream, the stream closes when the run finishes — drain it concurrently with awaiting stream.result.

Multi-step tool loops#

An agent with tools and maxSteps greater than 1 runs a model-driven loop inside a single run: the model requests a tool (tool_use), the runtime executes it, the result goes back to the model (tool_result), and the model continues — until it produces a final answer or exhausts the step budget.

const assistant = agent({
  name: "assistant",
  role: "Answers order questions using lookup tools",
  model: anthropic({ model: "claude-sonnet-4-5" }),
  instructions: "Look up order data before answering.",
  tools: [lookupOrderTool, lookupShipmentTool],
  maxSteps: 5,
});

Tools attach directly to the agent through tools; allowedToolNames filters them when set. Each tool's inputJsonSchema describes its parameters to the model (a permissive object schema is sent when omitted).

toolChoice in the run options controls the loop:

Every iteration emits the usual tool trace events, and policies apply to each call in the loop exactly as in a workflow.

Repairing failed tool calls#

Models sometimes almost get a tool call right — a snake_cased tool name, a stringified number in the input. repairToolCall on the agent definition gives you one chance to fix such a call before it becomes an error. It fires when the model names an unknown tool or produces input that fails the tool's schema, receives { toolCall, error, attempt }, and returns a corrected ModelToolCall — or null to fall through to the normal error path. One repair attempt is made per failed call, so a repair that fails again is not re-repaired.

const assistant = agent({
  name: "assistant",
  role: "Answers order questions using lookup tools",
  model: anthropic({ model: "claude-sonnet-4-5" }),
  instructions: "Look up order data before answering.",
  tools: [lookupOrderTool],
  maxSteps: 5,
  repairToolCall: async ({ toolCall }) => {
    if (toolCall.toolName === "lookup_order") {
      return { ...toolCall, toolName: "lookupOrder" };
    }

    return null;
  },
});

Stopping the loop early#

stopWhen is a per-step predicate evaluated after each tool-executing step. It receives a step summary — { stepNumber, toolCalls, lastText, stopReason, tokenUsage } — and returning true ends the loop gracefully with the output produced so far, instead of spending the rest of the step budget:

const researcher = agent({
  name: "researcher",
  role: "Searches internal docs",
  model: anthropic({ model: "claude-sonnet-4-5" }),
  instructions: "Search until you find a relevant document.",
  tools: [searchDocsTool],
  maxSteps: 8,
  stopWhen: (step) => step.lastText.includes("FINAL ANSWER"),
});

Use it as a semantic stop condition (the model signaled it is done), a spend guard (step.tokenUsage.totalTokens crossed a threshold), or a result-count cutoff (step.toolCalls already gathered enough) — anywhere maxSteps alone is too blunt.

Multimodal input#

With the Anthropic adapter, message content can be an array of typed blocks instead of a string:

const result = await receiptReader.run([
  {
    role: "user",
    content: [
      { type: "text", text: "Extract the total from this receipt." },
      { type: "image", mediaType: "image/png", dataBase64: receiptPngBase64 },
    ],
  },
]);

Assistant prefill#

A trailing assistant message is passed through to the model, which continues it rather than starting fresh. Use it to force a format or a first token:

const result = await classifier.run([
  { role: "user", content: "Classify: my refund never arrived" },
  { role: "assistant", content: '{"category":' },
]);

Supabase Edge Functions and Deno#

The SDK and providers packages ship as pure ESM with a static import graph, importable in Deno via npm: specifiers — no build step, no dynamic require. A Deno smoke test runs in CI, so this path stays exercised, not aspirational.

import { agent } from "npm:@assemble-dev/sdk";
import { anthropic } from "npm:@assemble-dev/providers";

const summarizer = agent({
  name: "summarizer",
  role: "Summarizes support conversations",
  model: anthropic({ model: "claude-haiku-4-5" }),
  instructions: "Summarize the conversation in three sentences.",
});

Deno.serve(async (request: Request): Promise<Response> => {
  const { conversation } = await request.json();
  const result = await summarizer.run(conversation, {
    abortSignal: request.signal,
  });

  return Response.json({
    summary: result.output,
    costUsd: result.costUsd,
  });
});

Set ANTHROPIC_API_KEY as a function secret and the adapter resolves it exactly as on Node — see providers. Because agent runs never write to disk, nothing about the edge runtime's ephemeral filesystem matters here.

The console sink defaults to process.stdout, which does not exist on every runtime — pass a custom writeLine where needed:

import { createConsoleSink } from "npm:@assemble-dev/sdk";

const consoleSink = createConsoleSink({
  writeLine: (line: string) => console.log(line),
});

For forwarding trace events to your own collector instead of the console, implement a sink — see Bring your own observability.

Foundation types from the SDK root#

The trace and validation foundation types are re-exported from the @assemble-dev/sdk root, so sinks, collectors, and tooling never need subpath imports from @assemble-dev/shared-types — which also keeps a Deno import map to a single entry:

import {
  JsonValueSchema,
  TokenUsageSchema,
  TraceEventSchema,
  traceEventTypes,
} from "@assemble-dev/sdk";
import type {
  JsonValue,
  SchemaValidationResult,
  TokenUsage,
  TraceEvent,
  TraceEventType,
  ValidationSchema,
} from "@assemble-dev/sdk";

TraceEvent (with its Zod schema, the TraceEventType union, and the traceEventTypes list), TokenUsage, and JsonValue cover everything a custom sink touches; ValidationSchema and SchemaValidationResult are the structural schema types behind Zod versions, for code that accepts schemas generically.