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)

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.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, including at least one Execution Timeline step synthesized from the wrapped call’s input/output if nothing more granular was recorded. They differ in what you get back:

	Decorator	Context manager
Sends the trace	Always fire-and-forget (background thread)	Fire-and-forget by default, or blocking with `sync=True`
Can retrieve `trace_id`	No. The wrapped function returns its own value, not a span	Yes, with `sync=True`
Manual enrichment (`add_tool_call`, `record_retrieval`, multiple steps)	No, one function call produces one synthesized step	Yes, call 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 tracked per-thread. 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

Parameter	Type	Description
`name`	`str`	Agent or operation label shown in the UI
`framework`	`str`	`"langchain"`, `"crewai"`, `"openai-agents"`, `"anthropic"`, or custom
`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, max 16 KB)
`sync`	`bool`	Context manager only. `False` (default) queues the trace on a background thread. Never blocks, but `span.trace_id` stays `None`. `True` sends synchronously so `span.trace_id` is populated once the block exits.
`monitor`	`bool \| None`	Three states. `True`: check this trace against Monitor patterns immediately. `False`: opt this trace out of every ingest-time check (patterns, online evaluators, 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.

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 online evaluators 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,
)

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

Every trace you send can also be checked against detection patterns and turned into a triage-ready signal, either immediately via tracer.trace(..., monitor=True, pattern_ids=[...]) with no dashboard setup, or automatically for every trace once a monitoring profile is enabled for the agent in Governance > Observe > Agents. See Monitor for the full picture.

Limits

Limit	Value
Max tool calls per trace	50 (excess calls are silently truncated)
Rate limiting	None currently enforced per API key
`input` / `output` / `metadata` size	No application-level limit; bounded by the backing store’s row limits (hosted: 16 MB)

​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