Anthropic

pip install rdk anthropic --extra-index-url https://pypi.fury.io/021labs/

from anthropic import Anthropic
from rdk import observe, shutdown

@observe(name="anthropic-chat")
def chat(message: str) -> str:
    client = Anthropic()
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}]
    )
    return response.content[0].text

result = chat("Explain quantum computing in simple terms")
print(result)

shutdown()

import asyncio
from anthropic import AsyncAnthropic
from rdk import observe, shutdown

@observe(name="async-chat")
async def async_chat(message: str) -> str:
    client = AsyncAnthropic()
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}]
    )
    return response.content[0].text

result = asyncio.run(async_chat("Hello!"))
shutdown()

from anthropic import Anthropic
from rdk import observe, shutdown

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

@observe(name="weather-agent")
def weather_agent(question: str) -> str:
    client = Anthropic()
    messages = [{"role": "user", "content": question}]

    # First call - may request tool use
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    # Handle tool calls
    while response.stop_reason == "tool_use":
        tool_use = next(b for b in response.content if b.type == "tool_use")

        # Execute tool (mock implementation)
        tool_result = f"Weather in {tool_use.input['location']}: 72°F, sunny"

        messages.append({"role": "assistant", "content": response.content})
        messages.append({
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": tool_result
            }]
        })

        # Continue conversation
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

    return response.content[0].text

result = weather_agent("What's the weather in San Francisco?")
shutdown()