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

Sessions

Created on init, completed on result — with agent name, user ID, and metadata.

Tool Calls

Every tool execution (Bash, Read, Write, etc.) with input, output, and tool use ID.

Tokens & Cost

Prompt tokens, completion tokens, total tokens, and estimated cost in USD.

Conversation Threading

Group related sessions with convo_id — multi-turn conversations appear linked.

Errors

Failed sessions and generator exceptions — all recorded with failure reasons.

Duration

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

pip install sentrial claude-agent-sdk httpx

​
Quick Start

Wrap the query() function once, then use it exactly like normal. 
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

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. 
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: 
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. 
# 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. 
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

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

Session Overview

Agent name, user ID, conversation thread, status, duration, cost.

Input / Output

The prompt and Claude’s final response.

Tool Call Timeline

Every Bash, Read, Write, Glob call — with input, output, and tool use ID.

Token & Cost Breakdown

Prompt tokens, completion tokens, total, estimated USD cost.

​
Next Steps

Claude Code (TypeScript)

Same integration for the TypeScript Claude Agent SDK.

Python SDK Reference

Full SDK documentation with all methods and options.