Integrations

Adapters turn what your framework already emits into stream events. Each one is a separate entry point with no dependencies, so it doesn't pull in the framework it adapts.

Entries are paths under @dan-build/agentstat, for example @dan-build/agentstat/ai-sdk.

How adapters work

Every adapter takes a destination and, where it applies, an agent id (for LangChain it's an option; for OpenTelemetry it's optional). The destination is a stream from createAgentStream(), or anything else with the same push(agentId, event) method.

Adapters run in the browser and on the server. When your agents run on a server, push into a server-side stream and send it to the browser over SSE.

Vercel AI SDK

useChat

Create a chat tracker and call it whenever messages or status change:

'use client';
import { useEffect, useMemo } from 'react';
import { useChat } from '@ai-sdk/react';
import { AgentStat, useAgentStream } from '@dan-build/agentstat';
import { createChatTracker } from '@dan-build/agentstat/ai-sdk';

export default function MonitoredChat() {
  const stream = useAgentStream();
  const track = useMemo(() => createChatTracker(stream, 'chat'), [stream]);
  const { messages, status, sendMessage } = useChat();
  useEffect(() => track(messages, status), [track, messages, status]);

  return (
    <div>
      <AgentStat source={stream} metric="tokens" anomalyDetection height={320} />
      <button type="button" onClick={() => sendMessage({ text: 'Write a haiku about telemetry.' })}>
        Send
      </button>
    </div>
  );
}

streamText

Feed its full stream to tapAiSdkStream. It resolves when the stream ends:

import { streamText } from 'ai';
import { openai } from '@ai-sdk/openai';
import { tapAiSdkStream } from '@dan-build/agentstat/ai-sdk';

const result = streamText({ model: openai('gpt-4o-mini'), prompt: 'Summarize the incident report.' });
await tapAiSdkStream(stream, 'writer', result.fullStream);

To handle parts yourself, createAiSdkTap(stream, 'writer') returns a function to call with each part. Push { type: 'finish' } yourself if the stream can end without a finish part.

What it reports

• Output: text and reasoning count as output; with the stream taps, so do tool-call arguments.
• Tool calls are waiting until their result arrives.
• After a request ends, the chat tracker keeps the agent waiting only on a call that awaits approval, or on tools your frontend runs, which you list in { clientTools: ['askUser'] }. Any other open call ends with the request (the user pressed Stop).
• History: messages present on the tracker's first call, or changed while no request runs (restored history, setMessages), aren't reported as output.
• Token counts are estimated at about 4 characters per token. Pass { countTokens } with your tokenizer for exact counts.

LangChain and LangGraph

Pass a callback handler to an invocation; child runs inherit it. Attach it once per invocation:

import { createLangChainCallbacks } from '@dan-build/agentstat/langchain';

await agent.invoke(
  { messages: [{ role: 'user', content: 'Find three sources on this topic.' }] },
  { callbacks: [createLangChainCallbacks(stream, { agentId: 'researcher' })] },
);

For a LangGraph supervisor or swarm, bySubgraph gives each sub-agent its own line. It works with callbacks and with streamEvents:

import { bySubgraph, tapLangChainStream } from '@dan-build/agentstat/langchain';

await tapLangChainStream(
  stream,
  agent.streamEvents({ messages: [{ role: 'user', content: 'Plan the release.' }] }, { version: 'v2' }),
  { agentId: bySubgraph('supervisor') },
);

• Status: an agent is busy while it has a model or tool run open; tool runs are waiting. It's complete once nothing is running.
• Between runs an agent counts as busy (thinking after a tool). A graph node that works for over 30 seconds without a model or tool call reads as a thinking stall until the run ends.
• Errors: the end of the invocation closes whatever it left open, failing the agents if it errored. A run stopped through an AbortSignal isn't an error.
• streamEvents has no error events. Feed it through tapLangChainStream, or with createLangChainTap call tap.end(true) when the stream throws.
• Routing: agentId can be a function of the run ({ name, metadata, tags }). bySubgraph(fallback) routes a run to its createAgent agent's name, else to the LangGraph subgraph it runs in, else to fallback. Nested teams collapse to the outer team.

OpenTelemetry

If your app is instrumented with the OpenTelemetry GenAI conventions, add the span processor next to your exporter (SDK 1.x and 2.x). It maps spans by their gen_ai.operation.name:

import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { createOtelProcessor } from '@dan-build/agentstat/otel';

const provider = new NodeTracerProvider({
  spanProcessors: [createOtelProcessor(stream), new BatchSpanProcessor(new OTLPTraceExporter())],
});
provider.register();

• Model output is gen_ai.usage.output_tokens, else the text of gen_ai.output.messages. A call asks for tools through gen_ai.response.finish_reasons.
• Spans don't stream, so a call shows as one point at its average rate. Its time to first token comes from gen_ai.response.time_to_first_chunk when the instrumentation reports it.
• Errors: a span with status ERROR counts an error.
• Agents: a span belongs to its own gen_ai.agent.name, else its agent span's, else 'agent'. Pass { agentId } to change that: a string replaces the 'agent' fallback, a function decides every span. Two concurrent invocations of one agent name share a line.
• Which instrumentations work: those that set gen_ai.operation.name, such as the official OpenAI instrumentation, OpenLLMetry and AI SDK v7's @ai-sdk/otel. AI SDK v6 and earlier telemetry doesn't set it; use the AI SDK adapter there.

AG-UI

import { createAgUiTap } from '@dan-build/agentstat/ag-ui';

const tap = createAgUiTap(stream, 'researcher');
agent.subscribe({
  onEvent: ({ event }) => tap(event),
  // A failed connection sends no event: end the run yourself.
  onRunFailed: () => tap({ type: 'RUN_ERROR' }),
});

• Runs map to start and finish; RUN_ERROR counts an error, unless its code is abort (a stopped run).
• Output: message, reasoning and tool-argument deltas.
• Tool calls are waiting from TOOL_CALL_END to TOOL_CALL_RESULT, or to the end of the run for tools your frontend runs.
• Subagents: their events are left to their own agents. Give a child agent's stream its own tap.
• Pass events as the AG-UI client delivers them. It expands *_CHUNK events; raw chunk streams lose tool calls.

From a server (SSE)

When agents run on a server, keep a stream there, serve it as Server-Sent Events, and connect the chart to it.

// lib/agent-stream.ts — server only
import { createAgentStream } from '@dan-build/agentstat/sse';

export const agentStream = createAgentStream(); // feed it with the adapters above

// app/api/agents/route.ts — any Fetch-API runtime (Next.js, Hono, Bun, Deno, Workers)
import { createSseHandler } from '@dan-build/agentstat/sse';
import { agentStream } from '@/lib/agent-stream';

export const GET = createSseHandler(agentStream);

// In the browser
'use client';
import { useMemo } from 'react';
import { AgentStat } from '@dan-build/agentstat';
import { connectSse } from '@dan-build/agentstat/sse';

export default function Monitor() {
  const source = useMemo(() => connectSse('/api/agents'), []);
  return <AgentStat source={source} metric="tokens" anomalyDetection />;
}

• History: a chart that connects gets the stream's last 5 minutes, then live events.
• One connection serves every chart that uses the same connectSse source.
• Reconnects: after a drop or an error response (a proxy during a deploy) it reconnects with backoff and resumes where it stopped. A restarted server's replay isn't shown twice.
• Clocks: event times travel as ages, so they don't depend on the two machines' clocks agreeing.
• Slow clients are disconnected rather than buffered once they fall more than maxQueued (1,000) messages behind; they reconnect and resume.
• Keep-alive: a comment line every heartbeatMs (15 s) keeps proxies from closing the connection.

The stream lives in one server process. Several replicas or serverless instances each have their own, so the agents a browser sees depend on which instance it reaches.

Custom sources

For any other transport, push its messages into a stream. Here a WebSocket server sends { agentId, event } messages:

'use client';
import { useEffect } from 'react';
import { AgentStat, useAgentStream, type AgentEvent } from '@dan-build/agentstat';

export default function Monitor({ url }: { url: string }) {
  const stream = useAgentStream();

  useEffect(() => {
    const socket = new WebSocket(url);
    socket.onmessage = (message) => {
      const { agentId, event }: { agentId: string; event: AgentEvent } = JSON.parse(message.data);
      // Times are in the sender's clock, which means nothing here.
      delete event.time;
      if (event.type === 'tokens') delete event.since;
      stream.push(agentId, event);
    };
    return () => socket.close();
  }, [stream, url]);

  return <AgentStat source={stream} metric="tokens" anomalyDetection />;
}

Events pushed without a time are stamped on arrival. Drop since too, since it's in the same clock. To implement a source directly instead, see Your own source.