Mastra Overview

Mastra is a TypeScript agent framework. It is the only non-Python runtime in the measured set, and the only one whose upstream AG-UI integration ships no HTTP endpoint at all, so the backend here is a small hosting service written by hand. The running example is a camping trip planner, and this page walks the four files the Code tab carries: two backend files under deployments/ag-ui-mastra, an application config, and a component. Quickstart runs the same example locally.

What the demo does

The Run tab shows the prebuilt <chat> composition beside a panel that mirrors the agent's packing list. Three welcome suggestions set it up: "Start a packing list" seeds a titled list with a tent and two sleeping bags and the panel fills in as the agent writes it, "Check trail conditions" makes the agent call a backend tool, and "Reserve the campsite" asks for two nights at North Pines.

The reservation is the interesting one. It suspends the run rather than booking anything, and an approval card appears with the campsite, the number of nights, and the total. Approve resumes the agent with a confirmation number, and Cancel resumes it with a refusal.

Ask for a weather forecast and the agent delegates to a second agent instead of answering itself. The child's answer streams into a subagent card in the transcript while it is still being written.

How it is built

Four files carry the example. Two are the backend: agents.mjs defines the Mastra agents, and server.mjs serves them over HTTP. The other two are the Angular side: a config that registers the agent, and a component that adds the state panel and the approval card around <chat>. Open the Code tab to read them in place.

A backend tool

createTool describes a tool with Zod schemas for its input and output and an execute body that runs on the server. check_conditions returns a fixed forecast, so the demo behaves the same on every run.

agents.mjs — the backend tool

/** Deterministic backend tool — no external calls, stable for fixtures. */
const checkConditionsTool = createTool({
  id: 'check_conditions',
  description: 'Check current trail and weather conditions for a location.',
  inputSchema: z.object({ location: z.string().describe('Park or trailhead name') }),
  outputSchema: z.object({
    location: z.string(),
    forecast: z.string(),
    high_c: z.number(),
    low_c: z.number(),
  }),
  execute: async (inputData) => ({
    location: inputData.location,
    forecast: 'Clear skies, light afternoon breeze',
    high_c: 18,
    low_c: 4,
  }),
});

Nothing in this tool is protocol-aware; the bridge turns the call into TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END, and TOOL_CALL_RESULT on the wire.

Pausing a run for approval

A tool becomes a human-in-the-loop step by adding a suspendSchema and a resumeSchema to that same shape. The first call arrives with no resume data, so the body calls suspend() with the payload the approval card renders. The resumed call arrives with the operator's decision and either books the site or reports the refusal.

agents.mjs — the tool that suspends

/**
 * Human-in-the-loop tool. First call suspends the run (persisted to LibSQL —
 * suspend/resume REQUIRES persistent storage, a spike finding); the frontend
 * shows an approval card and resumes with `{ approved: boolean }`.
 */
const reserveCampsiteTool = createTool({
  id: 'reserve_campsite',
  description:
    'Reserve a campsite. Requires explicit user approval: the tool pauses the run and shows the user a confirmation card before booking.',
  inputSchema: z.object({
    site: z.string().describe('Campsite name'),
    nights: z.number().int().min(1).describe('Number of nights'),
  }),
  suspendSchema: z.object({
    site: z.string(),
    nights: z.number(),
    total_usd: z.number(),
  }),
  resumeSchema: z.object({
    approved: z.boolean().optional(),
  }),
  execute: async (inputData, context) => {
    const { resumeData, suspend } = context?.agent ?? {};
    if (!resumeData) {
      return suspend?.({
        site: inputData.site,
        nights: inputData.nights,
        total_usd: inputData.nights * NIGHTLY_RATE_USD,
      });
    }
    if (resumeData.approved) {
      return `Reserved ${inputData.site} for ${inputData.nights} night(s) — total $${
        inputData.nights * NIGHTLY_RATE_USD
      }. Confirmation TP-${String(inputData.nights).padStart(2, '0')}88.`;
    }
    return `Reservation for ${inputData.site} was declined by the user. Nothing was booked.`;
  },
});

The suspended run is written to storage, which is why the database path further down has to be durable.

The child agent

A sub-agent is an ordinary Mastra Agent with its own instructions. Its description is what the parent model reads when it decides whether to delegate.

agents.mjs — the child agent

/**
 * Sub-agent (spike: wire-capture-subagents.md). Registered on the
 * supervisor via `agents:`; Mastra surfaces it as a backend tool named
 * `agent-weather_forecaster` whose TOOL_CALL_RESULT carries the child's
 * final text — server.mjs's subagent emitter turns that into SUBAGENT_*
 * frames. The `description` becomes the delegation tool's description.
 */
const weatherForecaster = new Agent({
  id: 'weather_forecaster',
  name: 'weather_forecaster',
  description: 'Forecasts weather for a campsite and date range. Use for any weather question.',
  instructions:
    'You are a weather forecaster. Given a campsite and dates, give a 3-bullet forecast summary. Be concise.',
  model: MODEL,
});

The trip agent and its working memory

The parent agent collects everything: the instructions, the two tools, the child agent under agents, and a Memory whose working memory is a typed Zod schema. That schema is the shared state the frontend reads. Registering the child under agents is the entire delegation wiring in the agent definition.

agents.mjs — the trip agent

  const tripAgent = new Agent({
    id: 'mastra',
    name: 'mastra',
    instructions: `You are a terse camping trip planner.
The packing list in working memory is the user's shared state: whenever the user adds, removes, or changes items (or starts a list), update working memory to match. 'items' is an array of {name, qty}. Never mention memory or the list mechanics.
For questions about trail conditions you MUST call check_conditions.
For questions about weather forecasts you MUST delegate to the weather_forecaster agent.
When the user asks to reserve or book a campsite you MUST call reserve_campsite; after it resumes, confirm the outcome.
Always answer in one short sentence.`,
    model: MODEL,
    tools: {
      check_conditions: checkConditionsTool,
      reserve_campsite: reserveCampsiteTool,
    },
    agents: { weather_forecaster: weatherForecaster },
    memory: new Memory({
      storage: store('mastra-topic-memory'),
      options: {
        workingMemory: {
          enabled: true,
          schema: z.object({
            packing_list: z.object({
              title: z.string().describe('Packing list title'),
              items: z
                .array(z.object({ name: z.string(), qty: z.number() }))
                .describe('All items on the list'),
            }),
          }),
        },
      },
    }),
  });

MODEL is the plain string openai/gpt-4o-mini, which Mastra's model router resolves from the standard OPENAI_API_KEY and OPENAI_BASE_URL variables with no provider SDK wiring.

Note: Working memory is the state channel

Mastra has no separate state object to publish. The packing list is working memory. The bridge streams it as a STATE_SNAPSHOT plus real JSON-Patch STATE_DELTA events while the agent edits the list, and the adapter applies those patches to the state signal the component reads.

The route contract

server.mjs is a plain node:http server. Its contract mirrors the Python lane's: GET /ok is unauthenticated for health checks, every other route requires the X-Internal-Token header, and topics are served at POST /agent/<topic>, resolved against the Mastra instance by name.

server.mjs — the route contract

if (req.method === 'GET' && path === '/ok') {
  json(res, 200, { ok: true });
  return;
}
 
if (req.headers['x-internal-token'] !== AG_UI_INTERNAL_TOKEN) {
  json(res, 401, { detail: 'unauthorized' });
  return;
}
 
const match = /^\/agent\/([a-z0-9_-]+)$/.exec(path);
if (!match || req.method !== 'POST') {
  json(res, 404, { detail: 'not found' });
  return;
}
 
const topic = match[1];
let agent;
try {
  agent = mastra.getAgent(topic);
} catch {
  agent = undefined;
}
if (!agent) {
  json(res, 404, { detail: `no such topic: ${topic}` });
  return;
}

The service refuses to boot at all without AG_UI_INTERNAL_TOKEN, rather than serve an unauthenticated model proxy.

One event, one Server-Sent Events frame

The whole encoding is a single function. An AG-UI event becomes one data: frame.

server.mjs — the frame encoder

/** One SSE frame per AG-UI event — the wire shape @ag-ui/client parses. */
function sseFrame(event) {
  return `data: ${JSON.stringify(event)}\n\n`;
}

That is the wire shape @ag-ui/client parses, and the same shape the FastAPI bridges emit for the Python runtimes.

Running the bridge

Each request builds a fresh MastraAgent, because the bridge carries per-run state, and scopes Mastra memory to the conversation by keying resourceId on the inbound thread identifier. bridge.run(input) returns an Observable of AG-UI events, and every event it emits is written as a frame. An Observable error becomes a RUN_ERROR frame rather than a dropped socket, so the client finalizes the run instead of hanging.

A per-run createSubagentInjector sits in front of the frame writer with two methods: injector.chunk reads the raw stream-tee chunks and emits the eager subagent events, and injector.eventsFor reads the bridge's own AG-UI events and drops its later, redundant delegation copies.

server.mjs — running the bridge

// One injector per run with two inputs, both writing through the same
// SSE frame writer:
// - `chunk()`: raw Mastra fullStream chunks observed through the stream
//   tee BEFORE the bridge processes them — the eager TOOL_CALL_* +
//   SUBAGENT_STARTED on the delegation `tool-call`, the child's
//   attributed TEXT_MESSAGE_* deltas from `tool-output`, and
//   SUBAGENT_FINISHED/ERROR on `tool-result`.
// - `eventsFor()`: the bridge's own AG-UI events, with its later buffered
//   TOOL_CALL_START/ARGS/END copies for a synthesized id dropped.
const injector = createSubagentInjector(bridgeCapability);
const write = (event) => res.write(sseFrame(event));
const observe = (chunk) => {
  for (const e of injector.chunk(chunk)) write(e);
};
 
// A fresh bridge per request: MastraAgent carries per-run state.
// resourceId scopes Mastra memory (threads live under a resource);
// keying it by AG-UI threadId gives per-conversation memory. The bridge
// receives the teed agent; it is otherwise unmodified.
const bridge = new MastraAgent({
  agentId: topic,
  agent: withDelegationTee(agent, observe),
  resourceId: input.threadId,
});
 
const sub = bridge.run(input).subscribe({
  next: (event) => {
    for (const e of injector.eventsFor(event)) write(e);
  },
  error: (err) => {
    // Map failures into the protocol instead of killing the socket:
    // the client finalizes the run as an error rather than hanging.
    const runError = { type: 'RUN_ERROR', message: String(err?.message ?? err) };
    for (const e of injector.eventsFor(runError)) write(e);
    res.end();
  },
  complete: () => {
    res.end();
  },
});
 
req.on('close', () => sub.unsubscribe());

The agent handed to the bridge is wrapped first, which is what makes the subagent card stream live; the section below explains why that wrapper exists.

The agent provider

provideAgent() registers the agent once for the whole application, and it is the only provider the <chat> composition requires. Nothing about it is Mastra-specific: this is the same config every AG-UI example uses. The example passes a factory because it resolves its endpoint at runtime from the host that serves the demo.

app.config.ts

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/ag-ui';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(() => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'ag-ui') {
        throw new Error('incompatible runtime');
      }
      return {
        url: connection.url,
        interruptTransport: 'mastra-command',
      };
    }),
  ],
};

Your own application passes the URL directly:

provideAgent({
  url: 'https://your-backend.example.com/agent',
});

Reading shared state in the component

The component reads the packing list off agent.state(), the runtime-neutral signal the adapter maintains from the state events. Working memory arrives under its schema key, so the component narrows that key and treats a list without a title as absent.

mastra.component.ts — reading shared state

/** Mastra working memory, bridged into AG-UI shared state. */
protected readonly packingList = computed(() => {
  // Example apps compile lib source with strict:false — cast at the read site.
  const state = this.agent.state() as { packing_list?: PackingList } | undefined;
  const list = state?.packing_list;
  return list && list.title ? list : undefined;
});

The panel then renders that signal like any other computed value, with an empty state before the first snapshot.

mastra.component.ts — the state panel

<aside class="state-panel" data-testid="packing-state">
  <h2 class="state-title">Shared state — packing list</h2>
  @if (packingList(); as list) {
    <h3 class="list-title">{{ list.title }}</h3>
    <ul class="list-items">
      @for (item of list.items ?? []; track item.name) {
        <li>
          <span class="item-name">{{ item.name }}</span>
          <span class="item-qty">×{{ item.qty }}</span>
        </li>
      }
    </ul>
  } @else {
    <p class="state-empty">No list yet — Mastra working memory streams here as STATE_SNAPSHOT + STATE_DELTA patches.</p>
  }
</aside>

The approval card

<chat-approval-card> renders whenever the agent has a pending interrupt and reports the operator's choice through its action output. The example projects a body template into it so the card shows the campsite, the nights, and the total rather than a generic prompt.

mastra.component.ts — the approval card

<chat-approval-card
  [agent]="agent"
  title="Reservation approval required"
  (action)="onAction($event)"
>
  <ng-template #body>
    <div class="approval-body">
      @if (suspendPayload(); as s) {
        <div class="approval-row">
          <span class="approval-label">Campsite</span>
          <strong>{{ s.site }}</strong>
        </div>
        <div class="approval-row">
          <span class="approval-label">Nights</span>
          <strong>{{ s.nights }}</strong>
        </div>
        <div class="approval-row">
          <span class="approval-label">Total</span>
          <strong>{{ s.total_usd | currency }}</strong>
        </div>
      } @else {
        <p>The agent requests approval for <code class="approval-code">{{ approvalToolName() }}</code>.</p>
      }
    </div>
  </ng-template>
</chat-approval-card>

The class supplies that payload and maps the two actions onto resume calls.

mastra.component.ts — reading the interrupt and resuming

/**
 * The pending Mastra suspend. The session retains the parsed CUSTOM
 * `on_interrupt` payload: `{ type: 'mastra_suspend', toolCallId, toolName,
 * suspendPayload, args, resumeSchema, runId }` alongside the native
 * RUN_FINISHED outcome that follows it on the wire.
 */
private readonly suspendValue = computed(() => {
  return this.agent.interruptSession().legacy?.value as
    | { toolName?: string; suspendPayload?: { site?: string; nights?: number; total_usd?: number } }
    | undefined;
});
 
protected readonly approvalToolName = computed(() => this.suspendValue()?.toolName ?? 'a tool call');
 
protected readonly suspendPayload = computed(() => this.suspendValue()?.suspendPayload);
 
protected send(text: string): void {
  void this.agent.submit({ message: text });
}
 
protected onAction(action: ChatApprovalAction): void {
  if (action === 'approve') {
    // The adapter turns this into forwardedProps.command
    // { resume: { approved: true }, interruptEvent: { toolCallId, runId } }.
    void this.agent.submit({ resume: { approved: true } });
  } else if (action === 'cancel') {
    void this.agent.submit({ resume: { approved: false } });
  }
}

agent.interruptSession().legacy?.value carries the parsed CUSTOM on_interrupt payload alongside the native interrupt batch. The provider selects interruptTransport: 'mastra-command', so submit({ resume }) goes back out as forwardedProps.command = { resume, interruptEvent: { toolCallId, runId } } — the shape the Mastra bridge reads to reopen the suspended run.

What the integration demonstrates

Surface	Status	How
Messages	Supported	Streamed assistant text over `TEXT_MESSAGE_CHUNK`.
Tool calls	Supported	`check_conditions` executes server-side with no pause.
Shared state	Supported	A working-memory packing list, over snapshots and real JSON-Patch deltas.
Interrupts	Supported	`reserve_campsite` suspends the run and resumes from a persisted snapshot.
Subagents	Supported (streaming)	An in-tree stream tee observes the parent stream ahead of the bridge and emits the delegation tool call eagerly, `SUBAGENT_*` lifecycle, and the child's token deltas under the subagent identity.

Upstream ships no HTTP endpoint

@ag-ui/mastra provides an in-process MastraAgent bridge and a mount for its own chat frontend runtime. It does not provide a plain AG-UI HTTP endpoint, which is what @threadplane/ag-ui connects to.

Threadplane therefore maintains the hosting service walked above, deployments/ag-ui-mastra. It is deliberately hand-written rather than generated: the Python generator targets one aggregated FastAPI process, and Mastra is a different language on a different hosting lane.

This is the honest shape of the Mastra integration. The adapter needed no changes, but somebody has to serve the events, and upstream does not.

Persistence is a hard requirement

Mastra persists memory and suspended-run snapshots to LibSQL file storage, which is the dbUrl passed into createMastra. Resume loads the suspended snapshot back, so the database path has to survive between HTTP requests and across restarts. An in-memory store breaks resume, and an ephemeral filesystem orphans every pending interrupt on redeploy.

How subagents surface

Mastra registers a child agent as a delegation tool named agent-<childKey>, and while the child runs its every chunk is forwarded on the parent stream as a public tool-output chunk. The runtime's bridge drops those chunks and withholds the delegation tool call until the child resolves, so on its own the wire would carry only the child's final text, after a silent gap.

The hosting service therefore wraps the agent in a small stream tee, streaming-tee.mjs, that observes each chunk before the bridge processes it, and a per-run injector, subagent-emitter.mjs, maps them to the protocol: the delegation tool-call chunk becomes an eager TOOL_CALL_START, TOOL_CALL_ARGS, and TOOL_CALL_END plus SUBAGENT_STARTED; each child text delta becomes a TEXT_MESSAGE_CONTENT attributed to the subagent; and the delegation result becomes SUBAGENT_FINISHED or SUBAGENT_ERROR. The bridge's own copy of the delegation tool call, flushed at the result, is dropped so the wire carries it once. Both files sit next to server.mjs rather than in the Code tab.

The bridge itself is unmodified; the tee touches only the public agent members the bridge reads. The child's tokens reach the card while it is still running, which is what flips this cell to Supported. The component contains no subagent code at all, because the card is rendered from the projection the adapter maintains.

How the Mastra row was measured

Its cells come from a real Mastra server driven with live model calls. The raw Server-Sent Events were captured off the wire and are committed and replayed like every other runtime's, including the interrupt and resume round trip driven through the real @ag-ui/client. How it connects records the resulting wire conventions.

The subagent capture behind the Subagents row is committed beside the example as angular/docs/wire-capture-subagents.md.