Writing an Adapter

Learn how to implement a custom Agent adapter so @threadplane/chat components work with any backend — a custom RPC service, an in-process mock, or an exotic streaming protocol.

#When to Write Your Own Adapter

@threadplane/langgraph covers LangGraph backends and @threadplane/ag-ui covers any AG-UI-compatible backend. Everything else needs a hand-rolled adapter. Common scenarios:

• Custom RPC or HTTP API — your backend speaks neither LangGraph Server nor the AG-UI protocol.
• In-process logic — you want the chat UI without any network call (demos, playgrounds, offline-first apps).
• Testing — a deterministic in-process adapter is faster and more reliable than hitting a real agent in unit tests.
• Exotic transports — WebSockets, gRPC-Web, or any other streaming mechanism.

#The Contract

Every @threadplane/chat primitive and composition accepts an Agent object. The type lives in @threadplane/chat and is intentionally runtime-neutral — it says nothing about HTTP, LangGraph, or any specific backend.

import type { Agent } from '@threadplane/chat';

An Agent is a set of Angular signals (reactive state) plus an RxJS observable of events, a submit method to send a message or resume an interrupted run, and a stop method to abort the in-flight run.

#Field-by-Field Reference

#Worked Example: An In-Process Echo Adapter

Below is a complete EchoAgent factory — roughly 80 lines — that satisfies the full Agent contract without any network call. It shows the signal pattern clearly and is a solid starting point for your own adapter.

On submit, the factory appends a complete user message and starts an assistant delivery generation. After a short delay it completes that same generation with a successful echo; stopping early completes it as aborted. There are no tool calls, custom events, or interrupts.

import { signal, type Signal } from '@angular/core';
import { EMPTY, type Observable } from 'rxjs';
import type {
  Agent, Message, AgentStatus, ToolCall,
  AgentEvent, AgentSubmitInput, AgentSubmitOptions,
} from '@threadplane/chat';
import {
  completeDelivery,
  staticDelivery,
  streamingDelivery,
} from '@threadplane/chat';
 
export interface EchoAgentOptions {
  /** Delay before the echoed reply appears, in ms. Defaults to 400. */
  delayMs?: number;
}
 
export function createEchoAgent(opts: EchoAgentOptions = {}): Agent {
  const messages = signal<Message[]>([]);
  const status = signal<AgentStatus>('idle');
  const isLoading = signal(false);
  const error = signal<unknown>(null);
  const toolCalls = signal<ToolCall[]>([]);
  const state = signal<Record<string, unknown>>({});
  let pending: ReturnType<typeof setTimeout> | undefined;
  let activeReply: { id: string; generation: string } | undefined;
 
  const submit = async (input: AgentSubmitInput, _opts?: AgentSubmitOptions) => {
    if (input.message === undefined) return;
 
    const text = typeof input.message === 'string'
      ? input.message
      : input.message.map((b) => b.type === 'text' ? b.text : '').join('');
 
    const userId = cryptoRandomId();
    const assistantId = cryptoRandomId();
    const generation = cryptoRandomId();
 
    // Static user input is already complete. The assistant owns a new
    // generation that remains stable until this attempt reaches an outcome.
    messages.update((prev) => [
      ...prev,
      {
        id: userId,
        role: 'user',
        content: text,
        delivery: staticDelivery(userId),
      },
      {
        id: assistantId,
        role: 'assistant',
        content: '',
        delivery: streamingDelivery(generation),
      },
    ]);
    activeReply = { id: assistantId, generation };
 
    status.set('running');
    isLoading.set(true);
    error.set(null);
 
    pending = setTimeout(() => {
      messages.update((prev) => prev.map((message) =>
        message.id === assistantId
          ? {
              ...message,
              content: `You said: ${text}`,
              delivery: completeDelivery(generation, 'success'),
            }
          : message
      ));
      status.set('idle');
      isLoading.set(false);
      pending = undefined;
      activeReply = undefined;
    }, opts.delayMs ?? 400);
  };
 
  const stop = async () => {
    if (pending !== undefined) clearTimeout(pending);
    if (activeReply) {
      const { id, generation } = activeReply;
      messages.update((prev) => prev.map((message) =>
        message.id === id
          ? { ...message, delivery: completeDelivery(generation, 'aborted') }
          : message
      ));
    }
    pending = undefined;
    activeReply = undefined;
    status.set('idle');
    isLoading.set(false);
  };
 
  const regenerate = async (assistantMessageIndex: number) => {
    // Discard the assistant message at the index AND everything after it,
    // keeping the preceding user message, then re-run against the trimmed tail.
    const prior = messages()[assistantMessageIndex - 1];
    messages.update((prev) => prev.slice(0, assistantMessageIndex - 1));
    if (prior?.role === 'user') {
      await submit({ message: prior.content });
    }
  };
 
  return {
    messages,
    status,
    isLoading,
    error,
    toolCalls,
    state,
    events$: EMPTY satisfies Observable<AgentEvent>,
    submit,
    stop,
    regenerate,
  };
}
 
function cryptoRandomId(): string {
  return Math.random().toString(36).slice(2);
}

#Wiring It Into a Component

For me, the cleanest approach is to register your factory behind an Angular injection token and inject it into your component. It's a touch more boilerplate than a bare new, but it keeps the wiring testable and swappable.

// app.config.ts
import { ApplicationConfig, InjectionToken } from '@angular/core';
import type { Agent } from '@threadplane/chat';
import { createEchoAgent } from './echo-agent';
 
export const ECHO_AGENT = new InjectionToken<Agent>('ECHO_AGENT');
 
export const appConfig: ApplicationConfig = {
  providers: [
    { provide: ECHO_AGENT, useFactory: () => createEchoAgent({ delayMs: 250 }) },
  ],
};

// app.ts
import { Component, inject } from '@angular/core';
import { ChatComponent } from '@threadplane/chat';
import { ECHO_AGENT } from './app.config';
 
@Component({
  selector: 'app-root',
  standalone: true,
  imports: [ChatComponent],
  template: `<chat [agent]="agent" />`,
})
export class App {
  protected readonly agent = inject(ECHO_AGENT);
}

#Validating with the Conformance Suite

@threadplane/chat ships a conformance helper that checks every contract field and a handful of semantic invariants (for example, isLoading() must only be true when status() === 'running'). Run it against your factory in a Vitest spec:

// echo-agent.conformance.spec.ts
import { runAgentConformance } from '@threadplane/chat/testing';
import { createEchoAgent } from './echo-agent';
 
runAgentConformance('createEchoAgent', () => createEchoAgent());

The conformance suite verifies:

• Every required signal is present and returns the correct type.
• isLoading() is false when status() is 'idle'.
• events$ is a valid RxJS Observable.
• submit and stop return a Promise.

There is no separate package to install — the testing entry point ships as part of @threadplane/chat.

#AgentWithHistory (Optional)

If your backend supports checkpointing or thread history, extend the basic contract with AgentWithHistory:

import type { AgentWithHistory } from '@threadplane/chat';

AgentWithHistory adds a history: Signal<AgentCheckpoint[]> field. The implementation pattern is identical — add the signal to your factory return value.

Use runAgentWithHistoryConformance from @threadplane/chat/testing in your spec instead of runAgentConformance to cover the additional field.

#Hydrating from a server-stored thread

AgentWithHistory is a structural choice — exposing the checkpoint list. There is a parallel behavioral choice an adapter has to make:

When threadId changes to a non-null id the consumer didn't just create, should messages and values re-populate from the server's record of that thread?

The chat UI assumes "yes" implicitly — clicking a thread in a sidebar list, or rehydrating a saved id on reload, both rely on the adapter pulling the prior conversation back into view. If the adapter only clears local state on a thread switch, the UI shows an empty welcome surface and the user loses their conversation.

How adapters in this repo answer the question:

• @threadplane/langgraph answers yes. On every threadId change the bridge calls transport.getHistory(id) (LangGraph SDK client.threads.getHistory), takes the latest checkpoint, and seeds messages$ and values$ before any new user turn. The LangGraph protocol exposes a checkpoint endpoint per thread, so this is straightforward. See the LangGraph persistence guide for the consumer-side shape.
• @threadplane/ag-ui answers no, by design. The AG-UI protocol is event-stream-only — it doesn't define a server-side "get the messages on this thread" endpoint. The adapter's threadId is therefore a plain string accepted once at construction, and switching threads means tearing down the provider and creating a new one (or pre-loading messages from the host service before the agent boots). See AG-UI architecture › Provider choices.

If your protocol does store thread state, follow the LangGraph adapter's pattern: react to threadId changes, fetch the latest checkpoint, and surface a isThreadLoading signal so the UI can show a skeleton while the fetch runs. If it doesn't, follow AG-UI and be explicit in your docs that consumers own the load step.

#Client Tools (Optional)

If your runtime lets the browser declare tools the model can call, implement the optional clientTools capability. Consumers then get client tools with no further work.

import type { ClientToolsCapability } from '@threadplane/chat';

Use selectPendingClientToolCalls from @threadplane/chat for pending rather than hand-rolling the predicate — a call is pending when the run is not in flight, its name is in the catalog, it has no backend result, and it has not already been settled locally.

#The invariant

The server thread must never hold a client tool call without a corresponding tool result.

Violate it and the thread carries an assistant message with tool_calls and no matching tool message; most providers reject that history outright on the next user turn.

resolve() alone cannot uphold this, because some groups never continue — every tool in them is terminal (followUp: false), the user pressed stop, or a continuation limit tripped. That is what settle + flush are for.

#Implementation rules

Four rules keep settlement, persistence, and recovery aligned:

1. Give every result message a stable ID derived from its tool-call ID. Replay and overlap are idempotent only when the receiving store uses ID-aware upsert or merge semantics. LangGraph's add_messages reducer does; a stable ID alone does not deduplicate an append-only sink.
2. Snapshot without forgetting, and acknowledge only after that exact handoff reports success. Failure, interruption, abort, pause, or supersession leaves the result staged for retry or an ordinary later submit. Overlapping operations may carry the same stable ID only when the receiving path provides the ID-aware merge described above.
3. Chain concurrent flushes instead of short-circuiting them. Results settled after one snapshot need their own write; returning only the older in-flight promise can report success without persisting the newer batch.
4. Discard staged results and retry associations on a thread switch. Advance the staging generation as part of the reset so acknowledgments from old in-flight work become no-ops on the new thread.

Adapters without a durable write path can fall back to attaching staged results to the next outbound run. The results stay buffered and an ordinary next message can still carry them, but a browser page reload first loses that in-memory fallback and leaves the server thread with an unanswered tool call.

#Publishing Your Adapter

Want to distribute your adapter as an npm package? Keep the following in mind.

Peer dependencies to declare in your package.json:

{
  "peerDependencies": {
    "@angular/core": "^20.0.0",
    "@threadplane/chat": "^0.0.2",
    "rxjs": "^7.0.0"
  },
  "devDependencies": {
    "@threadplane/chat": "^0.0.2"
  }
}

The @threadplane/chat/testing entry point is part of the same package as the main entry point, so there is nothing extra to install for the conformance tests.

Naming convention: @your-org/your-backend-agent works well (e.g., @acme/supabase-realtime-agent). The -agent suffix signals that the package satisfies the Agent contract.

Angular library setup: Use Nx (nx g @nx/angular:library) or the Angular CLI (ng g library) to scaffold an Angular library with ng-packagr. Point your package.json exports at the compiled output. See the Nx Angular library guide for the full setup.

Optional: license-key gating. If you want to restrict usage to paying customers, @threadplane/licensing provides a browser-safe license verification API. Declare it as an optional peer dependency.

#What's Next