Tracing - AgentX

The Tracer captures your agent’s inputs, outputs, latency, and tool calls, with a single decorator or context manager. Traces appear in the Live Traces tab and can be evaluated against your test datasets.

Running self-hosted? If your app is already instrumented with OpenTelemetry, you don’t need this SDK to get traces in at all - point your OTel exporter straight at the engine. See Connect via OpenTelemetry.

Decorator

The simplest usage. AgentX captures function arguments as input, the return value as output, and wall-clock time as latencyMs.

from agentx import AgentX

client = AgentX.from_env()
tracer = client.tracer

@tracer.trace("customer-support-agent", framework="langchain", model="gpt-4o")
def handle(query: str) -> str:
    return chain.invoke(query)

handle("How do I reset my password?")

Works with async functions too:

@tracer.trace("async-agent", framework="openai-agents")
async def handle_async(query: str) -> str:
    result = await runner.run(agent, query)
    return result.final_output

Context manager

Use when you need to set input/output manually or record tool calls mid-span.

with tracer.trace("rag-agent", framework="langchain") as span:
    span.input = {"query": query, "user_id": user_id}

    kb_result = search_knowledge_base(query)
    span.add_tool_call("search_knowledge_base", input=query, output=kb_result, latency_ms=190)

    answer = llm.invoke(f"Context: {kb_result}\n\nQuery: {query}")
    span.output = answer

To time a tool call and capture its failures automatically, wrap the execution itself with tracer.trace_tool_call() instead of reporting it after the fact:

with tracer.trace_tool_call("search_knowledge_base", input=query) as t:
    t.output = search_knowledge_base(query)

Long-term-memory operations get the same treatment through tracer.trace_memory() / tracer.record_memory() - a span_kind="memory" child span with the read/write operation in its metadata, deliberately distinct from retrieval (see Span Kinds):

with tracer.trace_memory("user prefs", operation="read", query=user_id) as m:
    m.output = memory.search(user_id, question)

An exception escaping the block records the call as failed (success=False plus the error text - what Monitor’s built-in “Tool failure” check reads) and then propagates unchanged. To set the outcome yourself, use tracer.record_tool_call(name, input=..., output=..., success=False, error="...").

`_TraceSpan` attributes and methods

Name	Description
`span.input = value`	Override the captured input
`span.output = value`	Set the output (required in context manager mode)
`span.add_tool_call(name, *, input, output, latency_ms)`	Record a tool call made during this span
`span.set_error(message)`	Mark the span as failed with a custom error message
`span.child_span(name, *, input=, output=, model=, span_kind=, metadata=, start_time=, end_time=, duration_ms=, ...)`	Send one finished child-span row parented to this span - for callers with their own real per-step timing. Returns the child; its `.span_id` can parent grandchildren
`span.trace_id`	The ingested trace’s id, once the block has exited. Only populated when opened with `sync=True` (see below)

On a root span, sync=True covers the whole tree: child spans recorded inside the block (tool calls, LLM calls) are drained before the root is sent, so a read immediately after the block sees every span. Child-only spans keep their async fire-and-forget behavior; client.tracer.flush() remains available for manual control.

Decorator vs. context manager

Both forms capture a usable trace - the root span itself renders as a row on the Execution Timeline, with the wrapped call’s input/output, even when nothing more granular was recorded. They differ in what you get back:

	Decorator	Context manager
Can retrieve `trace_id`	No. The wrapped function returns its own value, not a span - even with `sync=True` there is no span object to read the id from	Yes, with `sync=True`
Manual enrichment	Partial. The tracer-level recorders (`tracer.record_tool_call()`, `tracer.record_retrieval()`, `tracer.trace_memory()`) called inside the wrapped function attach to the decorated span; only `span.add_tool_call()` is unavailable, since you hold no span object	Yes, call any of these anywhere inside the block
Best for	Fire-and-forget background tracing where nothing downstream needs the id in the same call	Anything that needs the id afterward, e.g. linking a trace to an evaluation result, or `evaluate_trace()`

If you don’t need the id back, the decorator is simplest. If you do, use the context manager with sync=True. It adds one blocking network round-trip (typically well under a second) in exchange for the id being ready the moment the with block exits.

Parallel work across threads

The active span is context-local (a ContextVar): asyncio tasks inherit a copy of their creator’s span stack, but a bare thread starts with an empty one. If you fan work out to a ThreadPoolExecutor (or any other thread) from inside a tracer.trace(...) block, worker threads don’t automatically see the span opened on the calling thread, so each LLM or tool call inside them would start its own independent trace instead of landing as a step on the parent span. Wrap each worker’s body in tracer.use_span(span) to attach it to the parent span for the duration of that block:

with client.tracer.trace("billing-dispute-orchestrator") as orch:
    orch.input = user_message

    # Policy and risk specialists run in parallel, each on its own thread.
    # `orch` was opened on the main thread, so each worker must explicitly
    # attach to it via `use_span`, otherwise these steps become their own
    # separate traces instead of landing on the orchestrator trace.
    def run_policy_specialist():
        with client.tracer.use_span(orch):
            return chain.invoke(policy_query, config={"callbacks": [handler]})

    def run_risk_specialist():
        with client.tracer.use_span(orch):
            return chain.invoke(risk_query, config={"callbacks": [handler]})

    with ThreadPoolExecutor(max_workers=2) as ex:
        fut_policy = ex.submit(run_policy_specialist)
        fut_risk = ex.submit(run_risk_specialist)
        policy_result = fut_policy.result()
        risk_result = fut_risk.result()

    orch.output = combine(policy_result, risk_result)

use_span is safe to call concurrently from multiple threads for the same span. Each thread pushes and pops on its own stack, so parallel workers don’t interfere with each other.

`tracer.trace()` parameters

Every parameter except input works in both decorator and context-manager mode - the decorator forwards sync, monitor, pattern_ids, agent_id, and span_kind to each call’s span the same way the context manager does. input is the exception: in decorator mode the captured function arguments always overwrite it.

Parameter	Type	Description
`name`	`str`	Agent or operation label shown in the UI. Traces resolve to one stable agent per distinct name
`input`	`Any`	Initial input for the span. In decorator mode the function’s arguments are captured automatically; in context-manager mode you can also assign `span.input` inside the block
`framework`	`str`	The platform label: `"langchain"`, `"crewai"`, `"openai-agents"`, or any custom string (`"my-inhouse-runner"`). Left unset, integrations stamp their own literal, and otherwise the SDK auto-detects the single orchestration framework imported in the process - ambiguous stays unlabeled
`model`	`str`	LLM model used, e.g. `"gpt-4o"`, `"claude-sonnet-4-6"`
`session_id`	`str`	Groups traces from the same user session or thread
`metadata`	`dict`	Arbitrary key-value metadata (not indexed; size capped by the Limits table)
`agent_id`	`str`	Pins this trace to an exact agent id (one you already know, e.g. from the dashboard) instead of resolving by `name`. Omit it and the trace resolves from `name` alone
`sync`	`bool`	`False` (default) queues the trace on a background thread. Never blocks, but `span.trace_id` stays `None`. `True` sends synchronously - on a root span the whole tree is drained first - so `span.trace_id` is populated once the block exits
`monitor`	`bool \| None`	Three states. Ingest-time checks run on every root trace either way; `True` is what makes `pattern_ids` scoping take effect. `False`: opt this trace out of every ingest-time check - pattern/built-in detection, online and custom evaluators, and topics - use it for traces produced inside evaluation runs, whose dataset judge already scores them, so nothing gets judged twice. `None` (default): the engine’s normal behavior.
`pattern_ids`	`list[str]`	Restricts detection to these pattern ids when `monitor=True`. Omit to run the full default sweep (built-ins plus every active pattern) instead.
`span_kind`	`str`	What kind of step this span is: `"agent"`, `"llm"`, `"tool"`, `"retrieval"`, `"chain"`, `"embedding"`, `"reranker"`, `"guardrail"`, `"evaluator"`, `"prompt"`, `"memory"`. Also accepted on `child_span()`. Optional - a span that says nothing is classified by the engine’s fallback rules. See Span Kinds.

Framework examples

The decorator shown here records one trace per call - simple and dependable. For the full execution tree (graph nodes, every LLM call, every tool call as its own timed span), use the per-framework integrations instead: a LangChain/LangGraph callback handler, an OpenAI Agents trace processor, a CrewAI/AutoGen/ADK hook, or MLflow OTLP export. Both approaches share the same tracer and the same dashboard views.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from agentx import AgentX

client = AgentX.from_env()

llm = ChatOpenAI(model="gpt-4o")

@client.tracer.trace("support-agent", framework="langchain", model="gpt-4o")
def handle(query: str) -> str:
    return llm.invoke([HumanMessage(content=query)]).content

handle("How do I reset my password?")
client.tracer.flush(timeout=10)

from crewai import Agent, Task, Crew
from agentx import AgentX

client = AgentX.from_env()

@client.tracer.trace("research-crew", framework="crewai")
def run_crew(topic: str) -> str:
    agent = Agent(role="Researcher", goal=f"Research {topic}", backstory="...")
    task = Task(description=f"Research {topic}", agent=agent, expected_output="...")
    crew = Crew(agents=[agent], tasks=[task])
    return str(crew.kickoff())

run_crew("What is the refund policy?")
client.tracer.flush(timeout=10)

import asyncio
from agents import Agent, Runner
from agentx import AgentX

client = AgentX.from_env()
agent = Agent(name="Support", instructions="You are a helpful support agent.")

@client.tracer.trace("openai-support-agent", framework="openai-agents", model="gpt-4o")
async def run(query: str) -> str:
    result = await Runner.run(agent, query)
    return result.final_output

asyncio.run(run("How do I cancel my subscription?"))
client.tracer.flush(timeout=10)

import anthropic
from agentx import AgentX

agentx_client = AgentX.from_env()
claude_client = anthropic.Anthropic()

@agentx_client.tracer.trace(
    "claude-agent", framework="anthropic", model="claude-sonnet-4-6"
)
def run(query: str) -> str:
    msg = claude_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": query}],
    )
    return msg.content[0].text


run("How do I cancel my subscription?")
agentx_client.tracer.flush(timeout=10)

import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from agentx import AgentX

client = AgentX.from_env()

agent = Agent(
    name="support_agent",
    model="gemini-3.5-flash",
    instruction="You are a helpful support agent.",
)
runner = Runner(
    agent=agent,
    app_name="support-app",
    session_service=InMemorySessionService(),
)

@client.tracer.trace("google-support-agent", framework="google-adk", model="gemini-3.5-flash")
async def run(query: str) -> str:
    session = runner.session_service.create_session_sync(
        app_name="support-app", user_id="user-1"
    )
    for event in runner.run(
        user_id="user-1",
        session_id=session.id,
        new_message=types.Content(role="user", parts=[types.Part(text=query)]),
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return ""

asyncio.run(run("How do I cancel my subscription?"))
client.tracer.flush(timeout=10)

Session grouping

Use session_id to link traces from the same user conversation:

import uuid

session_id = str(uuid.uuid4())

@tracer.trace("support-agent", session_id=session_id)
def handle(query: str) -> str:
    ...

handle("First question")
handle("Follow-up question")
# Both traces appear linked in the UI

Traces sent without a session_id each get their own auto-generated session, so passing it is only about grouping - never required. On self-host, sessions are a first-class surface: a Sessions view lists each conversation with turn counts and a conversation-level coherence score, and judge scorers and scorer groups can judge whole sessions (scope="session") once they go idle.

Error handling

Exceptions inside a traced function are captured as the error field and re-raised. The trace is still submitted:

@tracer.trace("my-agent")
def risky(query: str) -> str:
    raise ValueError("Model unavailable")

try:
    risky("test")
except ValueError:
    pass
# Trace submitted with error: "Model unavailable"

To set an error manually in context manager mode:

with tracer.trace("my-agent") as span:
    try:
        result = call_agent(query)
        span.output = result
    except Exception as e:
        span.set_error(str(e))
        raise

Flushing

Traces are sent in a background thread. Call flush() before process exit in scripts:

tracer.flush(timeout=5.0)

Evaluating a trace

Score a previously recorded trace against a dataset without re-running the agent:

result = tracer.evaluate_trace(
    trace_id="6876abc123def456789abc01",
    dataset_id="6876ddd222bbb333ccc444ee",
    question_index=0,  # hosted only - the self-host engine scores the trace against the dataset's own cases
)

print(result["rating"])        # 0-10
print(result["justification"]) # LLM explanation

This is for scoring a trace you already recorded, standalone. To capture a trace while an evaluation run is scoring your agent, so the run’s own results show a full Execution Timeline and not just a rating, see the next section instead.

Linking a trace to an evaluation result

client.evaluations.run(...).execute(your_fn) calls your_fn once per test case and scores whatever it returns. Wrap the call inside your_fn with tracer.trace(..., sync=True) and return span.trace_id alongside your output. The eval result gets linked to the real trace, so its row in the dashboard shows a “View trace” action opening the full Execution Timeline, not just the score:

from agentx import AgentX

client = AgentX.from_env()

def support_agent(case):
    with client.tracer.trace(
        "support-agent-call",
        input={"query": case.query},
        framework="openai",
        model="gpt-4o-mini",
        sync=True,       # required, the default async mode never returns a trace_id
        monitor=False,   # the run's judge already scores this case - skip ingest-time judging
    ) as span:
        response = call_llm(case.query)
        span.output = response

    return {
        "output": response,
        "trace_id": span.trace_id,   # <- links this result to the trace above
    }

report = (
    client.evaluations
    .run(dataset_id="...", subject={"kind": "custom_agent", "framework": "openai"})
    .execute(support_agent)
    .finalize()
    .analyze()
)

trace_id is a plain top-level key in the dict your function returns, in the same place as output and metadata. It works the same way if your function returns an EvaluationResult directly (set trace_id=span.trace_id on it) instead of a dict.

sync=True blocks until the trace is ingested (typically well under a second) before your function returns. For a high test-case count where that latency adds up, keep tracing async (the default) and skip the timeline for that run. You’ll still get the score.

Monitor

Ingest-time checks run automatically on every root trace you send - the built-in checks plus every active detection pattern, with no flag and no dashboard setup - and matches become triage-ready signals. The opt-outs are tracer.trace(..., monitor=False), which skips every ingest-time check for that trace, and disabling the agent’s monitoring profile. Passing monitor=True with pattern_ids=[...] restricts detection for that trace to exactly those patterns instead of the full sweep. See Monitor for the full picture.

Limits

Limit	Value
Tool calls per trace	No count limit - the list is bounded by the serialized-size cap below, with an explicit `agentx.truncated` marker; judges see the first 40 in the trajectory
Rate limiting	Per IP, per minute: 120 requests on credential routes (sign-in, key management), 6000 on the data plane (ingest and reads; `AGENTX_RATE_LIMIT_DATA_PLANE` tunes it). Exceeding a ceiling returns a 429 with standard draft-7 `RateLimit` headers - see Errors
`input` / `output` / `tool_calls` / `metadata` / `performance_summary` size	100,000 characters each, serialized (`AGENTX_INGEST_MAX_FIELD_CHARS` tunes it). An oversized field is stored truncated with an explicit `agentx.truncated` marker, never rejected

​Decorator

​Context manager

​_TraceSpan attributes and methods

​Decorator vs. context manager

​Parallel work across threads

​tracer.trace() parameters

​Framework examples

​Session grouping

​Error handling

​Flushing

​Evaluating a trace

​Linking a trace to an evaluation result

​Monitor

​Limits

Decorator

Context manager

`_TraceSpan` attributes and methods

Decorator vs. context manager

Parallel work across threads

`tracer.trace()` parameters

Framework examples

Session grouping

Error handling

Flushing

Evaluating a trace

Linking a trace to an evaluation result

Monitor

Limits