Bring your own observability

Pure SDK usage is fully local. A run emits trace events to whatever sinks you pass — and to nothing else. The Assemble platform is never contacted unless you explicitly attach createPlatformTraceSink; there is no phone-home, no background upload, no implicit default. If your observability stack is Langfuse, an OpenTelemetry collector, Datadog, or a homegrown pipeline, you point the same event stream at it by implementing one small interface.

Local by default#

A run with no sinks produces no output anywhere — the events still exist on the result as result.events, but nothing leaves the process. A run with only your own sink sends events only to your collector. The platform sink is one sink among equals, added the same way as any other:

const result = await summarizer.run(conversation, {
  sinks: [myCollectorSink],
});

This is the whole privacy model for library usage: your model provider sees your prompts because it must; everyone else sees exactly what you forward.

The TraceSink interface#

A sink is an object with a name and a write method, plus optional lifecycle hooks. This is the real shape from the SDK:

import type { TraceEvent } from "@assemble-dev/shared-types/trace-events";

export type TraceSink = {
  name: string;
  writeTraceEvent(event: TraceEvent): void | Promise<void>;
  flush?(): Promise<void>;
  close?(): void;
};

writeTraceEvent is called once per event, in order. It may return a promise, but slow sinks slow the run — buffer internally and drain in flush() if your collector is remote. flush() is awaited when the run finishes, and close() releases whatever the sink holds. See tracing for the event union: every event carries id, runId, workflowName, timestamp, and parentEventId, plus type-specific fields.

Your own code never needs that subpath import: TraceEvent, TraceEventSchema, TraceEventType, traceEventTypes, TokenUsage, and JsonValue are all re-exported from the @assemble-dev/sdk root — see foundation types.

Forwarding events to your own collector#

Trace events map naturally onto spans: paired *.started / *.completed events open and close a span, parentEventId gives you the parent relationship, and failure events carry the error. A sketch of a sink that forwards to a Langfuse- or OTel-style collector:

import type { TraceEvent, TraceSink } from "@assemble-dev/sdk";

export function createCollectorSink(collector: SpanCollector): TraceSink {
  const openSpans = new Map<string, SpanHandle>();
  const buffered: TraceEvent[] = [];

  return {
    name: "collector",
    writeTraceEvent(event: TraceEvent): void {
      buffered.push(event);

      if (event.type.endsWith(".started") || event.type === "tool.requested") {
        openSpans.set(event.id, collector.startSpan({
          name: event.type,
          parentSpanId: event.parentEventId ?? undefined,
          attributes: { runId: event.runId, workflowName: event.workflowName },
        }));
      }

      if (event.type.endsWith(".completed") || event.type.endsWith(".failed")) {
        openSpans.get(event.parentEventId ?? "")?.end({ event });
      }
    },
    async flush(): Promise<void> {
      await collector.export(buffered);
      buffered.length = 0;
    },
  };
}

The mapping details depend on your backend — Langfuse wants generations for agent events and spans for tools, OTel wants span kinds and status codes — but the shape is always the same: open on started, close on completed or failed, attach everything else as attributes or span events.

Nesting runs inside existing traces#

An agent run triggered from an instrumented request should appear inside that request's trace, not as an orphaned root. Pass your active trace context through run metadata:

const activeSpan = tracer.getActiveSpan();

const result = await summarizer.run(conversation, {
  sinks: [myCollectorSink],
  metadata: {
    parentTraceId: activeSpan.traceId,
    parentSpanId: activeSpan.spanId,
  },
});

parentTraceId and parentSpanId appear on every event the run emits, so your sink can attach the run's root span under the caller's span instead of starting a new trace. The result: one continuous waterfall from HTTP request through agent steps to tool calls, in the tracing UI you already use.

Cost attribution with run metadata#

The other metadata fields — sessionId, userId, tags — are stamped on every trace event of the run:

await summarizer.run(conversation, {
  sinks: [myCollectorSink],
  metadata: { sessionId: "sess_42", userId: "user_7", tags: ["support", "tier-1"] },
});

Because every event carries them, your collector can aggregate token usage and cost per user, per session, or per tag without joining against application state — or against run results. Each agent.completed event already carries the aggregate tokenUsage and costUsd for that agent call (costUsd is null when the model's per-token cost is unknown — see cost estimation), so grouping those events by userId and summing is the whole job. This is how you answer "which customers cost the most to serve" from your own warehouse, with the platform never in the loop.

Batch runs#

Anthropic message batches run outside the run pipeline, so the batch client does not take sinks — it takes an onTraceEvent callback. Bridging that callback to your sink closes the cost-attribution gap: without it, batch spend is invisible to the pipeline that already prices your live runs.

const batchClient = createAnthropicBatchClient({
  apiKeyName: "batch",
  onTraceEvent: (traceEvent) => myCollectorSink.writeTraceEvent(traceEvent),
  metadata: { userId: "user_7", tags: ["nightly-classify"] },
});

Each batch.result event carries the request's tokenUsage and costUsd, and the client's metadata is stamped on every event — so the same group-by-userId aggregation covers batch spend too. See batch observability for the full event list.