Python auto-instrumentation for the Claude Agent SDK (Python). Traces query() and ClaudeSDKClient as OpenInference AGENT spans with prompt input, result output, session/model metadata, token counts, and tool child spans via hook injection.
query() – One span per call (one-off sessions).ClaudeSDKClient – One span per response turn: each time you iterate receive_response(), a span is created for that turn. Use for continuous conversations.Task is grouped under a nested AGENT span, with the subagent’s own tool calls as its children.For detailed LLM and tool spans inside agent runs, use openinference-instrumentation-anthropic together with this package; the Agent SDK uses the Anthropic API under the hood.
Traces are OpenTelemetry-compatible and can be sent to any OTLP collector, Arize Phoenix (local), Phoenix Cloud, or Arize AX.
pip install openinference-instrumentation-claude-agent-sdk
pip install openinference-instrumentation-claude-agent-sdk claude-agent-sdk arize-phoenix opentelemetry-sdk opentelemetry-exporter-otlp
Option A – Remote Phoenix: Set PHOENIX_COLLECTOR_ENDPOINT to your collector endpoint (e.g. https://<host>/v1/traces). If auth is enabled on that Phoenix (including Phoenix Cloud), also set PHOENIX_API_KEY; the snippet below sends it as a bearer token.
Option B – Local Phoenix: Start Phoenix, then run your script:
python -m phoenix.server.main serve
Then in Python:
import asyncio
import os
from claude_agent_sdk import query, ClaudeAgentOptions
from openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk import trace as trace_sdk
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
# Remote Phoenix: set PHOENIX_COLLECTOR_ENDPOINT, plus PHOENIX_API_KEY if auth is enabled. Defaults to local Phoenix.
endpoint = os.environ.get("PHOENIX_COLLECTOR_ENDPOINT", "http://127.0.0.1:6006/v1/traces")
api_key = os.environ.get("PHOENIX_API_KEY")
headers = {"authorization": f"Bearer {api_key}"} if api_key else None
tracer_provider = trace_sdk.TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint, headers=headers)))
ClaudeAgentSDKInstrumentor().instrument(tracer_provider=tracer_provider)
async def main():
async for message in query(
prompt="What files are in this directory?",
options=ClaudeAgentOptions(allowed_tools=["Bash", "Glob"]),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(main())
View traces in Phoenix Cloud, at http://localhost:6006 when running Phoenix locally, or in Arize AX.
Run the example in this repo from the package directory:
pip install -r examples/requirements.txt
export ANTHROPIC_API_KEY=your-key
python examples/example.py
The example always exports spans over OTLP, defaulting to a local Phoenix at http://127.0.0.1:6006 (start it first, or set PHOENIX_COLLECTOR_ENDPOINT to another Phoenix and, if it has auth enabled, PHOENIX_API_KEY). See examples/README.md for what the example does.
query() – Each call is wrapped in a single AGENT span named ClaudeAgentSDK.query with:
llm.output_messages including any tool callssession.id, llm.model_name, llm.finish_reason, llm.provider/llm.system (anthropic), token counts (prompt, completion, total, cache read/write), and llm.cost.total when availabletool.name, input parameters, and outputClaudeAgentSDK.<tool> (e.g. ClaudeAgentSDK.Task) with agent.name set, parenting the subagent’s TOOL spansClaudeSDKClient – For multi-turn conversations:
connect(prompt=...) and query(prompt) record the prompt for the next response.receive_response() iteration is wrapped in an AGENT span named ClaudeAgentSDK.ClaudeSDKClient.receive_response with the same input/output/metadata/tool/subagent spans as above.receive_messages() is not wrapped; use receive_response() to get a span per turn.LLM spans for the SDK’s internal Anthropic API calls are not created by this package; add openinference-instrumentation-anthropic and instrument Anthropic for that.