Skip to main content
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.
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=client.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

Each messages.create() or messages.stream() call produces one trace.
FieldSource
inputmessages kwarg
outputresponse.content[0].text (or streamed final message text)
latencyMsWall-clock time of the API call
modelmodel kwarg
errorException message if the API call raises

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
ParameterDescription
clientThe anthropic.Anthropic() instance to patch
traceragentx.tracer from your AgentX instance
nameLabel shown in the AgentX UI for every trace
metadataStatic key-value metadata attached to every trace
session_idLinks 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)