@countly/ai-sdk-mastra

Countly AI observability adapter for Mastra.

Part of the Countly AI SDK — provider-agnostic LLM observability for every AI stack.

Install

npm install @countly/ai-sdk-mastra

@countly/ai-sdk-core is pulled in automatically.

Peer dependencies

@mastra/core          >= 1.26.0 < 2
@mastra/observability >= 1.11.0 < 2

The floor was raised in 0.0.6 (it was >= 1.0.0 for both, which was wishful). Cache-token capture reads usage.inputDetails / usage.outputDetails, a shape @mastra/observability only emits from 1.11.0. With auto-install-peers=true, an older pair installs cleanly and then reports zero cache tokens — which mis-prices every prompt-cached call in exactly the direction that hides the overbilling this release removes. Silence, not an error. Hence the floor, and hence CI's mastra-peer-floor job, which proves the declared minimum actually emits cache tokens instead of merely declaring that it does.

If you cannot move to those versions, pin @countly/[email protected] and accept the known cost error rather than run 0.0.6 below its floor.

Quick Start

Note: the exporter must be wrapped in new Observability({...}) and passed via Mastra's observability: field. Passing it directly as exporters: on Mastra won't work — no spans reach the exporter.

import { Mastra } from "@mastra/core/mastra";
import { Observability } from "@mastra/observability";
import { CountlyMastraExporter } from "@countly/ai-sdk-mastra";

new Mastra({
  observability: new Observability({
    configs: {
      default: {
        serviceName: "my-ai-app",
        exporters: [
          new CountlyMastraExporter({
            appKey: "YOUR_APP_KEY",
            url: "https://your-countly-server.com",
            requestContextDeviceIdKey: "countlyDeviceId",
          }),
        ],
      },
    },
  }),
});

In your request handler:

import { RequestContext } from "@mastra/core/request-context";

app.post("/chat", async (req, res) => {
  const ctx = new RequestContext();
  ctx.set("countlyDeviceId", req.user.id);
  await mastra.getAgent("intent").stream(messages, { requestContext: ctx });
});

How the user ID reaches the event

The bridge is Mastra's runtime, not our SDK. We read a public field Mastra puts on every exported span:

Your handler                     Mastra runtime                  @countly/ai-sdk-mastra
─────────────                    ──────────────                  ──────────────────────
ctx = new RequestContext()
ctx.set("countlyDeviceId", id)
                                 run scope carries ctx
agent.stream(msg, {              ↓
  requestContext: ctx            spans created during run
})                               (AGENT_RUN, MODEL_GEN, TOOL_CALL)
                                 ↓
                                 ExportedSpan.requestContext = {
                                   countlyDeviceId: id              adapter reads
                                 }                                  span.requestContext
                                                                    ↓
                                                                    event.deviceId = id
                                                                    POST /i?device_id=id

• Rename the key via requestContextDeviceIdKey (e.g. "myUserId")
• Set to null to disable — falls back to getDeviceId() / deviceId / process UUID
• No Mastra version requirement beyond >=1.0 — stable v1 observability surface

What's captured

• Per-event tracing via exportTracingEvent (span_started, span_updated, span_ended)
• Automatic trace completion when the root span ends
• One [CLY]_llm_interaction per model_generation span — its own model, provider, usage, cost, finish reason and span duration (see below)
• Tool calls within agent workflows (function_call and mcp_tool_call types), with per-call latency and the requesting generation as their parent
• Workflow steps — each createStep node's inputData captured as tool params (see below)
• Mastra's own scorer results and user feedback (onScoreEvent / onFeedbackEvent / addScoreToTrace) as [CLY]_llm_interaction_feedback rows
• Token usage from usage.inputDetails / usage.outputDetails, with cache read, cache write and reasoning as subsets of the input/output totals. usage_input is the provider's total, inclusive of both cache buckets, so the cache premium is billed exactly once
• Usage deduped by span id (it used to be counted twice), and usage the trace never reported is omitted with usage_state: "not_reported" rather than zeroed
• Error reporting for failed generations (a failed tool or step is reported on its own tool row, not as a model failure)
• Per-user aggregation across agent runs

One row per generation

A Mastra trace is not a turn. Mastra roots traces at both agent_run and workflow_run, so one user turn produces several traces — and a single trace can contain several generations with different models.

• Each model_generation span becomes exactly one interaction row, carrying its own model, provider, usage_*, cost_*, finish_reason and latency_total. Cost is therefore correct per call instead of pricing the whole turn at the first model's rate.
• Turn context is denormalized onto every row: run_id, generation_index (0-based, ordered by start time), run_latency_total (the root span's end-to-end duration), workflow_name, agent_name, thread_id, deviceId, trace_id and the text previews.
• A trace that contained no generation emits no interaction row. Workflow roots, rag_* pipelines, memory operations, tripwire aborts and directly executed tools are orchestration, not LLM calls. Their tool rows still ship, parented to the run (parent_event_key: "run").

⚠️ Row volume rises. A turn that previously emitted one interaction row now emits one per generation — roughly 2x for a single-agent turn, more for a multi-agent or multi-model one. Size your ingest accordingly. Conversely, prompt count drops by the share of traces that never contained a generation (200 of 601 rows on observed traffic), which is the point of the change.

🔴 api_host_type is "unknown" on every Mastra row

Every Mastra row reports api_host_type: "unknown" and omits api_host. Not a bug in this adapter — Mastra does not tell us the endpoint, and we will not guess one.

@mastra/core declares serverAddress and serverPort on ModelGenerationAttributes, but no code path assigns either. Verified against the installed artifacts rather than the type declarations: in @mastra/core 1.55.0 (what this workspace resolves) and 1.24.1, and in @mastra/observability 1.11.1, serverAddress appears only in dist/observability/types/tracing.d.ts and in source maps — zero occurrences in emitted JavaScript. What a model_generation span actually carries at runtime is:

| when | attributes populated | |---|---| | span created | model, provider, streaming, parameters | | span ended | finishReason, responseId, responseModel, usage |

No other attribute carries a URL, host or region, so there is nothing to fall back to. providerMetadata cannot help either: Mastra consumes it inside extractUsageMetrics and it is in the exporter's DEFAULT_KEYS_TO_STRIP, so it never reaches an exported span — and it holds no endpoint in any case.

Consequences, stated plainly:

• The dashboard's "By host type" card is empty for Mastra traffic. Gateway vs vendor-direct vs Azure cannot be broken out.
• "unknown" here means "Mastra did not report it", not "the call went somewhere unusual". Do not read it as a finding.

The classification is already wired (describeApiHost), so it starts working the moment Mastra populates the attribute — no change needed on our side. Every other adapter reads its client's baseURL directly and does report a real host; see the capability matrix.

run_id — grouping a turn

Every row of a turn shares one run_id, resolved in this order:

1. getPromptId() — your own id for the turn.
2. requestContextPromptIdKey read from the trace's requestContext.
3. Mastra's own run id (span.metadata.runId).
4. The trace's root span id.

Rules 1 and 2 are the only ones that can group a workflow trace with the agent traces it spawns, because those are separate traces with different traceIds. Without one, a workflow turn appears as several runs. Set the value before invoking the agent or workflow: Mastra clones requestContext in the span constructor, so a value set inside a step is only visible on that step's own span (the exporter falls back to scanning child spans for exactly that case).

flush() vs shutdown()

flush() drains the network buffer but does not emit still-running traces — ObservabilityBus calls it mid-request, and reporting a live turn would produce a zero-latency premature success plus a duplicate row when the root finally ends. shutdown() does emit whatever is left, explicitly marked status: "incomplete", so a process exiting mid-turn loses nothing and claims nothing.

Capturing workflow steps

If you orchestrate agents with Mastra workflows (createWorkflow / createStep), the steps only reach the exporter when the workflow is registered on the Mastra instance and run through that instance. A workflow imported and run standalone (myWorkflow.createRun()) has no observability wired to it — only the agent runs it invokes are traced, and the step decisions (routing, handoffs, gathered inputs) are invisible.

Register the workflow and run it via getWorkflow:

const mastra = new Mastra({
  agents: { intentAgent, pluginsAgent },
  workflows: { myWorkflow },        // ← register it
  observability: new Observability({ /* … exporter … */ }),
});

// Run THROUGH the instance so its spans inherit observability:
const run = await mastra.getWorkflow("myWorkflow").createRun();
await run.stream({ inputData, requestContext: ctx });

Once registered, each leaf createStep surfaces as a tool row:

• tool_name = the step's entityId (hyphens normalized to underscores, e.g. potential-handoff → potential_handoff), so it aggregates with function/MCP tool rows of the same name.
• [CLY]_llm_tool_usage_parameter rows are emitted per key of the step's inputData — { handoff_to, confidence, user_input, … } become named params you can break down in analytics.
• Structural wrapper steps are skipped. A .then(nestedWorkflow) / .branch([… nestedWorkflow]) surfaces as a workflow_step whose entityId is that nested workflow's id; those are excluded so only real decision steps become rows.
• A step is orchestration, not a tool the model chose, so its rows are parented to the run (parent_event_key: "run", prompt_id = run_id) rather than to a generation. They are labelled with workflow_name = the root workflow's entityName (and agent_name when an agent_run span exists in the trace).
• A workflow_run root produces no [CLY]_llm_interaction — no LLM was invoked, so there is no generation to report. [CLY]_llm_interaction now means strictly one LLM generation. Earlier versions emitted a row here with no model, provider: "unknown", zero tokens and $0 cost but a real end-to-end latency; on real traffic that was a third of all rows, inflating prompt counts, understating cost per prompt and mixing workflow durations into the latency percentiles.

Nested workflows are traced too: register only the top-level workflow you run — its .then() / .branch() children are traced automatically as part of the run.

Configuration

All adapters accept the same CountlyAIConfig object:

| Field | Default | Description | |-------|---------|-------------| | appKey | required | Countly app key | | url | required | Countly server URL | | requestContextDeviceIdKey | "countlyDeviceId" | requestContext key holding the device/user id. null disables it. | | requestContextThreadIdKey | "threadId" | requestContext key holding the conversation id. Falls back to Mastra's span.metadata.threadId. null disables it. | | requestContextWorkflowNameKey | "workflowName" | requestContext key holding the workflow name. Falls back to the root workflow_run's entityName. null disables it. | | requestContextPromptIdKey | — | requestContext key holding your id for the turn → becomes run_id. Opt-in. | | getPromptId | — | Same thing as a function; outranks requestContextPromptIdKey. | | observabilityLevel | 0 | 0 = metrics only, 1 = + tool calls, 2 = + text previews and tool params | | tags | [] | Labels for cost attribution and filtering | | environment | "production" | Environment tag | | costModel | — | Custom pricing overrides | | flushInterval | 10000 | Buffer flush interval in ms | | maxBatchSize | 20 | Max events before auto-flush | | paramValueLength | 1024 | Cap on a tool argument's serialized value | | debug | false | Log transport errors and dropped/unrecognised spans | | disabled | false | Disable all telemetry |

Feedback

Mastra's native signals need no wiring: the exporter implements onScoreEvent, onFeedbackEvent and addScoreToTrace, so every scorer result and every recordedSpan.addFeedback() becomes a [CLY]_llm_interaction_feedback row against the run that produced it.

For feedback that comes from your own UI, capture the prompt_id of each tracked interaction via the onPrompt callback, then record feedback against it with createFeedbackTracker (re-exported from this package, so no extra install is needed):

import { CountlyMastraExporter, createFeedbackTracker, type PromptInfo } from "@countly/ai-sdk-mastra";

const countly = { appKey: "YOUR_APP_KEY", url: "https://your-countly-server.com" };

let lastPrompt: PromptInfo | undefined;
const exporter = new CountlyMastraExporter({
  ...countly,
  requestContextDeviceIdKey: "countlyDeviceId",
  onPrompt: (info) => { lastPrompt = info; }, // fires once per exported generation
});
// wire the exporter into new Observability({...}) as shown in Quick Start

const feedback = createFeedbackTracker(countly, { sdk_adapter: "mastra" });

await mastra.getAgent("intent").generate(messages, { requestContext: ctx });

// ...later, when the user rates the answer:
feedback.track({
  prompt_id: lastPrompt!.prompt_id, // rates the whole TURN (== run_id)
  rating: "thumbs_up", // or "thumbs_down", or any custom string
  score: 0.9, // optional 0-1 numeric score
  category: "helpful", // optional: hallucination, irrelevant, harmful, ...
  comment: "Great answer", // optional free-form text
  deviceId: user.id, // attribute to the same user as the interaction
});

// ...or rate ONE generation of a multi-generation turn:
feedback.track({
  prompt_id: lastPrompt!.event_id,  // that row's own id
  run_id: lastPrompt!.run_id,
  rating: "thumbs_down",
});

PromptInfo carries prompt_id (identical to run_id, so this flow is unchanged from earlier versions), run_id, the generation's own event_id, generation_index, plus provider / model / status / thread_id / agent_name / workflow_name / trace_id. It fires only when a generation was actually recorded — never for an orchestration trace.

Each track() call emits a [CLY]_llm_interaction_feedback event whose prompt_id links back to the run (or to one generation's event_id), with parent_event_key recording which — powering prompt → feedback funnels and per-model satisfaction breakdowns in Countly. In a real app, store the id alongside the rendered message (or return it to your client) and read it back when the user rates the answer. Feedback is batched like interaction events; call feedback.flush() to send immediately, or feedback.shutdown() on process exit.

Full documentation

See the Countly AI SDK repository for the schema v2 wire contract (one row per generation, RULE A dimensions, RULE B measures with their usage_state / cost_priced markers, and the common envelope), the adapter capability matrix, observability levels (0/1/2), cost calculation, privacy controls, and Countly plugin integration (Drill, Funnels, Cohorts, APM, Crash Analytics).

License

MIT