Capture AI SDK token usage, tool calls, and cost with evlog

evlog/ai adds model and token data from the calls you wrap or capture to the request's wide event, a single structured log record. The event also holds HTTP details and any business context your code adds.

An evlog drain sends events to a destination such as Axiom or Datadog. Group them by route, user, or customer to compare usage and find failures.

Copy link to headingWhat evlog records for an AI SDK request

Token counts in the request's ai object accumulate across wrapped or captured calls. Fields such as the model and finish reason describe the last recorded result. This illustrates a completed, single-model streaming request:

{
  "method": "POST",
  "path": "/api/chat",
  "status": 200,
  "durationMs": 4512,
  "ai": {
    "calls": 1,
    "model": "claude-opus-5.5",
    "provider": "anthropic",
    "inputTokens": 3312,
    "outputTokens": 814,
    "totalTokens": 4126,
    "reasoningTokens": 225,
    "finishReason": "stop",
    "msToFirstChunk": 234,
    "msToFinish": 4500,
    "tokensPerSecond": 181
  }
}

Use these fields to filter, group, and alert on AI activity in your drain:

Fields you add with log.set(), such as a user ID or plan, sit beside the ai object on the same event. Prompts, model output, and tool inputs are excluded by default. See Control what evlog records to opt in.

Copy link to headingWhen to use evlog alongside OpenTelemetry traces

AI SDK telemetry can send OpenTelemetry spans to a tracing backend so you can inspect individual model and tool calls. evlog adds request-level fields for questions across requests, such as which route or customer uses the most tokens. Choose the view that answers your question:

On AI SDK 7, telemetry.integrations replaces the global integrations for that call. Include your OpenTelemetry integration alongside createEvlogIntegration(ai) to keep sending spans to your tracing backend.

For eve agents, use Edgenivo Agent Runs to trace sessions and tool calls.

Copy link to headingAdd evlog to an AI SDK app

evlog/ai wraps your model with AI SDK middleware that observes each call and writes the results to the request's wide event. The middleware doesn't change your prompts, callbacks, or responses, and it re-throws any error after recording it.

Copy link to heading1. Install packages and configure evlog

Install evlog and the AI SDK:

pnpm i ai evlog

npm i ai evlog

yarn add ai evlog

bun add ai evlog

The examples use the Next.js App Router with AI SDK 7 and Node.js 22 or later.

For local AI Gateway requests, set AI_GATEWAY_API_KEY in .env.local using the authentication setup. Edgenivo deployments can use the project's OIDC credentials.

Create lib/evlog.ts with the Next.js integration:

import { createEvlog } from 'evlog/next'

export const { withEvlog, useLogger, log, createError } = createEvlog({
  service: 'my-app',
})

Copy link to heading2. Wrap the model

Create an AI logger from the request logger, then pass ai.wrap() the model you already use. ai.wrap() accepts an AI Gateway model string, such as anthropic/claude-opus-5.5, or a model object:

import { streamText } from 'ai'
import { createAILogger } from 'evlog/ai'
import { withEvlog, useLogger } from '@/lib/evlog'

export const POST = withEvlog(async (request: Request) => {
  const { prompt } = await request.json()
  const log = useLogger()
  const ai = createAILogger(log)

  const result = streamText({
    model: ai.wrap('anthropic/claude-opus-5.5'),
    prompt,
  })

  return result.toUIMessageStreamResponse()
})

withEvlog() passes the incoming Request to your handler, and useLogger() takes no arguments because it reads the request's logger from AsyncLocalStorage.

The route returns an AI SDK UI message stream and still expects a { prompt } request body. evlog waits for the stream to finish before emitting the request event so completed calls can add their usage and timing.

Update clients that expect plain text to read the UI message protocol. With useChat, also configure its request body to send prompt because the default transport sends messages.

Copy link to heading3. Add tool timing

The middleware records model calls; the telemetry integration adds tool execution timing. Inside the same route handler, pass the parsed prompt and your application's tool definitions to generateText(). Both observers share the AI logger:

import { generateText } from 'ai'
import { createAILogger, createEvlogIntegration } from 'evlog/ai'

// Inside withEvlog(), using the request's log and prompt.
// getWeather and searchDocs are your application's AI SDK tools.
const ai = createAILogger(log)

const result = await generateText({
  model: ai.wrap('anthropic/claude-opus-5.5'),
  prompt,
  tools: { getWeather, searchDocs },
  telemetry: {
    integrations: [createEvlogIntegration(ai)],
  },
})

The integration adds ai.tools[] with each executed tool's name, duration, and success status, plus an error when execution fails. It also records the generation's total duration in ai.totalDurationMs.

Use telemetry on AI SDK 7 or experimental_telemetry on AI SDK 6. Version 7 also reports stream aborts and unrecoverable errors through the integration.

Copy link to heading4. Estimate cost

Pass a cost map with input and output prices in dollars per million tokens. For a request that uses one language model, evlog applies those rates to the accumulated token counts. Use the result as an estimate; it does not account for separate cache prices, discounts, or other provider charges:

const ai = createAILogger(log, {
  cost: {
    'claude-opus-5.5': { input: 4, output: 20 },
  },
})

Keep the rates with your model configuration and check them against the model's current pricing. ai.getEstimatedCost() reads the estimate for usage recorded so far; it does not predict a future call.

In evlog 2.34.0, the estimate applies the last language model's rates to all accumulated tokens. Mixed-model requests and RAG requests with captured embeddings therefore need a separate calculation using each model's usage and rates. Use provider or Gateway usage records to reconcile charges.

Copy link to heading5. Record embeddings

Embedding calls use a separate API. On AI SDK 7, pass createEvlogIntegration(ai) with the call to record usage in the request event:

import { embedMany } from 'ai'
import { createEvlogIntegration } from 'evlog/ai'

// Reuse the request's AI logger and your text chunks.
const { embeddings } = await embedMany({
  model: 'openai/text-embedding-3-small',
  values: chunks,
  telemetry: {
    integrations: [createEvlogIntegration(ai)],
  },
})

On AI SDK 6, capture the returned usage explicitly. Use this alternative instead of automatic capture to avoid counting embeddings twice:

const { embeddings, usage } = await embedMany({
  model: 'openai/text-embedding-3-small',
  values: chunks,
})

ai.captureEmbed({
  usage,
  model: 'openai/text-embedding-3-small',
  dimensions: embeddings[0]?.length,
  count: embeddings.length,
})

If another library already wraps your model, such as guardrails or memory middleware, pass the wrapped model to ai.wrap(). To control middleware order yourself, use createAIMiddleware(log) with the AI SDK's wrapLanguageModel().

Copy link to headingUse AI metadata in your handler

The AI logger exposes accumulated request data while the handler runs. Use it to display estimated usage, save completed calls, or decide whether to continue a multi-step run. Keep billing records separate from sampled observability events.

Copy link to headingStore usage after a call

ai.getMetadata() returns the usage recorded so far. Read it after generateText() completes to store the result. In this route, getCurrentUser() represents your application's authentication helper and must return the verified user or null:

import { generateText } from 'ai'
import { createAILogger } from 'evlog/ai'
import { withEvlog, useLogger } from '@/lib/evlog'
import { getCurrentUser } from '@/lib/auth'

export const POST = withEvlog(async (request: Request) => {
  const user = await getCurrentUser(request)
  if (!user) return new Response('Unauthorized', { status: 401 })

  const { prompt } = await request.json()
  const userId = user.id
  const log = useLogger()
  log.set({ user: { id: userId } })

  const ai = createAILogger(log, {
    cost: { 'claude-opus-5.5': { input: 4, output: 20 } },
  })

  const { text } = await generateText({
    model: ai.wrap('anthropic/claude-opus-5.5'),
    prompt,
  })

  const usage = ai.getMetadata()
  await saveUsage({
    userId,
    model: usage.model,
    inputTokens: usage.inputTokens,
    outputTokens: usage.outputTokens,
    estimatedCost: usage.estimatedCost,
  })

  return Response.json({ text })
})

saveUsage() represents your database write. Treat the metadata snapshot as read-only. Copy nested tool inputs before changing them because those objects can still share references with the logger's state.

ai.getEstimatedCost() returns the current estimate when matching rates and nonzero token usage are available. Otherwise, it returns undefined.

Copy link to headingReact to each step

ai.onUpdate() runs when evlog updates the accumulated metadata, including after a model step or captured embedding. The listener below aborts the run after the reported estimate exceeds $0.50:

import { generateText, stepCountIs } from 'ai'

const controller = new AbortController()
const ai = createAILogger(log, {
  cost: { 'claude-opus-5.5': { input: 4, output: 20 } },
})

const stopWatching = ai.onUpdate((metadata) => {
  if ((metadata.estimatedCost ?? 0) > 0.5) controller.abort()
})

try {
  await generateText({
    model: ai.wrap('anthropic/claude-opus-5.5'),
    tools: { searchDocs, queryDatabase },
    stopWhen: stepCountIs(10),
    abortSignal: controller.signal,
    prompt,
  })
} finally {
  stopWatching()
}

The check happens after usage is recorded, so it cannot cap a call's cost or undo charges already incurred. Use it to stop further work, with a step limit and output-token limits as additional bounds.

onUpdate() returns a function that unsubscribes the listener. Errors thrown inside a listener are isolated and never break the AI call.

Copy link to headingControl what evlog records and run it in production

evlog/ai records metadata by default and captures content only when you enable it by name. Decide what your drain destination may store before you turn content capture on, then tune sampling so failed and expensive requests are always kept.

Copy link to headingKnow the defaults

Copy link to headingCapture content with limits

Enable prompt, output, and tool-input capture separately. Each option accepts a transform to redact content before the maxLength limit truncates it:

const ai = createAILogger(log, {
  prompt: { maxLength: 500, transform: (text) => redact(text) },
  output: { maxLength: 500 },
  toolInputs: {
    maxLength: 200,
    transform: (input, toolName) =>
      toolName === 'queryDatabase' ? { sql: '***' } : input,
  },
})

redact() represents your own redaction function. ai.prompt contains the first wrapped model call's formatted prompt, and ai.output contains the last wrapped call's text. Runs stopped by a tool or step limit may have no final answer.

With toolInputs enabled, ai.toolCalls contains { name, input } objects. The input may be truncated text after the transform and length limit are applied.

Prompt and output capture are middleware features. If you use createEvlogIntegration(log) on its own, without ai.wrap() or createAIMiddleware(), ai.prompt and ai.output stay empty.

When NODE_ENV is production, evlog auto-redacts common personal data before console output and drain delivery. It partially masks emails, credit card numbers, IP addresses, phone numbers, JWTs, bearer tokens, and IBANs. For example, [email protected] becomes a***@***.com.

Auto-redaction is off in development and can be changed with the redact option. Use transform for content those patterns don't recognize, such as SQL, customer names, or internal IDs.

Warning: Captured prompts, outputs, and tool inputs can still contain sensitive data that auto-redaction doesn't recognize, and anyone with access to your drain destination can read them. If your app handles personal, sensitive, or regulated data, review what you record and whether your privacy disclosures cover it.

Copy link to headingConfigure for production

Batch drain writes to reduce outbound requests. The Next.js integration keeps drain work alive after the response; it still performs event processing as part of the request lifecycle. Configure batching and sampling in lib/evlog.ts:

import { createEvlog } from 'evlog/next'
import { createAxiomDrain } from 'evlog/axiom'
import { createDrainPipeline } from 'evlog/pipeline'

export const { withEvlog, useLogger, log, createError } = createEvlog({
  service: 'my-app',
  sampling: {
    rates: { info: 10 },
    keep: [{ status: 400 }, { duration: 5000 }],
  },
  drain: createDrainPipeline({ batch: { size: 50, intervalMs: 5000 } })(
    createAxiomDrain(),
  ),
})

Sampling rates are percentages, so info: 10 keeps roughly 1 in 10 info-level events. Errors default to 100%, but setting rates.error to any lower percentage samples them. Trace events default to 0%; fatal events are always kept.

The keep rules retain requests with a status of 400 or higher or a duration of at least 5 seconds. Configure AXIOM_API_KEY and AXIOM_DATASET for this drain, or use the adapter configuration for your destination.

Copy link to headingKeep expensive requests

Add a keep callback to the existing lib/evlog.ts configuration. Retain its imports, drain, and status/duration rules so high-token requests and failed tools are kept alongside those earlier conditions:

export const { withEvlog, useLogger, log, createError } = createEvlog({
  service: 'my-app',
  sampling: {
    rates: { info: 10 },
    keep: [{ status: 400 }, { duration: 5000 }],
  },
  keep(ctx) {
    const ai = ctx.context.ai as {
      totalTokens?: number
      tools?: Array<{ success: boolean }>
    } | undefined
    if ((ai?.totalTokens ?? 0) > 10_000) ctx.shouldKeep = true
    if (ai?.tools?.some((tool) => !tool.success)) ctx.shouldKeep = true
  },
  drain: createDrainPipeline({
    batch: { size: 50, intervalMs: 5000 },
  })(createAxiomDrain()),
})

Tail sampling runs after the request completes, so it can act on the final ai totals before head sampling applies.

Copy link to headingInvestigate an expensive or failing AI request

Start from the wide events to find which requests went wrong, then read the ai fields to narrow the cause. Suppose the daily cost of a support chat route doubled, and users report that some answers stop partway through.

Copy link to heading1. Find the requests in your drain

Filter on path, such as /api/chat. Calculate cost totals and averages from unsampled usage records or comparable periods with 100% capture.

The earlier policy keeps expensive requests while dropping many routine ones, so raw totals and averages from that sample do not represent all traffic.

• Group usage by authenticated user.id or plan to find who drove the increase. Account for the cost-estimation limits when comparing dollar totals.
• Compare token counts and model steps across equivalent traffic. Count single-step calls in the average even though ai.steps is absent.
• Inspect ai.error and abnormal finish reasons such as length, error, or abort to find incomplete answers.

The keep callback retains requests over 10,000 tokens and requests with failed tools recorded by createEvlogIntegration(ai). These events help diagnose individual requests but overrepresent expensive or failed work.

Copy link to heading2. Read the request's ai fields

Each wide event is self-contained, so you can often find the cause without a trace. Compare what the request recorded against what you expected:

In this example, costly requests averaged 8 steps with searchDocs called 6 times, and the truncated answers all ended with ai.finishReason set to length.

Copy link to heading3. Look at one request in detail

Use a trace when the event fields leave the cause unclear. If you export AI SDK telemetry, inspect the model and tool spans for the affected request.

For a reproducible content problem, enable limited prompt, output, and toolInputs capture on a preview deployment. Reproduce it with test data and inspect the captured fields before deciding whether to collect that content in production.

Copy link to heading4. Fix the cause and confirm it in the events

Test a more specific searchDocs description and a lower step limit, such as stepCountIs(4). Confirm that the agent can still complete the task.

Raise maxOutputTokens only if the expected answer needs more room. Compare equivalent, unsampled test runs before and after each change:

• Check for fewer steps and tokens without reducing answer quality.
• Inspect how often searchDocs runs and whether its results are useful.
• Confirm that answers complete without reaching the output limit.

Copy link to headingRecord AI calls in background jobs and Edgenivo Daemons

Outside a request, no middleware creates or emits the wide event, so you create a logger for each job and emit it yourself. Use this pattern for queue workers, scripts, and Edgenivo Daemons, where one long-running process handles many jobs.

Cron Jobs and Edgenivo Queues in push mode call a route instead. Wrap those routes with withEvlog() as in the earlier steps, which flushes the drain after the response.

Copy link to headingCreate one wide event per job

Set AXIOM_API_KEY and AXIOM_DATASET in the worker's environment before starting it. The Axiom adapter reads these values when it sends events.

Call initLogger() once at startup and create a logger for each job. Pass it to createAILogger() and emit the completed event in finally:

import { generateText } from 'ai'
import { createLogger, initLogger } from 'evlog'
import { createAILogger } from 'evlog/ai'
import { createAxiomDrain } from 'evlog/axiom'
import { createDrainPipeline } from 'evlog/pipeline'

export const drain = createDrainPipeline({
  batch: { size: 50, intervalMs: 5000 },
})(createAxiomDrain())

initLogger({ env: { service: 'ticket-summarizer' }, drain })

export async function summarizeTicket(job: {
  id: string
  ticketId: string
  text: string
}) {
  const log = createLogger({ jobId: job.id, ticketId: job.ticketId })
  const ai = createAILogger(log, {
    cost: { 'claude-opus-5.5': { input: 4, output: 20 } },
  })

  try {
    const { text } = await generateText({
      model: ai.wrap('anthropic/claude-opus-5.5'),
      system: 'Summarize this support ticket in three sentences.',
      prompt: job.text,
    })
    log.set({ status: 'complete' })
    return text
  } catch (error) {
    log.error(error instanceof Error ? error : new Error(String(error)))
    throw error
  } finally {
    log.emit()
  }
}

Each job produces one event with its jobId, ticketId, and ai totals, so you can query cost and failures per job type just as you would per route. Emitting in finally records failed jobs too.

Copy link to headingRun the worker as a Edgenivo Daemon

Edgenivo Daemons run containers continuously on Fluid compute and pull work from Edgenivo Queues, databases, or third-party systems instead of receiving HTTP traffic.

During the public beta, daemons run only in production.

Package the worker in worker/Dockerfile and set its command to run the job loop that calls summarizeTicket(). Follow the daemon setup steps, then declare its instances in vercel.json:

{
  "$schema": "https://openapi.vercel.sh/vercel.json",
  "daemons": {
    "summarizer": {
      "root": "worker/",
      "entrypoint": "Dockerfile",
      "instances": { "iad1": 1 }
    }
  }
}

During a rollout, Edgenivo sends SIGTERM and allows 30 seconds for shutdown by default. Stop pulling work, wait for active jobs, and flush the buffered events before exiting. This handler sets a 25-second deadline within that grace period:

import { drain } from './summarize'

process.once('SIGTERM', () => {
  const deadline = setTimeout(() => process.exit(1), 25_000)

  void (async () => {
    try {
      await stopPullingJobs()
      await waitForActiveJobs()
      await drain.flush()
      clearTimeout(deadline)
      process.exit(0)
    } catch (error) {
      console.error('Worker shutdown failed', error)
      clearTimeout(deadline)
      process.exit(1)
    }
  })()
})

stopPullingJobs() must prevent new work from starting, and waitForActiveJobs() must resolve after the current jobs finish. Their implementations depend on your worker. Save the previous module as summarize.ts so this import resolves.

Adjust the deadline if you change the daemon's terminationGracePeriodSeconds. Standalone scripts also need to await drain.flush() before exiting. Flush attempts delivery with the pipeline's retry policy; configure onDropped if you need to track batches that still fail.

Copy link to headingAttribute AI usage and agent actions in enterprise apps

Internal apps need AI usage tied to a verified person and a record of the actions an agent performs. Add Passport identity to the request event with log.set(), then correlate that event with the audits for individual actions.

Copy link to headingAttach a verified identity with Edgenivo Passport

Install @vercel/passport and configure your identity provider. On Enterprise, Passport signs visitors in before requests reach your app.

getIdentity() verifies tokens on the server and returns null if none is present.

Attach the identity to the wide event before the AI call:

import { getIdentity } from '@vercel/passport'
import { streamText } from 'ai'
import { createAILogger } from 'evlog/ai'
import { withEvlog, useLogger } from '@/lib/evlog'

export const POST = withEvlog(async (request: Request) => {
  const identity = await getIdentity(request)
  if (!identity) return new Response('Unauthorized', { status: 401 })

  const groups = Array.isArray(identity.payload.groups)
    ? identity.payload.groups
    : []

  const log = useLogger()
  log.set({ user: { id: identity.subject, groups } })

  const { prompt } = await request.json()
  const ai = createAILogger(log)

  const result = streamText({
    model: ai.wrap('anthropic/claude-opus-5.5'),
    prompt,
  })

  return result.toUIMessageStreamResponse()
})

Use identity.subject as the visitor ID for a single Passport issuer. If you accept identities from multiple Edgenivo teams, store identity.payload.iss with it. Treat email as an optional display value.

To group usage by team, forward your identity provider's group claims through Connect. Add the cost map to compare estimated spending.

In local development, getIdentity() returns a test identity with verified: false, so test real claims on a Passport-protected preview deployment.

Copy link to headingAudit actions your agent takes

log.audit() sets one audit object on a logger. Repeated calls update that object, so an agent that changes several records needs a separate event for each action.

Within a withEvlog() handler, use log.fork() to give each action its own logger. The child event retains the parent's request ID for correlation. This tool records successful and failed refund attempts:

import { tool } from 'ai'
import { z } from 'zod'
import { useLogger } from '@/lib/evlog'

// Define inside the authenticated withEvlog handler.
const fork = log.fork
if (!fork) throw new Error('Expected a withEvlog request logger')

const issueRefund = tool({
  description: 'Refund an order for the signed-in customer.',
  inputSchema: z.object({
    orderId: z.string(),
    amount: z.number().positive(),
  }),
  execute: async ({ orderId, amount }) => {
    let outcome: 'success' | 'failure' = 'failure'
    try {
      const refund = await refundOrder({
        userId: identity.subject,
        orderId,
        amount,
      })
      outcome = 'success'
      return refund
    } finally {
      await fork('order.refund', () => {
        const actionLog = useLogger()
        actionLog.set({ user: { id: identity.subject } })
        actionLog.audit({
          action: 'order.refund',
          actor: {
            type: 'agent',
            id: 'support-agent',
            model: 'claude-opus-5.5',
          },
          target: { type: 'order', id: orderId },
          outcome,
        })
      })
    }
  },
})

refundOrder() represents your application's refund operation. It must authorize the verified user's access to the order, validate the amount against the order, and prevent duplicate refunds. The model's arguments do not establish permission.

Each child event contains one audit and the visitor ID. Its _parentRequestId links it to the request event's user and ai fields. Audit-bearing events bypass head sampling; the request event still follows its own sampling policy.

Audit delivery can fail without undoing a completed refund.

Choose the context and storage controls your audit records need:

• auditEnricher() adds available request context. Supply a tenantId resolver if you need tenant attribution.
• auditOnly() filters for audits. Set access and retention in the destination.
• signed() adds HMAC signatures or a hash chain. Chain state is in memory by default; persistent state and an independently trusted checkpoint are needed to verify continuity across restarts and detect changes to stored history.

Passport records successful sign-ins as passport-access-granted events in Edgenivo Audit Logs. Your app records agent actions separately.

Copy link to headingNext steps

• Extend the integration with streaming, RAG, and multi-model patterns.
• Review the AI fields and metadata methods.
• Choose a drain destination for your wide events.
• Configure head and tail sampling to control log volume.
• Add audit enrichment and tamper evidence to agent action logs.
• Require company sign-in with Edgenivo Passport.
• Run continuous background workers with Edgenivo Daemons.
• Export AI SDK telemetry spans to an OpenTelemetry backend.