LLM APIs

# List models available to this key (read-only)
curl -s https://api.anthropic.com/v1/models \
  -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' \
  | jq -r '.data[] | [.id, .max_input_tokens, .max_tokens] | @tsv'

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?"},
    ],
)
text = next(b.text for b in resp.content if b.type == "text")
print(text)
print(resp.stop_reason, resp.usage.input_tokens, resp.usage.output_tokens)

from openai import OpenAI

client = OpenAI()                       # reads OPENAI_API_KEY
resp = client.responses.create(
    model="gpt-6-sol",
    instructions="Answer with commands, not prose.",
    input="How do I list pods that are not running?",
    max_output_tokens=1024,
)
print(resp.output_text)

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": prompt}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()   # full Message, same shape as a non-streaming call

from pydantic import BaseModel

class Incident(BaseModel):
    severity: str
    service: str
    summary: str

resp = client.messages.parse(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": transcript}],
    output_format=Incident,
)
incident = resp.parsed_output            # validated Incident instance

tools = [{
    "name": "get_pod_status",
    "description": "Return the status of pods in a namespace.",
    "strict": True,
    "input_schema": {
        "type": "object",
        "properties": {"namespace": {"type": "string"}},
        "required": ["namespace"],
        "additionalProperties": False,
    },
}]

messages = [{"role": "user", "content": "What is failing in my-namespace?"}]
while True:
    resp = client.messages.create(model="claude-sonnet-5", max_tokens=4096,
                                  tools=tools, messages=messages)
    messages.append({"role": "assistant", "content": resp.content})
    if resp.stop_reason != "tool_use":
        break
    results = [
        {"type": "tool_result", "tool_use_id": b.id, "content": run_tool(b.name, b.input)}
        for b in resp.content if b.type == "tool_use"
    ]
    messages.append({"role": "user", "content": results})   # all results in one message

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

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": long_static_context,
        "cache_control": {"type": "ephemeral"},   # 5-minute TTL; {"type": "ephemeral", "ttl": "1h"} for 1 hour
    }],
    messages=[{"role": "user", "content": question}],
)
print(resp.usage.cache_creation_input_tokens, resp.usage.cache_read_input_tokens)

client = anthropic.Anthropic(max_retries=5, timeout=120.0)   # timeout in seconds (Python)

system = """You are an on-call assistant for a Kubernetes platform team.
Answer with commands first, one short sentence of explanation after each.
If the question cannot be answered from the supplied context, reply exactly: INSUFFICIENT_CONTEXT.
Never invent flags. Only cite files that appear in <context>."""

user = f"""<context>
{runbook_excerpt}
</context>

<question>
{question}
</question>"""

import (
    "context"
    "encoding/json"

    "github.com/anthropics/anthropic-sdk-go"
)

podTool := anthropic.ToolParam{
    Name:        "get_pod_status",
    Description: anthropic.String("Return the status of pods in a namespace."),
    InputSchema: anthropic.ToolInputSchemaParam{
        Properties: map[string]any{"namespace": map[string]any{"type": "string"}},
        Required:   []string{"namespace"},
    },
    Strict: anthropic.Bool(true),
}
tools := []anthropic.ToolUnionParam{{OfTool: &podTool}}
messages := []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock("What is failing in my-namespace?"))}

for {
    resp, err := client.Messages.New(ctx, anthropic.MessageNewParams{
        Model: "claude-sonnet-5", MaxTokens: 4096, Tools: tools, Messages: messages,
    })
    if err != nil { return err }
    messages = append(messages, resp.ToParam())                 // thinking and tool_use blocks go back unchanged
    if resp.StopReason != anthropic.StopReasonToolUse { break }

    var results []anthropic.ContentBlockParamUnion
    for _, block := range resp.Content {
        if tu, ok := block.AsAny().(anthropic.ToolUseBlock); ok {
            var in struct{ Namespace string `json:"namespace"` }
            if err := json.Unmarshal([]byte(tu.JSON.Input.Raw()), &in); err != nil {
                results = append(results, anthropic.NewToolResultBlock(block.ID, "invalid input: "+err.Error(), true))
                continue
            }
            out, err := getPodStatus(ctx, in.Namespace)          // validate in.Namespace against an allow-list first
            results = append(results, anthropic.NewToolResultBlock(block.ID, out, err != nil))
        }
    }
    messages = append(messages, anthropic.NewUserMessage(results...))
}

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": 1024,
    "tools": [{"name": "get_pod_status", "description": "Return the status of pods in a namespace.", "strict": true,
               "input_schema": {"type": "object", "properties": {"namespace": {"type": "string"}}, "required": ["namespace"], "additionalProperties": false}}],
    "messages": [
      {"role": "user", "content": "What is failing in my-namespace?"},
      {"role": "assistant", "content": [{"type": "tool_use", "id": "toolu_01A", "name": "get_pod_status", "input": {"namespace": "my-namespace"}}]},
      {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_01A", "content": "api-7d9f CrashLoopBackOff (3 restarts)"}]}
    ]}' | jq -r '.content[] | select(.type == "text") | .text'

resp = client.responses.create(model="gpt-6-sol", tools=tools, input=input_list)
input_list += resp.output                                       # keep the model's items in history
for item in resp.output:
    if item.type == "function_call":
        args = json.loads(item.arguments)
        input_list.append({"type": "function_call_output", "call_id": item.call_id, "output": run_tool(item.name, args)})
resp = client.responses.create(model="gpt-6-sol", tools=tools, input=input_list)

jq -n --rawfile t transcript.txt '{
  model: "claude-sonnet-5", max_tokens: 1024,
  messages: [{role: "user", content: $t}],
  output_config: {format: {type: "json_schema", schema: {
    type: "object",
    properties: {
      severity: {type: "string", enum: ["sev1", "sev2", "sev3"]},
      service:  {type: "string"},
      summary:  {type: "string", description: "One sentence, no more than 30 words"}
    },
    required: ["severity", "service", "summary"], additionalProperties: false
  }}}}' \
| 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 @- \
| jq -r '.content[] | select(.type == "text") | .text' | jq .

stream := client.Messages.NewStreaming(ctx, anthropic.MessageNewParams{
    Model: "claude-sonnet-5", MaxTokens: 64000,
    Messages: []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock(prompt))},
})
var msg anthropic.Message
for stream.Next() {
    ev := stream.Current()
    msg.Accumulate(ev)                                          // builds the complete Message as events arrive
    if d, ok := ev.AsAny().(anthropic.ContentBlockDeltaEvent); ok {
        if td, ok := d.Delta.AsAny().(anthropic.TextDelta); ok { fmt.Print(td.Text) }
    }
}
if err := stream.Err(); err != nil { return err }             // includes error events delivered mid-stream
fmt.Println(msg.StopReason, msg.Usage.OutputTokens)

with client.responses.stream(model="gpt-6-sol", input=prompt) as stream:   # OpenAI
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
    final = stream.get_final_response()

# Submit: one requests[] entry per prompt, each with a custom_id you can join on later
jq -n '{requests: [inputs | {custom_id: .id, params: {model: "claude-haiku-4-5", max_tokens: 512,
        messages: [{role: "user", content: .text}]}}]}' items.jsonl \
| curl -s https://api.anthropic.com/v1/messages/batches \
    -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' -H 'content-type: application/json' -d @- \
| tee batch.json | jq '{id, processing_status, request_counts}'

# Poll until processing_status is "ended", then stream the JSONL results
batch=$(jq -r .id batch.json)
until [ "$(curl -s https://api.anthropic.com/v1/messages/batches/$batch -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' | jq -r .processing_status)" = ended ]; do sleep 60; done
curl -s "https://api.anthropic.com/v1/messages/batches/$batch/results" -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' \
| jq -r 'select(.result.type == "succeeded") | [.custom_id, (.result.message.content[] | select(.type == "text") | .text)] | @tsv'

from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

batch = client.messages.batches.create(requests=[
    Request(custom_id=item.id, params=MessageCreateParamsNonStreaming(
        model="claude-haiku-4-5", max_tokens=512, messages=[{"role": "user", "content": item.text}]))
    for item in items
])
while client.messages.batches.retrieve(batch.id).processing_status != "ended":
    time.sleep(60)
for r in client.messages.batches.results(batch.id):
    if r.result.type == "succeeded":
        store(r.custom_id, r.result.message)
    elif r.result.type == "errored" and r.result.error.type != "invalid_request":
        retry_later(r.custom_id)                                # server-side errors are safe to resubmit; invalid_request is not

import json, statistics

cases = [json.loads(l) for l in open("evals/triage.jsonl", encoding="utf-8")]   # {"input": ..., "expected": "sev2"}
scores, costs = [], []
for c in cases:
    resp = client.messages.parse(model=MODEL, max_tokens=512, system=SYSTEM,
                                 messages=[{"role": "user", "content": c["input"]}], output_format=Incident)
    scores.append(resp.parsed_output.severity == c["expected"])
    costs.append(resp.usage.input_tokens * IN_PRICE + resp.usage.output_tokens * OUT_PRICE)
print(f"accuracy={statistics.mean(scores):.3f} cost_per_case=${statistics.mean(costs):.4f} n={len(cases)}")

from openai import OpenAI

local = OpenAI(base_url="http://127.0.0.1:11434/v1", api_key="ollama")   # Ollama; vLLM defaults to :8000/v1
resp = local.chat.completions.create(
    model="llama3.3",
    messages=[{"role": "system", "content": "Answer briefly."}, {"role": "user", "content": "ping"}],
    max_tokens=64,
    temperature=0,
)
print(resp.choices[0].message.content, resp.usage)

# Same call from the shell; works against OpenAI, vLLM, Ollama and LiteLLM with only the URL and model changed
curl -s http://127.0.0.1:8000/v1/chat/completions -H 'content-type: application/json' \
  -d '{"model":"my-model","messages":[{"role":"user","content":"ping"}],"max_tokens":64,"stream":false}' \
  | jq -r '.choices[0].message.content'

# List what the server serves
curl -s http://127.0.0.1:8000/v1/models | jq -r '.data[].id'

# Load the key from Vault at runtime (never commit it)
export ANTHROPIC_API_KEY="$(vault kv get -mount=secret -field=key llm/anthropic)"

# Smoke test a key (prints the reply text)
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-haiku-4-5","max_tokens":64,"messages":[{"role":"user","content":"ping"}]}' \
  | jq -r '.content[] | select(.type == "text") | .text'

# Show token usage and stop reason for one call
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-haiku-4-5","max_tokens":64,"messages":[{"role":"user","content":"ping"}]}' \
  | jq '{stop_reason, usage}'

# Stream text deltas 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":1024,"stream":true,"messages":[{"role":"user","content":"list three checks"}]}' \
  | sed -un 's/^data: //p' | jq -rj --unbuffered 'select(.type == "content_block_delta" and .delta.type == "text_delta") | .delta.text'

# Show rate limit headroom and the request ID (spends a few tokens)
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-haiku-4-5","max_tokens":1,"messages":[{"role":"user","content":"x"}]}' \
  | grep -iE '^(anthropic-ratelimit|retry-after|request-id)'

# Count tokens in a file (free, separate rate limit, no generation)
jq -n --rawfile p prompt.txt '{model:"claude-sonnet-5", messages:[{role:"user", content:$p}]}' \
  | curl -s https://api.anthropic.com/v1/messages/count_tokens \
      -H "x-api-key: $ANTHROPIC_API_KEY" -H 'anthropic-version: 2023-06-01' -H 'content-type: application/json' -d @- \
  | jq .input_tokens

# Redact AWS access key IDs and JWTs from a log before using it as context
sed -E 's/(AKIA[0-9A-Z]{16}|eyJ[A-Za-z0-9_-]{20,})/[REDACTED]/g' app.log > safe.log

# OpenAI smoke test
curl -s https://api.openai.com/v1/responses \
  -H "Authorization: Bearer $OPENAI_API_KEY" -H 'content-type: application/json' \
  -d '{"model":"gpt-6-luna","input":"ping","max_output_tokens":64}' \
  | jq -r '.output[] | select(.type == "message") | .content[] | select(.type == "output_text") | .text'