> ## Documentation Index
> Fetch the complete documentation index at: https://www.sentrial.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Claude Code (Python)
> Wrap the Claude Agent SDK's query() or ClaudeSDKClient to automatically track sessions, tool calls, tokens, costs, and conversation threading.
**Requires:** `claude-agent-sdk` v0.1.0+ and `httpx` as peer dependencies. Works with both `query()` (one-shot) and `ClaudeSDKClient` (multi-turn conversations).
## What Gets Tracked Automatically
Created on init, completed on result — with agent name, user ID, and metadata.
Every tool execution (Bash, Read, Write, etc.) with input, output, and tool use ID.
Prompt tokens, completion tokens, total tokens, and estimated cost in USD.
Group related sessions with `convo_id` — multi-turn conversations appear linked.
Failed sessions and generator exceptions — all recorded with failure reasons.
Wall-clock time from start to result, plus API-side duration when available.
Both `wrap_claude_agent` (for `query()`) and `wrap_claude_client` (for `ClaudeSDKClient`) require
consuming the stream until the `result` message to record final tokens/cost/output. If you close the
generator early, the session is marked failed with `Session interrupted (generator closed early)`.
## Installation
```bash theme={null}
pip install sentrial claude-agent-sdk httpx
```
## Quick Start
Wrap the `query()` function once, then use it exactly like normal.
```python theme={null}
from claude_agent_sdk import query
from sentrial import AsyncSentrialClient
from sentrial.claude_code import wrap_claude_agent
# 1. Create a Sentrial client
sentrial = AsyncSentrialClient(api_key="sentrial_live_xxx")
# 2. Wrap query()
tracked_query = wrap_claude_agent(
query,
client=sentrial,
default_agent="my-agent",
user_id="user-123",
)
# 3. Use normally — everything is tracked
async for message in tracked_query(
prompt="Fix the failing tests in src/auth.py",
options={"max_turns": 10, "permission_mode": "bypassPermissions"},
):
if message.type == "result":
print(message.result)
# Session auto-completed with tokens, cost, duration, and all tool calls
await sentrial.close()
```
## Configuration Options
```python theme={null}
tracked_query = wrap_claude_agent(
query,
client=sentrial, # required — AsyncSentrialClient instance
default_agent="my-agent", # optional — agent name for grouping (default: "claude-agent")
user_id="user-123", # optional — ties sessions to your end users (default: "anonymous")
convo_id="convo-abc", # optional — group sessions into a conversation thread
extra_metadata={ # optional — merged into every session's metadata
"environment": "production",
"version": "1.2.0",
},
)
```
**Fail-Safe by Default** — If Sentrial is unreachable, all messages pass through unchanged. Your agent never breaks due to tracking failures.
## Conversation Threading
Use `convo_id` to group related sessions into a conversation. Each call to `tracked_query()` creates a new session, but they appear linked in the dashboard.
```python theme={null}
convo_id = f"user-{user_id}-{uuid.uuid4().hex[:8]}"
tracked_query = wrap_claude_agent(
query,
client=sentrial,
default_agent="code-assistant",
user_id=user_id,
convo_id=convo_id,
)
options = {"max_turns": 10, "permission_mode": "bypassPermissions"}
# Session 1: Explore the codebase
async for msg in tracked_query(
prompt="List all API routes and their handlers",
options=options,
):
pass
# Session 2: Follow-up (same convo_id, linked in dashboard)
async for msg in tracked_query(
prompt="Add rate limiting to the /api/users endpoint",
options=options,
):
pass
```
## Using with ClaudeSDKClient
If you use `ClaudeSDKClient` for multi-turn conversations (instead of `query()`), use `wrap_claude_client`:
```python theme={null}
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from sentrial import AsyncSentrialClient
from sentrial.claude_code import wrap_claude_client
sentrial = AsyncSentrialClient(api_key="sentrial_live_xxx")
options = ClaudeAgentOptions(
permission_mode="bypassPermissions",
allowed_tools=["Bash", "Read", "Write"],
)
sdk_client = ClaudeSDKClient(options=options)
# Wrap the client — all receive_response() calls are tracked
tracked = wrap_claude_client(
sdk_client,
sentrial_client=sentrial,
default_agent="code-assistant",
user_id="user-123",
convo_id="convo-abc",
)
async with tracked:
# Turn 1
await tracked.query("List all API routes")
async for msg in tracked.receive_response():
print(msg)
# Turn 2 — new Sentrial session, same convo_id
await tracked.query("Add rate limiting to /api/users")
async for msg in tracked.receive_response():
print(msg)
await sentrial.close()
```
Each `query()` + `receive_response()` cycle creates a separate Sentrial session. All sessions share the same `convo_id` so they appear linked in the dashboard.
## How Tool Tracking Works
The wrapper observes the message stream to detect tool calls. When an `AssistantMessage` contains a `ToolUseBlock`, the wrapper captures the tool name and input. When the following `UserMessage` arrives with tool results, it matches them and records the tool call event. This approach is reliable across all SDK versions.
```python theme={null}
# No extra code needed — tool calls are tracked automatically
options = {"max_turns": 5, "allowed_tools": ["Bash", "Read", "Glob"]}
async for message in tracked_query(
prompt="Read package.json and tell me the project name",
options=options,
):
# Each tool call (Read, Bash, etc.) is recorded as an event
# with input args, output, and execution metadata
pass
```
In the dashboard, each tool call appears as an event under the session timeline.
## Error Handling
If the agent errors, the generator throws, or the consumer closes the stream early, the session is
automatically marked as failed with a clear failure reason.
```python theme={null}
try:
async for message in tracked_query(prompt="Deploy to production"):
pass
except Exception as error:
# Session already marked as failed in Sentrial
# with failure_reason = str(error)
print(f"Agent failed: {error}")
```
## Full Production Example
```python theme={null}
import asyncio
import uuid
from claude_agent_sdk import query
from sentrial import AsyncSentrialClient
from sentrial.claude_code import wrap_claude_agent
async def run_agent(user_prompt: str, user_id: str) -> str:
sentrial = AsyncSentrialClient(
api_key="sentrial_live_xxx",
fail_silently=True,
)
convo_id = f"session-{user_id}-{uuid.uuid4().hex[:8]}"
tracked_query = wrap_claude_agent(
query,
client=sentrial,
default_agent="code-review-agent",
user_id=user_id,
convo_id=convo_id,
extra_metadata={"environment": "production"},
)
options = {
"model": "claude-sonnet-4-20250514",
"max_turns": 15,
"permission_mode": "bypassPermissions",
"allowed_tools": ["Bash", "Read", "Write", "Glob", "Grep"],
}
result = ""
async for message in tracked_query(prompt=user_prompt, options=options):
if message.type == "result":
result = getattr(message, "result", "") or ""
await sentrial.close()
return result
# Run it
response = asyncio.run(run_agent("Review the auth module for security issues", "user-42"))
print(response)
```
## What You See in the Dashboard
Agent name, user ID, conversation thread, status, duration, cost.
The prompt and Claude's final response.
Every Bash, Read, Write, Glob call — with input, output, and tool use ID.
Prompt tokens, completion tokens, total, estimated USD cost.
## Next Steps
Same integration for the TypeScript Claude Agent SDK.
Full SDK documentation with all methods and options.