Anthropic Wrapper

Wrap your Anthropic client to automatically trace every messages.create() call with model, input messages, system prompt, output, token usage, and cost.

Installation

pip install twosignal[anthropic]

Usage

from twosignal import TwoSignal
from twosignal.wrappers.anthropic import wrap_anthropic
from anthropic import Anthropic

ts = TwoSignal()
client = wrap_anthropic(Anthropic())

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)

What Gets Captured

System Prompt

The system prompt is captured as part of the span input:

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="You are a helpful support agent.",
    messages=[{"role": "user", "content": "How do I reset my password?"}],
)
# span input: {"messages": [...], "system": "You are a helpful support agent."}

Async Client

from anthropic import AsyncAnthropic

async_client = wrap_anthropic(AsyncAnthropic())

response = await async_client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)

Supported Models

Cost tracking works automatically for:

• claude-opus-4-20250514
• claude-sonnet-4-20250514
• claude-haiku-4-20250414
• claude-3-5-sonnet-20241022
• claude-3-5-haiku-20241022
• claude-3-opus-20240229

Combining with @observe

@observe(span_type=SpanType.AGENT)
def my_agent(query):
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": query}],
    )
    return response.content[0].text

# trace tree:
# my_agent (AGENT)
# └── anthropic.messages.create(claude-sonnet-4-20250514) (LLM)

Error Tracking

try:
    response = client.messages.create(...)
except anthropic.RateLimitError:
    # span shows status: ERROR, error_message: "429 Rate limit exceeded"
    handle_rate_limit()

Limitations

• Streaming is not traced — streaming calls (stream=True) fall through to the unwrapped client. The call still works, but no span is created.
• Only messages.create() is wrapped. Other methods (completions, embeddings) are unaffected.

Migrating Existing Code

  from twosignal import TwoSignal
+ from twosignal.wrappers.anthropic import wrap_anthropic
  from anthropic import Anthropic

+ ts = TwoSignal()
- client = Anthropic()
+ client = wrap_anthropic(Anthropic())

  # all your existing code works unchanged