llm-integration

llm-integration

Practical patterns for integrating Claude (Anthropic) and OpenAI into production pipelines — prompting, streaming, tool use, cost tracking, and multi-model routing.

1) Anthropic client (Python)

import anthropic

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# Simple completion
message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": prompt}],
)
text = message.content[0].text

# With system prompt
message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    system="You are a precise SEO article writer. Output only valid markdown.",
    messages=[{"role": "user", "content": f"Write about: {topic}"}],
)

2) OpenAI client (Python)

from openai import AsyncOpenAI

client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

response = await client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user",   "content": user_prompt},
    ],
    max_tokens=2048,
    temperature=0.3,
)
text = response.choices[0].message.content
tokens_used = response.usage.total_tokens

3) Streaming responses

# Anthropic stream
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    messages=[{"role": "user", "content": prompt}],
) as stream:
    for chunk in stream.text_stream:
        print(chunk, end="", flush=True)
    message = stream.get_final_message()

# OpenAI stream
stream = await client.chat.completions.create(model="gpt-4o", messages=msgs, stream=True)
async for chunk in stream:
    delta = chunk.choices[0].delta.content or ""
    print(delta, end="", flush=True)

4) Tool use / function calling (Claude)

tools = [{
    "name": "search_web",
    "description": "Search the web and return top results",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
        },
        "required": ["query"],
    },
}]

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "What are the latest keto trends?"}],
)

# Handle tool call
if response.stop_reason == "tool_use":
    tool_use = next(b for b in response.content if b.type == "tool_use")
    tool_result = await search_web(tool_use.input["query"])

    # Continue conversation with result
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        tools=tools,
        messages=[
            {"role": "user", "content": "What are the latest keto trends?"},
            {"role": "assistant", "content": response.content},
            {"role": "user", "content": [{"type": "tool_result", "tool_use_id": tool_use.id, "content": str(tool_result)}]},
        ],
    )