Documentation

Claude Code (Python)

Wrap the Claude Agent SDK's query() or ClaudeSDKClient to automatically track sessions, tool calls, tokens, costs, and conversation threading.

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.

Installation

pip install guardy claude-agent-sdk httpx

Quick Start

Wrap the query() function once, then use it exactly like normal.

from claude_agent_sdk import query
from guardy import AsyncGuardyClient
from guardy.claude_code import wrap_claude_agent

# 1. Create a Guardy client
guardy = AsyncGuardyClient(api_key="guardy_live_xxx")

# 2. Wrap query()
tracked_query = wrap_claude_agent(
    query,
    client=guardy,
    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 guardy.close()

Configuration Options

tracked_query = wrap_claude_agent(
    query,
    client=guardy,            # required — AsyncGuardyClient 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",
    },
)

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=guardy,
    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 guardy import AsyncGuardyClient
from guardy.claude_code import wrap_claude_client

guardy = AsyncGuardyClient(api_key="guardy_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,
    guardy_client=guardy,
    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 Guardy session, same convo_id
    await tracked.query("Add rate limiting to /api/users")
    async for msg in tracked.receive_response():
        print(msg)

await guardy.close()

Each query() + receive_response() cycle creates a separate Guardy 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 Guardy
    # with failure_reason = str(error)
    print(f"Agent failed: {error}")

Full Production Example

import asyncio
import uuid
from claude_agent_sdk import query
from guardy import AsyncGuardyClient
from guardy.claude_code import wrap_claude_agent


async def run_agent(user_prompt: str, user_id: str) -> str:
    guardy = AsyncGuardyClient(
        api_key="guardy_live_xxx",
        fail_silently=True,
    )

    convo_id = f"session-{user_id}-{uuid.uuid4().hex[:8]}"

    tracked_query = wrap_claude_agent(
        query,
        client=guardy,
        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 guardy.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.