Anthropic SDK

pip install "adrian-sdk[anthropic]"

import asyncio
import os

import adrian
import anthropic


async def main():
    adrian.init(api_key="adr_live_...", ws_url="wss://adrian.secureagentics.ai/ws")

    # Your Anthropic code runs normally, and every call is captured.
    client = anthropic.AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    async with adrian.anthropic_invocation():
        response = await client.messages.create(
            model="claude-opus-5",
            max_tokens=1024,
            system="You are a helpful assistant.",
            messages=[{"role": "user", "content": "What is 2 + 2?"}],
        )
        # Thinking blocks can precede the text block, so select by type.
        print(next(b.text for b in response.content if b.type == "text"))

    adrian.shutdown()


asyncio.run(main())

async with adrian.anthropic_invocation():
    first = await client.messages.create(...)
    second = await client.messages.create(...)   # same invocation_id

async with adrian.anthropic_invocation():
    async with client.messages.stream(
        model="claude-opus-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Count to five."}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

        message = await stream.get_final_message()   # emitted and gated here

adrian.init(api_key="adr_live_...", auto_instrument=False)
adrian.patch_anthropic()