Bry

Posted on Jul 21 • Originally published at Medium

Getting Started with the Claude API: Messages, Streaming, and Tool Use

#ai #python #claude #webdev

Key Points

Install anthropic, set ANTHROPIC_API_KEY, and you have a working LLM API in five lines of Python.
Three models cover 95% of use cases: Opus for complex reasoning, Sonnet for most production tasks, Haiku for cheap high-volume classification.
Always stream long-form responses — timeouts and bad UX come from blocking on a full response.
Tool use requires a two-turn round trip: Claude signals a function call, your code runs it, you return the result.
Prompt caching cuts input token costs by up to 90% for apps that send the same system prompt on every request.

Introduction

Most tutorials for LLM APIs stop at "send a message, print a response." That gets you a demo. It does not get you a production application. I've integrated the Claude API into customer-facing applications and internal automation pipelines, and the same three gaps appear every time a team goes from prototype to production: they don't stream, they don't handle tool use as a round trip, and they only think about cost after the first billing cycle. The Claude API has specific patterns for streaming, tool use, and cost control that are non-obvious from a basic example — and getting them wrong results in timeouts, doubled costs, or silent logic errors when Claude tries to call a function and your code ignores it.

This article covers the full picture: how the Messages API works, how to pick the right model, how to stream correctly, how to implement tool use as a complete round trip, and how to keep costs predictable at scale. All code is Python 3.11+. The companion code at ai-integration/claude-api-quickstart/src/main.py is a runnable ClaudeClient class that demonstrates all four patterns.

Setup

Install the SDK and dependencies:

pip install anthropic python-dotenv

Store your API key as an environment variable — never hardcode it:

# .env
ANTHROPIC_API_KEY=sk-ant-...

The Anthropic SDK reads ANTHROPIC_API_KEY automatically. Initialize the client once at module level:

import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()
client = Anthropic()  # reads ANTHROPIC_API_KEY from env

Create the client once. A common mistake is instantiating Anthropic() inside a request handler. The client maintains an HTTP connection pool — recreating it on every call wastes ~50ms in connection setup and adds unnecessary overhead.

Messages API Basics

Every interaction with Claude goes through client.messages.create(). The API uses three message roles:

system — sets the context and rules for the entire conversation. Not a message in the history.
user — input from the human (or your application).
assistant — Claude's previous responses. Include these to continue a multi-turn conversation.

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system="You are a senior backend engineer. Be concise and direct.",
    messages=[
        {"role": "user", "content": "What is the difference between process and thread?"}
    ]
)

print(response.content[0].text)
print(f"Tokens used — input: {response.usage.input_tokens}, output: {response.usage.output_tokens}")

Key parameters:

Parameter	Type	Notes
`model`	string	Required. See model selection section below.
`max_tokens`	int	Required. Upper bound on output — Claude stops here even mid-sentence.
`system`	string	Optional. Sets assistant persona and rules.
`messages`	list	Required. Alternating user/assistant turns. Must start with `user`.
`temperature`	float	0.0–1.0. Default 1.0. Lower = more deterministic. Use 0.0 for classification.
`stop_sequences`	list	Claude stops generating at any of these strings.

The response stop_reason tells you why Claude stopped:

end_turn — Claude finished naturally.
max_tokens — hit the max_tokens limit. Response may be truncated.
tool_use — Claude wants to call a function. Must handle this explicitly.
stop_sequence — hit one of your stop sequences.

Always check stop_reason. If you receive tool_use and ignore it, Claude is waiting for your function result — it did not finish answering your question.

Model Selection

Three production models cover nearly all use cases. Choose by task complexity and cost tolerance:

Model	ID	Best For	Not For
Claude Opus 4	`claude-opus-4-8`	Complex multi-step reasoning, code generation from ambiguous specs, research synthesis	High-volume tasks (cost), simple classification
Claude Sonnet 4	`claude-sonnet-4-6`	Most production API tasks, customer-facing chat, structured extraction	Tasks where you need the cheapest option
Claude Haiku 4	`claude-haiku-4-5-20251001`	Classification, summarization, high-volume extraction, quick one-turn answers	Multi-step reasoning, long-form generation

Default to claude-sonnet-4-6 for any new feature. I start every integration on Sonnet and only swap to Haiku after running a sample of real production inputs through both models and comparing outputs — not benchmarks, actual outputs for your specific task. Migrate to Haiku only after confirming the output quality is acceptable for that specific task.

Streaming Responses

For responses longer than a paragraph, stream. A non-streaming call for a 2,000-token response waits 10–30 seconds before returning anything to the user. Streaming begins returning tokens in ~200ms.

def stream_response(system: str, prompt: str) -> None:
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # newline after stream ends

The with client.messages.stream() context manager handles connection cleanup. Inside the block, stream.text_stream yields each text delta as Claude generates it.

When to stream:

Any user-facing response longer than ~100 tokens.
Document generation, code generation, long explanations.

When not to stream:

Classification tasks where you need the complete label before doing anything.
Tool use — stream the full round trip only if you handle partial tool input JSON.
Batch processing jobs where individual response latency does not matter.

In practice, I default to streaming for anything user-facing and non-streaming for anything running in a background job. The distinction matters because mixing the two in the same codebase without clear conventions leads to subtle bugs when someone reuses a non-streaming helper in a latency-sensitive context.

Tool Use (Function Calling)

Tool use is a two-turn exchange. In turn one, Claude signals which function to call and with what arguments. Your application executes the function. In turn two, you send the result back, and Claude incorporates it into its final answer.

Define a tool

CALCULATOR_TOOLS = [
    {
        "name": "calculator",
        "description": "Performs basic arithmetic. Use when the user asks for a calculation.",
        "input_schema": {
            "type": "object",
            "properties": {
                "operation": {
                    "type": "string",
                    "enum": ["add", "subtract", "multiply", "divide"],
                    "description": "The arithmetic operation to perform."
                },
                "a": {"type": "number", "description": "First operand."},
                "b": {"type": "number", "description": "Second operand."}
            },
            "required": ["operation", "a", "b"]
        }
    }
]

Handle the round trip

import json

def use_tool(prompt: str) -> str:
    messages = [{"role": "user", "content": prompt}]

    # Turn 1: Claude may request a tool call
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=CALCULATOR_TOOLS,
        messages=messages,
    )

    if response.stop_reason != "tool_use":
        # Claude answered directly — no tool call needed
        return response.content[0].text

    # Extract the tool use block
    tool_use_block = next(b for b in response.content if b.type == "tool_use")
    tool_name = tool_use_block.name
    tool_input = tool_use_block.input

    # Execute the function locally
    result = _run_calculator(tool_input)

    # Turn 2: Return the tool result and get the final answer
    messages.append({"role": "assistant", "content": response.content})
    messages.append({
        "role": "user",
        "content": [
            {
                "type": "tool_result",
                "tool_use_id": tool_use_block.id,
                "content": str(result),
            }
        ],
    })

    final = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=CALCULATOR_TOOLS,
        messages=messages,
    )
    return final.content[0].text


def _run_calculator(args: dict) -> float:
    a, b, op = args["a"], args["b"], args["operation"]
    if op == "add":       return a + b
    if op == "subtract":  return a - b
    if op == "multiply":  return a * b
    if op == "divide":
        if b == 0:
            raise ValueError("Division by zero")
        return a / b
    raise ValueError(f"Unknown operation: {op}")

The key detail: messages.append({"role": "assistant", "content": response.content}) — you must include Claude's entire response object (including the tool use block) in the history before sending the tool result. Sending only the text content breaks the conversation. I've seen this cause a 400 BadRequestError in production because a developer extracted just the text from the assistant response and discarded the tool use block before appending — the API rejects the resulting message history as structurally invalid, and the error message doesn't point to the exact field.

Error Handling and Retries

Rate limit errors and transient server errors are inevitable at any meaningful scale. I've seen a well-tested integration start hitting RateLimitError the week after launch simply because a new feature tripled the request volume — the app had no retry logic, and every error surfaced directly to the user as a 500. Implement exponential backoff from day one:

import time
import anthropic

def _create_with_retry(
    messages: list,
    model: str,
    max_tokens: int,
    system: str = "",
    max_retries: int = 3,
) -> anthropic.types.Message:
    for attempt in range(max_retries):
        try:
            kwargs = {
                "model": model,
                "max_tokens": max_tokens,
                "messages": messages,
            }
            if system:
                kwargs["system"] = system
            return client.messages.create(**kwargs)
        except anthropic.RateLimitError:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # 1s, 2s, 4s
            time.sleep(wait)
        except anthropic.APIStatusError as e:
            if e.status_code >= 500 and attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            raise

Common exceptions:

Exception	Cause	Action
`anthropic.RateLimitError`	Too many requests	Exponential backoff
`anthropic.APIStatusError` (5xx)	Server error	Retry with backoff
`anthropic.BadRequestError`	Invalid request (context too long, invalid params)	Do not retry — fix the request
`anthropic.AuthenticationError`	Bad API key	Do not retry — check env var

Check token count before sending a long conversation to avoid BadRequestError on context overflow:

token_count = client.messages.count_tokens(
    model="claude-sonnet-4-6",
    messages=messages,
)
print(f"Prompt uses {token_count.input_tokens} tokens")

Cost Management and Prompt Caching

Claude charges per token: input tokens (what you send) and output tokens (what Claude generates). Output tokens cost 3–5× more per token than input tokens. The usage breakdown appears in every response:

print(response.usage.input_tokens)   # tokens you sent
print(response.usage.output_tokens)  # tokens Claude generated

Prompt caching

If your application sends the same system prompt on every request (a common pattern), prompt caching can reduce input costs by 60–90%. Mark the cacheable prefix with cache_control:

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are a senior backend engineer...\n[very long system prompt]",
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[{"role": "user", "content": user_question}],
)

On the first call, Anthropic caches the marked prefix. Subsequent calls within the cache TTL pay ~10% of the normal input price for cached tokens. The response includes cache statistics:

print(response.usage.cache_creation_input_tokens)  # tokens written to cache
print(response.usage.cache_read_input_tokens)       # tokens read from cache (cheap)

When caching pays off: system prompts over ~1,000 tokens with more than a few calls per day. I enable caching by default on any app with a fixed system prompt longer than 500 tokens — the implementation cost is two lines and the savings compound quickly once you're past a few hundred daily active users. For shorter prompts or low-volume applications, the overhead is negligible.

Common Mistakes

Not streaming long-form responses. A blocking call for a 2,000-token response times out in some HTTP clients and always delivers a poor user experience. Use client.messages.stream() for any response that takes more than ~1 second.
Ignoring stop_reason. If stop_reason is tool_use and you return response.content[0].text, you are returning a partial or empty string — Claude was waiting to give you a real answer after executing the tool. I've seen this ship to production as a bug where the API "randomly" returned empty responses on certain questions — questions that happened to trigger tool use. Always check stop_reason before extracting text.
Setting max_tokens too low. If max_tokens is smaller than the full response, Claude stops mid-sentence and stop_reason returns max_tokens. The response is valid JSON but the content is truncated. Set max_tokens generously — you are not charged for unused headroom.
Recreating the Anthropic() client on every request. The client maintains an HTTP connection pool internally. Instantiating it per-request wastes connection setup time and bypasses the pool. Create one instance at module level and reuse it.
Sending unbounded conversation history. Every previous turn counts toward the input token cost. For long conversations, implement a windowing strategy: keep the system prompt, the last N turns, and trim the middle. Use client.messages.count_tokens() before every send to catch overruns early.

Full Example

The companion code at ai-integration/claude-api-quickstart/src/main.py implements all patterns above as a ClaudeClient class:

from dotenv import load_dotenv
from src.main import ClaudeClient

load_dotenv()
claude = ClaudeClient()

# Basic message
print(claude.send_message(
    system="You are a concise technical explainer.",
    prompt="What is a vector embedding in one sentence?",
))

# Streaming
for chunk in claude.stream_response(
    system="You are a Python expert.",
    prompt="Explain Python generators in detail.",
):
    print(chunk, end="", flush=True)

# Tool use
print(claude.use_tool("What is 2847 multiplied by 193?"))

Setup:

cd ai-integration/claude-api-quickstart
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env       # add your key
python -m src.main

Full source: GitHub link

State: The Conversation Turn Model

Conclusion

The difference between a Claude integration that works in a demo and one that holds up in production is not the model — it's whether you stream, whether you handle tool use as a complete round trip, and whether you have a cost strategy before your usage scales. Most production failures I've traced come back to one of those three. Ship an integration that gets all three right from the start, and you will spend your time building features instead of debugging silent errors and explaining surprise billing cycles to stakeholders.

DEV Community