Skip to main content

Anthropic Integration

Add observability to Anthropic Claude API calls with auto-instrumentation or manual context managers.

Quick Start​

import waxell_observe
waxell_observe.init(api_key="wax_sk_...", api_url="https://waxell.dev")

# Import Anthropic AFTER init()
import anthropic

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)
# Automatically traced with model, tokens, cost

Drop-in Import​

Alternative approach using pre-instrumented module:

from waxell_observe.anthropic import anthropic

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)
# Automatically traced

Context Manager Pattern​

For multi-step pipelines with full control:

import waxell_observe
waxell_observe.init(api_key="wax_sk_...")

from waxell_observe import WaxellContext, generate_session_id
from waxell_observe.errors import PolicyViolationError
import anthropic

client = anthropic.AsyncAnthropic()

async def analyze_content(query: str, user_id: str = ""):
    session = generate_session_id()

    async with WaxellContext(
        agent_name="anthropic-demo",
        workflow_name="content-analysis",
        inputs={"query": query},
        session_id=session,
        user_id=user_id,
        user_group="enterprise",
        enforce_policy=True,
        mid_execution_governance=True,
    ) as ctx:
        # Add tags for categorization
        ctx.set_tag("demo", "anthropic")
        ctx.set_tag("provider", "anthropic")
        ctx.set_metadata("sdk", "anthropic-python")

        try:
            # Step 1: Classify content
            classify_response = await client.messages.create(
                model="claude-sonnet-4",
                max_tokens=500,
                messages=[{
                    "role": "user",
                    "content": f"Classify this text: {query}"
                }],
            )
            classification = classify_response.content[0].text

            ctx.record_step("classify_content", output={
                "classification": classification[:200]
            })
            ctx.record_llm_call(
                model=classify_response.model,
                tokens_in=classify_response.usage.input_tokens,
                tokens_out=classify_response.usage.output_tokens,
                task="classify_content",
                prompt_preview=query[:200],
                response_preview=classification[:200],
            )

            # Step 2: Extract entities
            extract_response = await client.messages.create(
                model="claude-sonnet-4",
                max_tokens=500,
                messages=[{
                    "role": "user",
                    "content": f"Extract key entities from: {query}"
                }],
            )
            entities = extract_response.content[0].text

            ctx.record_step("extract_entities", output={
                "entities": entities[:200]
            })
            ctx.record_llm_call(
                model=extract_response.model,
                tokens_in=extract_response.usage.input_tokens,
                tokens_out=extract_response.usage.output_tokens,
                task="extract_entities",
            )

            # Step 3: Summarize
            summary_response = await client.messages.create(
                model="claude-sonnet-4",
                max_tokens=500,
                messages=[{
                    "role": "user",
                    "content": f"Summarize:\n\nText: {query}\nClassification: {classification}\nEntities: {entities}"
                }],
            )
            summary = summary_response.content[0].text

            ctx.record_step("summarize", output={
                "summary_length": len(summary)
            })
            ctx.record_llm_call(
                model=summary_response.model,
                tokens_in=summary_response.usage.input_tokens,
                tokens_out=summary_response.usage.output_tokens,
                task="summarize",
            )

            ctx.set_result({
                "classification": classification,
                "entities": entities,
                "summary": summary,
            })

            return {
                "classification": classification,
                "entities": entities,
                "summary": summary,
            }

        except PolicyViolationError as e:
            print(f"Policy violation: {e}")
            raise

# Run it
result = await analyze_content(
    "Analyze the impact of AI on healthcare",
    user_id="user_789"
)

Recording LLM Calls​

Anthropic's usage object has different field names than OpenAI:

# Anthropic response
response = await client.messages.create(...)

ctx.record_llm_call(
    model=response.model,
    tokens_in=response.usage.input_tokens,   # Note: input_tokens
    tokens_out=response.usage.output_tokens, # Note: output_tokens
    task="my_task",
    prompt_preview=prompt[:200],
    response_preview=response.content[0].text[:200],
)

Streaming with Anthropic​

Anthropic uses events instead of chunks:

async with WaxellContext(agent_name="streaming-claude") as ctx:
    stream = await client.messages.create(
        model="claude-sonnet-4",
        max_tokens=500,
        messages=[{"role": "user", "content": query}],
        stream=True,
    )

    content = ""
    tokens_in = 0
    tokens_out = 0

    async for event in stream:
        if event.type == "content_block_delta" and hasattr(event.delta, "text"):
            content += event.delta.text
        elif event.type == "message_start" and hasattr(event, "message"):
            if hasattr(event.message, "usage"):
                tokens_in = event.message.usage.input_tokens
        elif event.type == "message_delta" and hasattr(event, "usage"):
            if event.usage:
                tokens_out = event.usage.output_tokens

    ctx.record_step("stream_response", output={"length": len(content)})
    ctx.record_llm_call(
        model="claude-sonnet-4",
        tokens_in=tokens_in,
        tokens_out=tokens_out,
        response_preview=content[:200],
    )

Supported Models​

ModelAuto-InstrumentedCost Tracking
claude-opus-4YesYes
claude-sonnet-4YesYes
claude-3-5-sonnetYesYes
claude-3-5-haikuYesYes
claude-3-haikuYesYes

Tags and Metadata​

Enrich traces with contextual information:

async with WaxellContext(agent_name="claude-agent") as ctx:
    # Tags: searchable in UI
    ctx.set_tag("provider", "anthropic")
    ctx.set_tag("model_tier", "premium")
    ctx.set_tag("use_case", "analysis")

    # Metadata: arbitrary context
    ctx.set_metadata("sdk_version", "0.25.0")
    ctx.set_metadata("config", {"max_tokens": 500, "temperature": 0.7})

    response = await client.messages.create(...)

Error Handling​

from waxell_observe.errors import PolicyViolationError

try:
    async with WaxellContext(
        agent_name="claude-agent",
        enforce_policy=True,
    ) as ctx:
        response = await client.messages.create(...)

except PolicyViolationError as e:
    # Policy blocked execution (budget, rate limit, etc.)
    print(f"Blocked: {e.policy_result.reason}")

except anthropic.APIError as e:
    # Anthropic API error
    print(f"API error: {e}")

Best Practices​

  1. Call init() before importing anthropic -- enables auto-instrumentation
  2. Use input_tokens and output_tokens -- Anthropic's field names
  3. Handle streaming events by type -- content_block_delta, message_start, message_delta
  4. Set max_tokens -- Required by Anthropic API
  5. Add provider tag -- Makes filtering easy in the UI

Next Steps​