@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.0.0
@mastra/observability >= 0.1.0

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 (no parentSpanId)
• Token usage, cost, and latency aggregated across all spans of a trace
• Tool calls within agent workflows (function_call and mcp_tool_call types)
• Workflow steps — each createStep node's inputData captured as tool params (see below)
• Error reporting for failed traces
• Per-user aggregation across agent runs

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.
• The workflow_run root interaction (which carries no agent and no model) is labelled with workflow_name = the root workflow's entityName, so it is attributable rather than an unlabelled "unknown" row.

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" | Key to read from Mastra's requestContext. Set to null to disable. | | observabilityLevel | 0 | 0 = metrics only, 1 = + tool calls, 2 = + text previews | | 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 | | debug | false | Log transport errors | | disabled | false | Disable all telemetry |

Feedback

User feedback (thumbs up/down, ratings, comments) is not auto-collected — wire it from your 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 after every exported LLM interaction
});
// 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,
  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
});

Each track() call emits a [CLY]_llm_interaction_feedback event whose prompt_id links back to the [CLY]_llm_interaction event — powering prompt → feedback funnels and per-model satisfaction breakdowns in Countly. In a real app, store prompt_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 unified data model, observability levels, cost calculation, privacy controls, and Countly plugin integration (Drill, Funnels, Cohorts, APM, Crash Analytics).

License

MIT