Anthropic (Python) → OpenObserve

pip install openobserve-telemetry-sdk opentelemetry-instrumentation-anthropic python-dotenv

# OpenObserve instance URL
# Default for self-hosted: http://localhost:5080
OPENOBSERVE_URL=https://api.openobserve.ai/

# Your OpenObserve organisation slug or ID
OPENOBSERVE_ORG=your_org_id

# Basic auth token — Base64-encoded "email:password"
OPENOBSERVE_AUTH_TOKEN="Basic <your_base64_token>"

# Anthropic API key
ANTHROPIC_API_KEY=your-anthropic-key

from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor
from openobserve import openobserve_init

# Instrument before importing the Anthropic client
AnthropicInstrumentor().instrument()
openobserve_init()

from anthropic import Anthropic

client = Anthropic()

# Messages API
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What is distributed tracing?"}],
)
print(response.content[0].text)

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=512,
    system="You are a concise technical assistant.",
    messages=[{"role": "user", "content": "Explain a trace span in one sentence."}],
)
print(response.content[0].text)

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=256,
    messages=[{"role": "user", "content": "Write a haiku about observability."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

from anthropic import AsyncAnthropic
import asyncio

async_client = AsyncAnthropic()

async def main():
    response = await async_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=256,
        messages=[{"role": "user", "content": "Hello async!"}],
    )
    print(response.content[0].text)

asyncio.run(main())