LLM APIs

import anthropic

client = anthropic.Anthropic()          # reads ANTHROPIC_API_KEY

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    system="You are a precise assistant. Answer with commands, not prose.",
    messages=[
        {"role": "user", "content": "How do I list pods that are not running?"},
    ],
    temperature=0,
)
print(resp.content[0].text)
print(resp.usage.input_tokens, resp.usage.output_tokens)

from openai import OpenAI

client = OpenAI()                       # reads OPENAI_API_KEY
resp = client.responses.create(
    model="gpt-4.1",
    input="How do I list pods that are not running?",
)
print(resp.output_text)

with client.messages.stream(
    model="claude-sonnet-5", max_tokens=1024,
    messages=[{"role": "user", "content": prompt}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

tools = [{
    "name": "record_incident",
    "description": "Record a structured incident summary.",
    "input_schema": {
        "type": "object",
        "properties": {
            "severity": {"type": "string", "enum": ["sev1", "sev2", "sev3"]},
            "service": {"type": "string"},
            "summary": {"type": "string"},
        },
        "required": ["severity", "service", "summary"],
    },
}]

resp = client.messages.create(
    model="claude-sonnet-5", max_tokens=1024, tools=tools,
    tool_choice={"type": "tool", "name": "record_incident"},
    messages=[{"role": "user", "content": transcript}],
)
block = next(b for b in resp.content if b.type == "tool_use")
incident = block.input          # already parsed and schema-checked

count = client.messages.count_tokens(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": prompt}],
)
print(count.input_tokens)

system=[{
    "type": "text",
    "text": long_static_context,
    "cache_control": {"type": "ephemeral"},   # cached prefix
}]

from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_random_exponential
import anthropic

@retry(
    retry=retry_if_exception_type((anthropic.RateLimitError, anthropic.APIStatusError)),
    wait=wait_random_exponential(min=1, max=30),
    stop=stop_after_attempt(5),
    reraise=True,
)
def ask(prompt: str) -> str:
    r = client.messages.create(model="claude-sonnet-5", max_tokens=512,
                               messages=[{"role": "user", "content": prompt}])
    return r.content[0].text

export ANTHROPIC_API_KEY=$(vault kv get -mount=secret -field=key llm/anthropic)

# Smoke test a key
curl -s https://api.anthropic.com/v1/messages -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' -H 'content-type: application/json' -d '{"model":"claude-sonnet-5","max_tokens":16,"messages":[{"role":"user","content":"ping"}]}' | jq -r '.content[0].text'

# Token usage of a single call
curl -s ... | jq '.usage'

# Stream from the shell
curl -sN https://api.anthropic.com/v1/messages -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' -H 'content-type: application/json' -d '{"model":"claude-sonnet-5","max_tokens":256,"stream":true,"messages":[{"role":"user","content":"list three checks"}]}' | grep -o '"text":"[^"]*"'

# Check rate limit headroom
curl -sD - -o /dev/null https://api.anthropic.com/v1/messages -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' -H 'content-type: application/json' -d '{"model":"claude-sonnet-5","max_tokens":1,"messages":[{"role":"user","content":"x"}]}' | grep -i ratelimit

# Estimate tokens in a file, roughly
wc -c < prompt.txt | awk '{printf "~%d tokens\n", $1/4}'

# Redact obvious secrets before sending a log as context
sed -E 's/(AKIA[0-9A-Z]{16}|eyJ[A-Za-z0-9_-]{20,})/[REDACTED]/g' app.log > safe.log