Tracing setup

AgentMark uses OpenTelemetry to collect distributed traces and metrics for your prompt executions. This page covers everything from basic setup to advanced tracing patterns.

Install the SDK

TypeScript
Python

npm install @agentmark-ai/sdk

pip install agentmark-sdk

Initialize tracing

TypeScript
Python

import { AgentMarkSDK } from "@agentmark-ai/sdk";
import { createAgentMarkClient, VercelAIModelRegistry } from "@agentmark-ai/ai-sdk-v5-adapter";
import { openai } from "@ai-sdk/openai";
import { generateText } from "ai";

const sdk = new AgentMarkSDK({
  apiKey: process.env.AGENTMARK_API_KEY,
  appId: process.env.AGENTMARK_APP_ID,
  baseUrl: process.env.AGENTMARK_BASE_URL  // defaults to https://api.agentmark.co
});

// Initialize tracing
const tracer = sdk.initTracing();

// Configure client
const modelRegistry = new VercelAIModelRegistry();
modelRegistry.registerProviders({ openai });

const client = createAgentMarkClient({
  loader: sdk.getApiLoader(),
  modelRegistry
});

// Load and run prompt with telemetry
const prompt = await client.loadTextPrompt("greeting.prompt.mdx");
const input = await prompt.format({
  props: { name: 'Alice' },
  telemetry: {
    isEnabled: true,
    functionId: "greeting-function",
    metadata: {
      userId: "123",
      environment: "production"
    }
  }
});

const result = await generateText(input);

// Shutdown tracer (only for short-running scripts)
await tracer.shutdown();

import os
from agentmark_sdk import AgentMarkSDK
from agentmark_pydantic_ai_v0 import run_text_prompt
from agentmark_client import client

sdk = AgentMarkSDK(
    api_key=os.environ["AGENTMARK_API_KEY"],
    app_id=os.environ["AGENTMARK_APP_ID"],
)

# Initialize tracing
tracer = sdk.init_tracing()

# Load and run prompt with telemetry
prompt = await client.load_text_prompt("greeting.prompt.mdx")
params = await prompt.format(
    props={"name": "Alice"},
    telemetry={
        "isEnabled": True,
        "functionId": "greeting-function",
        "metadata": {
            "userId": "123",
            "environment": "production",
        },
    },
)

result = await run_text_prompt(params)

# Shutdown tracer (only for short-running scripts)
await tracer.shutdown()

TypeScript
Python

For local development with npx agentmark dev, traces are sent to http://localhost:9418 automatically. Pass disableBatch: true for short-running scripts:

const tracer = sdk.initTracing({ disableBatch: true });

For local development with npx agentmark dev, traces are sent to http://localhost:9418 automatically. Pass disable_batch=True for short-running scripts:

tracer = sdk.init_tracing(disable_batch=True)

Collected spans

AgentMark records these OpenTelemetry spans:

Span type	Description	Key attributes
`ai.inference`	Full inference call lifecycle	`ai.model.id`, `ai.prompt`, `ai.response.text`, `ai.usage.promptTokens`, `ai.usage.completionTokens`
`ai.toolCall`	Individual tool executions	`ai.toolCall.name`, `ai.toolCall.args`, `ai.toolCall.result`
`ai.stream`	Streaming response metrics	`ai.response.msToFirstChunk`, `ai.response.msToFinish`, `ai.response.avgCompletionTokensPerSecond`

Span attributes

Each span contains detailed attributes: Model information: ai.model.id (e.g., "gpt-4o-mini"), ai.model.provider (e.g., "openai") Token usage: ai.usage.promptTokens, ai.usage.completionTokens Telemetry metadata: ai.telemetry.functionId, ai.telemetry.metadata.* Response details: ai.response.text, ai.response.toolCalls, ai.response.finishReason

Grouping operations into a span

Use span() (TypeScript) or span_context() (Python) to wrap a block of work as a single parent span. Nested SDK calls automatically attach as children.

TypeScript
Python

import { span } from "@agentmark-ai/sdk";

const { result, traceId } = await span(
  { name: 'user-request-handler' },
  async (ctx) => {
    const prompt = await client.loadTextPrompt('handler.prompt.mdx');
    const input = await prompt.format({
      props: { query: 'What is AgentMark?' },
      telemetry: { isEnabled: true }
    });

    return await generateText(input);
  }
);

console.log('Trace ID:', traceId);
const output = await result;

from agentmark_sdk import span_context, SpanOptions
from agentmark_pydantic_ai_v0 import run_text_prompt
from agentmark_client import client

async with span_context(SpanOptions(name="user-request-handler")) as ctx:
    prompt = await client.load_text_prompt("handler.prompt.mdx")
    params = await prompt.format(
        props={"query": "What is AgentMark?"},
        telemetry={"isEnabled": True},
    )

    result = await run_text_prompt(params)

print(f"Trace ID: {ctx.trace_id}")

SpanOptions

Option	Type	Required	Description
`name`	`string`	Yes	Name for the span
`metadata`	`Record<string, string>`	No	Custom key-value metadata (strings only)
`sessionId`	`string`	No	Group traces into a session
`sessionName`	`string`	No	Human-readable session name
`userId`	`string`	No	Associate trace with a user
`datasetRunId`	`string`	No	Link to a dataset run
`datasetRunName`	`string`	No	Human-readable dataset run name
`datasetItemName`	`string`	No	Specific dataset item name
`datasetExpectedOutput`	`string`	No	Expected output for evaluation
`datasetPath`	`string`	No	Path to the dataset file

SpanResult

TypeScript
Python

interface SpanResult<T> {
  result: Promise<T>;  // The result of your callback (as a Promise)
  traceId: string;     // The trace ID for correlation
}

result is Promise<T>, not T. You need to await it to get the resolved value.

async with span_context(SpanOptions(name="my-operation")) as ctx:
    print(ctx.trace_id)  # Available immediately
    result = await my_async_function()

Creating child spans

Use ctx.span() inside a callback to create child spans under the current parent:

TypeScript
Python

import { span } from "@agentmark-ai/sdk";

const { result, traceId } = await span(
  { name: 'multi-step-workflow' },
  async (ctx) => {
    await ctx.span({ name: 'validate-input' }, async (spanCtx) => {
      spanCtx.setAttribute('input.length', 42);
    });

    const output = await ctx.span({ name: 'process-request' }, async (spanCtx) => {
      const prompt = await client.loadTextPrompt('process.prompt.mdx');
      const input = await prompt.format({
        props: { query: 'process this' },
        telemetry: { isEnabled: true }
      });
      return await generateText(input);
    });

    await ctx.span({ name: 'format-response' }, async (spanCtx) => {
      spanCtx.addEvent('formatting-complete');
    });

    return output;
  }
);

ctx.span() accepts { name: string; metadata?: Record<string, string> }. Use observe() (below) if you need to set a SpanKind on a span.

from agentmark_sdk import span_context, SpanOptions
from agentmark_pydantic_ai_v0 import run_text_prompt
from agentmark_client import client

async with span_context(SpanOptions(name="multi-step-workflow")) as ctx:
    async with ctx.span("validate-input") as span_ctx:
        span_ctx.set_attribute("input.length", 42)

    async with ctx.span("process-request") as span_ctx:
        prompt = await client.load_text_prompt("process.prompt.mdx")
        params = await prompt.format(
            props={"query": "process this"},
            telemetry={"isEnabled": True},
        )
        output = await run_text_prompt(params)

    async with ctx.span("format-response") as span_ctx:
        span_ctx.add_event("formatting-complete")

Wrapping functions with `observe()`

observe() wraps an async function with automatic input/output capture AND lets you set a SpanKind. Unlike span() / ctx.span() which create inline spans, observe() produces a reusable function so every call is automatically traced.

TypeScript
Python

import { observe, SpanKind } from "@agentmark-ai/sdk";

const searchWeb = observe(
  async (query: string) => {
    const response = await fetch(`https://api.search.com?q=${query}`);
    return response.json();
  },
  { name: "search-web", kind: SpanKind.TOOL }
);

// Every call is now automatically traced
const results = await searchWeb("AgentMark tracing");

from agentmark_sdk import observe, SpanKind

@observe(name="search-web", kind=SpanKind.TOOL)
async def search_web(query: str) -> dict:
    async with httpx.AsyncClient() as http:
        response = await http.get(f"https://api.search.com?q={query}")
        return response.json()

# Every call is now automatically traced
results = await search_web("AgentMark tracing")

`observe()` options

Option	Type	Description
`name`	`string`	Display name for the span (defaults to function name)
`kind`	`SpanKind`	Type of operation (defaults to `SpanKind.FUNCTION`)
`captureInput` / `capture_input`	`boolean`	Record function arguments (default: `true`)
`captureOutput` / `capture_output`	`boolean`	Record return value (default: `true`)
`processInputs` / `process_inputs`	`function`	Transform arguments before recording (useful for redacting sensitive data)
`processOutputs` / `process_outputs`	`function`	Transform return value before recording

Observed functions automatically attach to the active trace context — they nest correctly inside span() / span_context() without extra wiring.

SpanKind values

Kind	Description
`SpanKind.FUNCTION`	Generic computation step (default)
`SpanKind.LLM`	A call to a language model
`SpanKind.TOOL`	An external tool or API call
`SpanKind.AGENT`	An orchestration loop
`SpanKind.RETRIEVAL`	A vector database query or document search
`SpanKind.EMBEDDING`	A call to an embedding model
`SpanKind.GUARDRAIL`	A content safety or validation check

Using `SpanKind` in a pipeline

To set SpanKind on individual steps of a pipeline, wrap each step with observe() and call the wrapped functions inside span():

TypeScript
Python

import { span, observe, SpanKind } from "@agentmark-ai/sdk";

const searchKB = observe(
  async (query: string) => vectorDb.query({ query, topK: 5 }),
  { name: 'search-knowledge-base', kind: SpanKind.RETRIEVAL }
);

const guardrail = observe(
  async (question: string) => moderationService.check(question),
  { name: 'check-content-policy', kind: SpanKind.GUARDRAIL }
);

const generateAnswer = observe(
  async (question: string, context: unknown) => {
    const prompt = await client.loadTextPrompt('answer.prompt.mdx');
    const input = await prompt.format({
      props: { question, context },
      telemetry: { isEnabled: true }
    });
    return generateText(input);
  },
  { name: 'generate-answer', kind: SpanKind.LLM }
);

const { result } = await span(
  { name: 'rag-pipeline' },
  async () => {
    const docs = await searchKB(userQuestion);
    await guardrail(userQuestion);
    return generateAnswer(userQuestion, docs);
  }
);

from agentmark_sdk import span_context, SpanOptions, observe, SpanKind

@observe(name="search-knowledge-base", kind=SpanKind.RETRIEVAL)
async def search_kb(query: str):
    return await vector_db.query(query=query, top_k=5)

@observe(name="check-content-policy", kind=SpanKind.GUARDRAIL)
async def guardrail(question: str):
    await moderation_service.check(question)

@observe(name="generate-answer", kind=SpanKind.LLM)
async def generate_answer(question: str, context) -> str:
    prompt = await client.load_text_prompt("answer.prompt.mdx")
    params = await prompt.format(
        props={"question": question, "context": context},
        telemetry={"isEnabled": True},
    )
    return await run_text_prompt(params)

async with span_context(SpanOptions(name="rag-pipeline")):
    docs = await search_kb(user_question)
    await guardrail(user_question)
    answer = await generate_answer(user_question, docs)

Scoring traces

Use sdk.score() to attach quality scores to traces or spans:

TypeScript
Python

const { result, traceId } = await span(
  { name: 'scored-workflow' },
  async (ctx) => {
    return output;
  }
);

await sdk.score({
  resourceId: traceId,
  name: 'correctness',
  score: 0.95,
  label: 'correct',
  reason: 'Output matches expected result'
});

async with span_context(SpanOptions(name="scored-workflow")) as ctx:
    output = await my_async_function()

await sdk.score(
    resource_id=ctx.trace_id,
    name="correctness",
    score=0.95,
    label="correct",
    reason="Output matches expected result",
)

Best practices

Use meaningful function IDs — "customer-support-greeting", not "func1"
Add relevant metadata — userId, environment, query parameters
Always enable telemetry in production — monitor performance and set up alerts
Shutdown tracer for short scripts — call tracer.shutdown() before the process exits

Next steps

Sessions

Group related traces together

Metadata

Add custom context to traces

PII masking

Redact sensitive data from traces

Have Questions?

We’re here to help! Choose the best way to reach us:

Email us at hello@agentmark.co for support
Schedule an Enterprise Demo to learn about our business solutions

Introduction

Getting Started

Build

Evaluate

Observe

Configure

Deploy

Integrations

Install the SDK

Initialize tracing

Collected spans

Span attributes

Grouping operations into a span

SpanOptions

SpanResult

Creating child spans

Wrapping functions with `observe()`

`observe()` options

SpanKind values

Using `SpanKind` in a pipeline

Scoring traces

Best practices

Next steps

Sessions

Metadata

Tags

PII masking

Have Questions?

Introduction

Getting Started

Build

Evaluate

Observe

Configure

Deploy

Integrations

Documentation Index

​Install the SDK

​Initialize tracing

​Collected spans

​Span attributes

​Grouping operations into a span

​SpanOptions

​SpanResult

​Creating child spans

​Wrapping functions with observe()

​observe() options

​SpanKind values

​Using SpanKind in a pipeline

​Scoring traces

​Best practices

​Next steps

Sessions

Metadata

Tags

PII masking

​Have Questions?

Install the SDK

Initialize tracing

Collected spans

Span attributes

Grouping operations into a span

SpanOptions

SpanResult

Creating child spans

Wrapping functions with `observe()`

`observe()` options

SpanKind values

Using `SpanKind` in a pipeline

Scoring traces

Best practices

Next steps

Have Questions?