Anthropic Integration

from muxx import Muxx
from anthropic import Anthropic

muxx = Muxx()
client = muxx.wrap(Anthropic())

# All calls are now automatically traced
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

from muxx import Muxx
from anthropic import AsyncAnthropic

muxx = Muxx()
client = muxx.wrap(AsyncAnthropic())

response = await client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a poem"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="")

# Complete response is logged when streaming finishes

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system="You are a helpful coding assistant.",
    messages=[{"role": "user", "content": "Write a Python function to sort a list"}]
)

tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country"
                }
            },
            "required": ["location"]
        }
    }
]

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}]
)

# Tool use is captured in the trace

from muxx import trace

@trace("conversation")
def have_conversation():
    messages = []

    # Turn 1
    messages.append({"role": "user", "content": "What is Python?"})
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=messages
    )
    messages.append({"role": "assistant", "content": response.content[0].text})

    # Turn 2
    messages.append({"role": "user", "content": "What are its main uses?"})
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=messages
    )

    return response.content[0].text

# Both turns are grouped in one trace

import base64

with open("image.png", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": image_data
                    }
                },
                {"type": "text", "text": "Describe this image"}
            ]
        }
    ]
)

# Log shows image was included, token count reflects image tokens

# Latest and most capable
client.messages.create(model="claude-3-5-sonnet-20241022", ...)

# Fast and cost-effective
client.messages.create(model="claude-3-5-haiku-20241022", ...)

# Most powerful (when available)
client.messages.create(model="claude-3-opus-20240229", ...)