Add HoneyHive observability to your Claude Agent SDK applications
Claude Agent SDK is Anthropic’s Python SDK for building AI agents powered by Claude Code. It gives agents access to built-in tools like Bash, Read, Write, and Glob for autonomous task execution.HoneyHive integrates with the Claude Agent SDK via the OpenInference instrumentor, automatically capturing agent runs, tool calls, and multi-turn conversations.
import osfrom honeyhive import HoneyHiveTracerfrom openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentortracer = HoneyHiveTracer.init( api_key=os.getenv("HH_API_KEY"), project=os.getenv("HH_PROJECT"),)ClaudeAgentSDKInstrumentor().instrument(tracer_provider=tracer.provider)# Import claude_agent_sdk AFTER this setup block (see examples below)# Call tracer.force_flush() before your process exits
The query() function runs a one-off agent session. The agent can use built-in tools to complete tasks autonomously:
import asyncioimport osfrom openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentorfrom honeyhive import HoneyHiveTracertracer = HoneyHiveTracer.init( api_key=os.getenv("HH_API_KEY"), project=os.getenv("HH_PROJECT"),)ClaudeAgentSDKInstrumentor().instrument(tracer_provider=tracer.provider)# Import AFTER instrument() so references resolve to the patched versionsfrom claude_agent_sdk import AssistantMessage, ClaudeAgentOptions, TextBlock, queryasync def main(): async for message in query( prompt="Read the file config.json and summarize its contents.", options=ClaudeAgentOptions( system_prompt="You are a helpful assistant. Complete tasks concisely.", allowed_tools=["Read"], max_turns=3, permission_mode="bypassPermissions", ), ): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(block.text) tracer.force_flush()asyncio.run(main())
permission_mode="bypassPermissions" allows unrestricted tool use without interactive prompts. Use only in sandboxed or non-interactive environments.
ClaudeSDKClient maintains session continuity across multiple turns. The agent remembers context from previous interactions:
import asyncioimport osfrom openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentorfrom honeyhive import HoneyHiveTracertracer = HoneyHiveTracer.init( api_key=os.getenv("HH_API_KEY"), project=os.getenv("HH_PROJECT"),)ClaudeAgentSDKInstrumentor().instrument(tracer_provider=tracer.provider)# Import AFTER instrument() so references resolve to the patched versionsfrom claude_agent_sdk import ( AssistantMessage, ClaudeAgentOptions, ClaudeSDKClient, TextBlock,)async def main(): client = ClaudeSDKClient( options=ClaudeAgentOptions( system_prompt="You are a customer support agent. Keep responses concise.", allowed_tools=["Read"], max_turns=3, permission_mode="bypassPermissions", ) ) await client.connect() try: # Turn 1 await client.query(prompt="Read orders.json and tell me the status of ORD-1002.") async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(block.text) # Turn 2 - agent remembers context from Turn 1 await client.query(prompt="What about ORD-1003? Is it delayed?") async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(block.text) finally: await client.disconnect() tracer.force_flush()asyncio.run(main())
In HoneyHive, you’ll see each turn as a separate agent span with tool calls nested underneath.
Import after instrumenting - Import claude_agent_sdkafter calling instrument(). The instrumentor patches module-level functions, so importing before patching captures the original unpatched references:
# Correct - import after instrument()ClaudeAgentSDKInstrumentor().instrument(tracer_provider=tracer.provider)from claude_agent_sdk import query# Wrong - import before instrument()from claude_agent_sdk import queryClaudeAgentSDKInstrumentor().instrument(tracer_provider=tracer.provider)
Flush before exit - Call tracer.force_flush() before your process exits. Without this, buffered spans may be silently dropped in short-lived scripts:
async def main(): # ... your agent code ... tracer.force_flush()asyncio.run(main())
Check Anthropic credentials - Ensure ANTHROPIC_API_KEY is set