Anthropic

Install the integration extra:

pip install "agentx-python[anthropic]"

Usage

Call patch_anthropic_client() once after creating your Anthropic client. All subsequent client.messages.create() and client.messages.stream() calls are traced automatically. No changes to individual API calls are needed. Works with both anthropic.Anthropic and anthropic.AsyncAnthropic - pass whichever client you use, sync or async, streaming or not.

from agentx import AgentX
from agentx.integrations.anthropic import patch_anthropic_client
import anthropic

agentx = AgentX.from_env()
client = anthropic.Anthropic()

patch_anthropic_client(
    client,
    tracer=agentx.tracer,
    name="claude-support-agent",
    metadata={"env": "production"},
    session_id="session-xyz-789",
)

# Regular call, traced automatically
response = client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=256,
    messages=[{"role": "user", "content": "How do I cancel my subscription?"}],
)

# Streaming call, also traced automatically
with client.messages.stream(
    model="claude-haiku-4-5-20251001",
    max_tokens=256,
    messages=[{"role": "user", "content": "What is your refund policy?"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

What gets traced

By default, each messages.create() or messages.stream() call produces its own trace.

Field	Source
`input`	`messages` kwarg, with `system` prepended as a leading `{"role": "system", ...}` entry when set
`output`	`response.content[0].text` (or streamed final message text)
`latencyMs`	Wall-clock time of the API call - measured after the real response comes back, for both sync and async clients
`model`	`model` kwarg
`inputTokens`	`response.usage.input_tokens`, plus `cache_creation_input_tokens` / `cache_read_input_tokens` when prompt caching is used
`error`	Exception message if the API call raises

The raw Anthropic SDK has no built-in concept of “tool call” or “retrieval”; those only exist as plain Python code around your messages.create() calls, so the patch can’t see them on its own. Use tracer.trace_tool_call() and tracer.trace_retrieval() (below) to record them manually so they show up in the trace’s performance summary.

Tool use

Anthropic tool use is a manual loop: you call messages.create(), execute whatever tool the model requested yourself, then call messages.create() again with the result. The tool execution happens in plain Python between the two API calls, so wrap it in tracer.trace_tool_call() to record it:

from agentx import AgentX
from agentx.integrations.anthropic import patch_anthropic_client
import anthropic

agentx = AgentX.from_env()
client = anthropic.Anthropic()

patch_anthropic_client(client, tracer=agentx.tracer, name="claude-support-agent-tool")


def policy_lookup(topic: str) -> str:
    """Look up a company policy by topic."""
    db = {
        "cancel": "Go to Account → Subscription → Cancel.",
        "trial": "14-day free trial, no credit card required.",
        "refund": "Full refund within 30 days.",
    }
    for key, val in db.items():
        if key in topic.lower():
            return val
    return "No policy found."


tools = [
    {
        "name": "policy_lookup",
        "description": "Look up a company policy by topic.",
        "input_schema": {
            "type": "object",
            "properties": {"topic": {"type": "string"}},
            "required": ["topic"],
        },
    }
]

# Dispatch table. Don't hardcode a single tool name. Any tool the model
# calls gets looked up here and traced by its real name.
TOOL_REGISTRY = {"policy_lookup": policy_lookup}

messages = [{"role": "user", "content": "How do I cancel my subscription?"}]

# Wrap the whole loop in one span so every messages.create() call below
# collapses into a single trace instead of one trace per call, see
# "Multi-call agentic loops" below.
with agentx.tracer.trace("claude-support-agent-tool", framework="anthropic") as span:
    span.input = messages[0]["content"]

    while True:
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            tools=tools,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            break

        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue

            fn = TOOL_REGISTRY.get(block.name)
            with agentx.tracer.trace_tool_call(block.name, input=block.input) as t:
                result = fn(**block.input) if fn else f"Unknown tool: {block.name}"
                t.output = result

            tool_results.append(
                {"type": "tool_result", "tool_use_id": block.id, "content": result}
            )

        messages.append({"role": "user", "content": tool_results})

    for block in response.content:
        if block.type == "text":
            span.output = block.text
            print(block.text)

agentx.tracer.flush(timeout=10)

`tracer.trace_tool_call()`

tracer.trace_tool_call(name: str, *, input: Any = None) -> contextmanager

Times the block automatically. Assign t.output before the block exits.

Parameter	Description
`name`	Tool name, shown in the trace’s tool calls
`input`	The arguments passed to the tool

For cases where you already have the timing and result computed, use tracer.record_tool_call() directly instead of the context manager:

tracer.record_tool_call(
    name: str,
    *,
    input: Any = None,
    output: Any = None,
    latency_ms: int | None = None,
) -> None

Retrieval-augmented generation (RAG)

Like tool calls, a hand-rolled retrieval step (vector search, keyword lookup, etc.) run before messages.create() is invisible to the patch. Record it with tracer.trace_retrieval():

from agentx import AgentX
from agentx.integrations.anthropic import patch_anthropic_client
import anthropic

agentx = AgentX.from_env()
client = anthropic.Anthropic()

patch_anthropic_client(client, tracer=agentx.tracer, name="claude-support-agent-rag")

knowledge_base = [
    "To cancel your subscription, go to Account → Subscription → Cancel.",
    "We offer a 14-day free trial, no credit card required.",
    "Full refunds are available within 30 days of purchase.",
]


def retrieve(query: str, k: int = 2) -> list[str]:
    query_words = set(query.lower().split())
    score = lambda doc: len(query_words & set(doc.lower().split()))
    return sorted(knowledge_base, key=score, reverse=True)[:k]


question = "How do I cancel my subscription?"

with agentx.tracer.trace_retrieval("kb_search", query=question) as r:
    docs = retrieve(question)
    r.doc_count = len(docs)
    r.output = docs  # the retrieved chunks - what RAG judges grade against

context = "\n".join(f"- {doc}" for doc in docs)

response = client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=256,
    system="Answer using only the provided context.",
    messages=[
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
    ],
)
print(response.content[0].text)

agentx.tracer.flush(timeout=10)

`tracer.trace_retrieval()`

tracer.trace_retrieval(name: str = "Retrieval", *, query: str | None = None) -> contextmanager

Times the block automatically. Assign r.doc_count and r.output inside the block.

Parameter	Description
`name`	Step name, shown in the trace’s timeline (any name works - retrieval spans carry an explicit kind marker)
`query`	The retrieval query, shown in the trace
`r.output`	Set inside the block: the retrieved chunks. This is what RAG judging reads as `{context}` - without it the retrieval is timed but contributes no chunks
`r.doc_count`	Set inside the block: how many documents came back

The context manager attaches the retrieval to whichever trace is sent next: either the current active span (see below) or the next standalone messages.create() call.

A retrieval or tool call recorded with no active span is queued and merged into the very next trace this tracer sends, so call trace_retrieval() / trace_tool_call() immediately before the messages.create() call it belongs to.

Multi-call agentic loops

Tool-use loops and multi-turn agents call messages.create() more than once. By default each call is its own independent trace. Wrap the whole loop in with tracer.trace(...) to collapse it into one trace instead: every messages.create() call made while that span is active is folded in as an "LLM Call N" step (in call order, interleaved with any trace_tool_call() steps):

with agentx.tracer.trace("claude-support-agent-tool", framework="anthropic") as span:
    span.input = question
    # ... call client.messages.create() as many times as needed ...
    span.output = final_answer

This works because patch_anthropic_client() checks tracer.current_span on every call: if a span is active on the current thread, the call is attached to it; otherwise it sends its own trace as usual. No other code changes are needed. The same patched client works standalone or inside a span.

Without the with tracer.trace(...) wrapper, a tool-use loop like the one above still produces a working trace per call, but the tool call recorded via trace_tool_call() attaches to whichever trace is sent next, which is the following turn, not the turn that requested the tool. Wrapping the loop in a span keeps everything on one trace, in the right order.

`patch_anthropic_client()` reference

patch_anthropic_client(
    client: anthropic.Anthropic,
    tracer: Tracer,
    name: str = "anthropic-agent",
    metadata: dict | None = None,
    session_id: str | None = None,
) -> None

Parameter	Description
`client`	The `anthropic.Anthropic()` instance to patch
`tracer`	`agentx.tracer` from your `AgentX` instance
`name`	Label shown in the AgentX UI for every trace
`metadata`	Static key-value metadata attached to every trace
`session_id`	Links traces from the same conversation thread

Calling patch_anthropic_client() on an already-patched client is a no-op; it is safe to call multiple times.

Full example

from agentx import AgentX
from agentx.integrations.anthropic import patch_anthropic_client
import anthropic

agentx = AgentX.from_env()
client = anthropic.Anthropic()

patch_anthropic_client(
    client,
    tracer=agentx.tracer,
    name="claude-support-agent",
    metadata={"env": "production"},
    session_id="session-xyz-789",
)

response = client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=256,
    messages=[{"role": "user", "content": "How do I cancel my subscription?"}],
)
print(response.content[0].text)

agentx.tracer.flush(timeout=10)

Tracing

Framework Integrations

Usage

What gets traced

Tool use

`tracer.trace_tool_call()`

Retrieval-augmented generation (RAG)

`tracer.trace_retrieval()`

Multi-call agentic loops

`patch_anthropic_client()` reference

Full example

​Usage

​What gets traced

​Tool use

​tracer.trace_tool_call()

​Retrieval-augmented generation (RAG)

​tracer.trace_retrieval()

​Multi-call agentic loops

​patch_anthropic_client() reference

​Full example

Usage

What gets traced

Tool use

`tracer.trace_tool_call()`

Retrieval-augmented generation (RAG)

`tracer.trace_retrieval()`

Multi-call agentic loops

`patch_anthropic_client()` reference

Full example