DEV Community: syncore

How to Fine-Tune Your Prompts for Claude 3.5 Sonnet vs Claude 3 Opus

syncore — Mon, 10 Aug 2026 10:20:43 +0000

4 min read · 823 words

If you are building AI-powered features with Anthropic’s API, treating Claude 3.5 Sonnet and Claude 3 Opus as drop-in replacements for one another is a huge mistake.

While both models sit at the top of LLM benchmarks, they operate with fundamentally different strengths. Claude 3.5 Sonnet is lightning-fast, hyper-precise, and excels at instruction-following, refactoring, and structured outputs. Claude 3 Opus is a deep-thinking powerhouse designed for nuanced reasoning, highly complex domain analysis, and open-ended creative tasks.

Using the exact same prompt for both models will lead to over-engineering for Opus or under-specifying for Sonnet. Here is how to tailor your prompts for each model to get production-grade output every time.

The Fundamental Rule

Prompt Sonnet like a Senior Engineer executing a spec: Give it exact formats, strict constraints, and explicit boundaries (using XML tags).
Prompt Opus like a Principal Systems Architect: Give it full context, room to reason through trade-offs, and ask it to evaluate edge cases before delivering a solution.

Strategy 1: Prompting Claude 3.5 Sonnet (Precision & Structure)

Sonnet 3.5 thrives when you leverage Anthropic's recommended XML tag structure. Because Sonnet has lower latency and ultra-sharp directive follow-through, explicitly defining context, input data, and rules in distinct XML tags yields deterministic, perfect results.

Best Practices for Sonnet:

Wrap inputs in explicit XML tags (<code_to_review>, <formatting_rules>, <context>).
Be extremely specific about output constraints (e.g., "Return ONLY valid JSON").
Use System Prompts to lock in persona and response rules.

Here is a practical Python example sending a structured refactoring task to Sonnet 3.5:

import anthropic

client = anthropic.Anthropic()

# Sonnet performs best with clear structural boundaries
prompt = """
<task>Refactor the provided Python function to improve algorithmic complexity from O(n^2) to O(n).</task>

<code_to_refactor>
def find_duplicates(lst):
    dups = []
    for i in range(len(lst)):
        for j in range(i + 1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in dups:
                dups.append(lst[i])
    return dups
</code_to_refactor>

<formatting_rules>
1. Return ONLY the refactored Python function inside a standard markdown code block.
2. Include short inline comments explaining time and space complexity.
3. Do NOT include any conversational preamble or postscript.
</formatting_rules>
"""

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1000,
    system="You are a staff software engineer specializing in algorithmic optimization.",
    messages=[{"role": "user", "content": prompt}]
)

print(response.content[0].text)

Strategy 2: Prompting Claude 3 Opus (Deep Reasoning & Synthesis)

Opus shines when solving ambiguous, multi-layered problems that require high-level abstraction. If you constrain Opus too tightly with micro-managed rules, you suppress its core strength: deep chain-of-thought analysis.

Best Practices for Opus:

Encourage "thinking before acting": Ask Opus to outline its assumptions, potential failure modes, and architectural trade-offs first.
Provide broad context over rigid strictures: Explain why you need something, not just what to output.
Ask for multi-angle evaluation: Use prompts that force the model to critique its own logic.

Here is a Python example leveraging Opus for high-level system architecture design:

import anthropic

client = anthropic.Anthropic()

# Opus excels at multi-variable problem solving and trade-off analysis
prompt = """
We are migrating our monolithic e-commerce backend (50k requests/min) to a microservices architecture.

Current State:
- Monolithic PostgreSQL database with 40+ tightly coupled tables.
- Synchronous REST calls between domain services causing cascading failures.
- Read/Write ratio is roughly 85/15.

Goal:
1. Propose a phased migration path using the Strangler Fig pattern.
2. Address data consistency across service boundaries (e.g., Saga Pattern vs. Two-Phase Commit).
3. Identify top 3 operational failure modes during the transition and how to mitigate them.

Take your time to analyze trade-offs deeply before providing the final architectural roadmap.
"""

response = client.messages.create(
    model="claude-3-opus-20240229",
    max_tokens=2500,
    system="You are a principal enterprise architect. Prioritize operational resilience and fault tolerance in your responses.",
    messages=[{"role": "user", "content": prompt}]
)

print(response.content[0].text)

Quick Comparison Cheat Sheet

Feature	Claude 3.5 Sonnet	Claude 3 Opus
Primary Use Case	Code generation, strict JSON output, fast API routes	Strategic planning, complex writing, deep analysis
Prompting Style	Direct, highly structured, heavy XML tagging	Goal-oriented, contextual, step-by-step reasoning
Response Control	Strict formatting rules ("Return ONLY...")	Open-ended trade-off evaluations
Latency / Cost	Fast / Cost-Effective	Slower / Higher Cost

Practical Takeaways

Use Sonnet 3.5 by default for developer tools, automated code generation, and low-latency API workflows where schema compliance is required.
Switch to Opus when dealing with high-ambiguity prompts, complex legal/financial text analysis, or macro-level architectural planning.
Migrate Sonnet prompts to XML: If Sonnet isn't outputting what you expect, don't write longer paragraphs—wrap your input variables in <tags> and tighten the system prompt.

What's Your Default Model?

Are you running Sonnet 3.5 in production for code tasks, or relies on Opus for complex workflows? Let me know in the comments how your prompt performance changes between the two!

Streaming Responses with Claude: A Complete Developer Guide

syncore — Mon, 03 Aug 2026 12:29:51 +0000

4 min read · 789 words

When building LLM-powered applications, user experience lives or dies by latency. Waiting 5 to 10 seconds for a full completion to render makes apps feel broken. Streaming responses reduces your Time to First Token (TTFT) to a few hundred milliseconds, providing real-time text delivery that keeps users engaged.

In this guide, you will learn how to stream responses from Anthropic's Claude API using both Python and Node.js, handle stream lifecycle events, and pipe tokens directly to a browser client.

How Claude Streaming Works

Under the hood, Claude streaming relies on Server-Sent Events (SSE). Instead of returning a single JSON payload after generating the full response, the Anthropic API opens an HTTP connection and pushes small JSON chunks (deltas) as tokens are produced.

The Anthropic SDKs abstract raw SSE handling into convenient stream helpers that expose two interface paradigms:

Iterators: Yield raw tokens as string deltas.
Event Listeners: Emit named events for precise lifecycle tracking (e.g., stream start, text delta, completion).

Option 1: Streaming in Python

For Python services, backend scripts, or CLI tools, the standard approach uses the messages.stream() context manager. This handles opening and closing the HTTP connection automatically.

Prerequisites

Install the official Anthropic Python SDK:

pip install anthropic

Make sure your ANTHROPIC_API_KEY is set in your environment variables.

Code Example: Python Console Stream

import os
import anthropic

# Initialize client (reads ANTHROPIC_API_KEY from env)
client = anthropic.Anthropic()

def stream_claude_response(prompt: str):
    print("Claude: ", end="", flush=True)

    # Open streaming context
    with client.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        # text_stream yields text deltas directly as strings
        for text_delta in stream.text_stream:
            print(text_delta, end="", flush=True)

    print("\n")

if __name__ == "__main__":
    stream_claude_response("Write a 3-step action plan to optimize database indexing.")

Key Elements

stream.text_stream: An iterator yielding only text deltas, bypassing unnecessary metadata parsing.
flush=True: Ensures Python prints chunks instantly to stdout without buffering.

Option 2: Streaming to Web Clients (Node.js + Express)

To build web applications, you typically proxy requests through a Node.js backend and stream tokens to the browser via SSE.

Prerequisites

Install the required packages:

npm install @anthropic-ai/sdk express

Code Example: Express SSE Server

import Express from 'express';
import Anthropic from '@anthropic-ai/sdk';

const app = Express();
const anthropic = new Anthropic(); // Reads process.env.ANTHROPIC_API_KEY

app.use(Express.json());

app.get('/api/stream', async (req, res) => {
  const prompt = req.query.prompt || 'Explain event loops simply.';

  // 1. Configure SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  try {
    // 2. Initiate Claude stream
    const stream = anthropic.messages.stream({
      model: 'claude-3-5-sonnet-20241022',
      max_tokens: 1024,
      messages: [{ role: 'user', content: prompt }],
    });

    // 3. Attach stream lifecycle listeners
    stream.on('text', (textDelta) => {
      res.write(`data: ${JSON.stringify({ text: textDelta })}\n\n`);
    });

    stream.on('end', () => {
      res.write('data: [DONE]\n\n');
      res.end();
    });

    stream.on('error', (error) => {
      console.error('Streaming error:', error);
      res.status(500).end();
    });
  } catch (err) {
    console.error('Server error:', err);
    res.status(500).json({ error: 'Failed to initialize stream' });
  }
});

app.listen(3000, () => {
  console.log('Streaming server listening on http://localhost:3000');
});

How the Frontend Consumes This

On the client side, use the native EventSource API or fetch with readable streams to consume this endpoint:

const eventSource = new EventSource('/api/stream?prompt=Hello');

eventSource.onmessage = (event) => {
  if (event.data === '[DONE]') {
    eventSource.close();
    return;
  }
  const payload = JSON.parse(event.data);
  document.getElementById('output').innerText += payload.text;
};

Understanding Stream Events

If you need full control over the response object—such as tracking token usage metrics or capturing tool calls—listen to specific SSE events instead of raw text:

Event Name	Description
`message_start`	Contains high-level metadata (model, input token counts).
`content_block_start`	Signals the start of a block (text or tool call).
`text` / `text_delta`	The actual string payload generated in the current chunk.
`message_delta`	Emitted when output stops; includes output token usage and stop reason.
`end`	Triggered when the HTTP stream closes completely.

Essential Best Practices

Client Disconnects: Handle connection drops cleanly. In Node.js, listen for req.on('close') and call stream.controller.abort() to terminate API generation and avoid paying for unconsumed tokens.
Buffer Management: Do not try to parse partial JSON across streaming boundaries. Keep messages formatted as simple SSE events (data: {...}\n\n).
Error Boundaries: Wrap streaming initializations in try/catch blocks. Network failures can happen mid-stream, so ensure your UI handles partial output gracefully.

Conclusion

Streaming Claude responses converts slow LLM roundtrips into fluid, immediate user interfaces. Whether you use Python's stream context iterator or Node.js event hooks with SSE, integrating streaming takes under 50 lines of code.

What streaming patterns or UI frameworks are you combining with Claude? Share your approach or ask questions in the comments below!

Batch API Deep Dive: How to Process 1,000 Prompts at Half the Cost

syncore — Mon, 27 Jul 2026 12:28:38 +0000

4 min read · 748 words

If you are running large-scale LLM workloads—like content generation, data enrichment, offline evaluation, or customer feedback analysis—you are probably spending way too much money.

When your application doesn't require instantaneous, sub-second response times, sending thousands of synchronous API requests is a waste of cash and rate limits.

Enter the Batch API. Major providers like OpenAI and Anthropic offer dedicated batch endpoints that cut your API costs by 50% while giving you significantly higher rate limits. The catch? Results are processed asynchronously within a 24-hour window (though they often finish much faster).

In this deep dive, we’ll walk through how the Batch API works and how to write a production-ready Python pipeline to process 1,000 prompts for half the price.

Why Use the Batch API?

50% Discount: You pay exactly half price per token compared to standard REST calls.
Higher Rate Limits: Batch requests sit in a separate queue, meaning you don't hit standard requests-per-minute (RPM) or tokens-per-minute (TPM) ceilings.
Fewer Retries: The provider handles queueing and transient network errors internally.

Step 1: Preparing Your Batch File

The Batch API expects a .jsonl (JSON Lines) file where each line represents an individual API request payload with a unique custom_id.

Here is how you generate a 1,000-prompt JSONL file and submit it using Python and the official openai SDK:

import json
from openai import OpenAI

client = OpenAI()

# Generate 1,000 sample prompts
dataset = [
    f"Extract the main sentiment and product features from review #{i}: 'The battery life is fantastic, but charging is slow.'"
    for i in range(1, 1001)
]

# Write tasks to a local JSONL file
batch_filename = "tasks.jsonl"
with open(batch_filename, "w") as f:
    for idx, prompt in enumerate(dataset):
        task = {
            "custom_id": f"req-review-{idx}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o-mini",
                "messages": [
                    {"role": "system", "content": "You are a concise data extraction assistant."},
                    {"role": "user", "content": prompt}
                ],
                "max_tokens": 100
            }
        }
        f.write(json.dumps(task) + "\n")

# Upload the batch file to OpenAI
uploaded_file = client.files.create(
    file=open(batch_filename, "rb"),
    purpose="batch"
)

# Launch the batch job
batch_job = client.batches.create(
    input_file_id=uploaded_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h", # Currently the standard window
    metadata={"job_type": "sentiment_analysis_v1"}
)

print(f"Batch Job Created Successfully!")
print(f"Batch ID: {batch_job.id}")
print(f"Status: {batch_job.status}")

Step 2: Polling for Completion and Downloading Results

Once the batch is submitted, you need a mechanism to monitor status. You can either listen for webhooks (if supported by your stack) or run a simple polling script.

Here is a robust script to poll job status and save the output when finished:

import time
from openai import OpenAI

client = OpenAI()

# Replace with your actual batch ID from Step 1
BATCH_ID = "batch_abc123xyz"

def retrieve_batch_results(batch_id: str):
    while True:
        job = client.batches.retrieve(batch_id)
        counts = job.request_counts
        print(f"Status: {job.status} | Completed: {counts.completed}/{counts.total} | Failed: {counts.failed}")

        if job.status == "completed":
            print("\nBatch processing complete! Downloading output...")

            # Fetch the completed output file content
            output_content = client.files.content(job.output_file_id)

            # Save results locally
            with open("results.jsonl", "wb") as f:
                f.write(output_content.content)

            print("Saved results to results.jsonl")
            break

        elif job.status in ["failed", "canceled", "expired"]:
            print(f"\nBatch failed with status: {job.status}")
            if job.error_file_id:
                errors = client.files.content(job.error_file_id)
                print(errors.text)
            break

        # Poll every 60 seconds
        time.sleep(60)

retrieve_batch_results(BATCH_ID)

Parsing the Output

The output file (results.jsonl) mirrors your input file. Each JSON object contains the original custom_id, allowing you to map responses directly back to your database records:

{
  "id": "batch_req_123",
  "custom_id": "req-review-0",
  "response": {
    "status_code": 200,
    "body": {
      "choices": [
        {
          "message": {
            "content": "Sentiment: Positive. Features: Battery life (good), Charging speed (slow)."
          }
        }
      ]
    }
  },
  "error": null
}

Practical Takeaways & Best Practices

Validate your JSONL locally: One syntax error on line 500 won't invalidate the whole file, but bad prompt parameters inside body will cause individual request failures. Always test a 5-item batch first.
Use custom_id wisely: Pass primary keys (e.g., user_12345 or doc_9876) into the custom_id field so you don't need complex mapping logic when saving responses.
Set realistic expectations: Most batch jobs complete within 1 to 3 hours, but the SLA is 24 hours. Don't use Batch API for user-facing, real-time features.

Conclusion

If your workload isn't constrained by live user interaction, using standard synchronous endpoints is leaving money on the table. Switching to the Batch API takes under an hour to write and slashes your LLM bill by 50% overnight.

Have you integrated batch processing into your LLM pipelines yet? Drop a comment below with your setup or any edge cases you've run into!

Build an Autonomous Agent in Under 50 Lines of Python with Claude 3.5 Sonnet

syncore — Sat, 30 May 2026 12:31:22 +0000

LLMs are incredibly smart, but they suffer from a major limitation: they are frozen in time. They don't know today's stock prices, they can't access your private database, and they can't execute calculations reliably.

To bridge this gap, we use Tool Use (also known as function calling).

By giving Claude access to external tools, we transform it from a passive text generator into an autonomous agent capable of fetching real-time data, executing code, and solving complex workflows step-by-step.

In this tutorial, we will build a fully autonomous financial agent in under 50 lines of clean Python code using the official Anthropic SDK and Claude 3.5 Sonnet.

The Agent Blueprint

An autonomous agent operates in a loop:

Receive a user prompt.
Decide if an external tool is needed to answer the prompt.
Execute the tool locally if requested, and capture the output.
Feed the tool results back to Claude.
Repeat until Claude has enough information to deliver the final answer.

Let’s write the code to make this happen.

Step 1: Define the Tools and Client

First, we need to set up our Anthropic client and define the tools our agent can use. For this example, we'll create a mock tool that retrieves stock prices.

Make sure you have the SDK installed:

pip install anthropic

Here is the setup code:

import os
import json
import anthropic

# Initialize the client (Make sure ANTHROPIC_API_KEY is in your environment)
client = anthropic.Anthropic()

# Define the local Python function the agent can call
def get_stock_price(ticker: str) -> str:
    mock_market = {"AAPL": "175.50", "MSFT": "420.20", "GOOG": "150.10"}
    price = mock_market.get(ticker.upper(), "unknown")
    return json.dumps({"ticker": ticker, "price": price})

# Define the schema Claude needs to understand how to use the tool
tools = [{
    "name": "get_stock_price",
    "description": "Get the current stock price for a given ticker symbol.",
    "input_schema": {
        "type": "object",
        "properties": {
            "ticker": {"type": "string", "description": "The stock ticker symbol, e.g. AAPL"}
        },
        "required": ["ticker"]
    }
}]

Step 2: Build the Autonomous Loop

Now, let's write the core execution loop. This function will continuously handle Claude’s requests to run tools until Claude decides it has enough information to formulate a final response.

def run_agent(prompt: str):
    messages = [{"role": "user", "content": prompt}]
    print(f"User: {prompt}\n" + "="*40)

    while True:
        # Send current conversation context to Claude
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

        # Append Claude's response to the conversation history
        messages.append({"role": "assistant", "content": response.content})

        # Check if Claude requested a tool call
        tool_use = next((block for block in response.content if block.type == "tool_use"), None)

        if not tool_use:
            # No tool call requested; Claude has answered the prompt
            final_text = next((b.text for b in response.content if b.type == "text"), "")
            print(f"Agent: {final_text}")
            break

        # Execute the tool requested by Claude
        print(f"[Tool Call] Running {tool_use.name} with arguments: {tool_use.input}")

        if tool_use.name == "get_stock_price":
            tool_result = get_stock_price(tool_use.input["ticker"])
        else:
            tool_result = "Error: Tool not found."

        # Send the tool output back to Claude
        messages.append({
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_id := tool_use.id,
                "content": tool_result
            }]
        })

# Execute the agent
run_agent("I own 10 shares of AAPL and 5 shares of MSFT. What is my total portfolio value?")

How It Works Under the Hood

When you run this script, here is what happens step-by-step:

The Prompt: You ask Claude to calculate your portfolio value. Claude doesn't know the current prices of Apple or Microsoft, but it notices it has access to the get_stock_price tool.
The Decision: Claude responds with a tool_use request for AAPL.
The Execution: Our Python code intercepts this request, runs the local get_stock_price("AAPL") function, and sends the output back to Claude.
The Loop Continues: Claude realizes it still needs the price of MSFT. It requests a second tool call. Our code executes it and sends the result back.
The Resolution: With both prices in hand, Claude performs the math autonomously and prints the final, accurate calculation.

Why Claude 3.5 Sonnet?

While other models support function calling, Claude 3.5 Sonnet is uniquely suited for autonomous agents because of its:

High Tool Accuracy: It rarely hallucinates arguments or formats JSON incorrectly.
Complex Multi-Step Reasoning: It handles nested logic and multiple sequential tool calls with ease.
Speed: It processes context quickly, keeping the agent's response loop feeling snappy and real-time.

Practical Takeaways

Always Validate Input: Never trust tool arguments blindly. If a tool writes to a database or runs a shell command, validate and sanitize the input first.
Handle Errors Gracefully: If your local function fails, send the error back to Claude as the tool_result. Claude can often self-correct and try a different approach.
Set Iteration Limits: In production, use a counter to limit the maximum number of loops (e.g., max 5 tool calls) to prevent infinite loops and runaway API costs.

What Will You Build Next?

Now that you have a working agent template in under 50 lines of code, the possibilities are endless. You can swap out the stock price tool for a database query tool, a web scraper, or an API client for Slack or GitHub.

What tools are you going to build for your Claude agent? Let me know in the comments below!

Slash Your Claude API Costs by 90% with Prompt Caching: A Practical Guide

syncore — Fri, 29 May 2026 19:08:02 +0000

If you are building production-grade AI applications, you already know the pain of LLM API bills. As your context grows—whether you are feeding Claude large codebases, legal documents, or long chat histories—the cost of input tokens scales linearly.

But it doesn't have to.

Anthropic's Prompt Caching feature allows you to cache frequently used context. Instead of paying full price to process the same system instructions or reference documents over and over, you can reuse them for a 90% discount on input tokens, while simultaneously cutting latency by up to 80%.

In this hands-on guide, you will learn how to implement prompt caching in Python, understand the pricing dynamics, and walk away with production-ready code.

How Prompt Caching Works (The TL;DR)

Normally, every time you send a request to Claude, the API processes your entire prompt from scratch.

With Prompt Caching, you define specific cache breakpoints in your prompt. When Claude processes a request containing a breakpoint, it saves that prefix of the prompt to a temporary cache (with a 5-minute Time-To-Live, refreshed on every hit).

Here is the pricing breakdown for Claude 3.5 Sonnet:

Base Input Tokens: $3.00 / million tokens
Cache Write (Creation) Tokens: $3.75 / million tokens (a 25% premium to write to cache)
Cache Read (Hit) Tokens: $0.30 / million tokens (a 90% saving)

If you query a 50,000-token document 10 times, you go from paying $1.50 per request to just $0.15 per request after the first turn.

Implementation 1: Caching Large Reference Documents

To use prompt caching, you must target a minimum token threshold. For Claude 3.5 Sonnet and Claude 3 Haiku, the minimum cacheable prompt size is 1,024 tokens (roughly 750 words). For Claude 3 Opus, it is 2,048 tokens.

Here is how to cache a massive reference document using the Python SDK. We will use the cache_control parameter with the type ephemeral.

import os
from anthropic import Anthropic

# Initialize the client (ensure ANTHROPIC_API_KEY is in your environment)
client = Anthropic()

# Let's simulate a large knowledge base (must be > 1024 tokens to cache)
large_knowledge_base = """
SYSTEM RULES AND KNOWLEDGE BASE (v4.2):
[Imagine 2,000 words of complex enterprise documentation, API specs, or code guidelines here...]
""" * 50 

def query_with_cached_context(user_query: str):
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=500,
        temperature=0,
        system=[
            {
                "type": "text",
                "text": large_knowledge_base,
                # This flag tells Anthropic to cache everything up to this block
                "cache_control": {"type": "ephemeral"}
            }
        ],
        messages=[
            {"role": "user", "content": user_query}
        ]
    )

    # Inspect token usage to verify caching behavior
    usage = response.usage
    print(f"Response: {response.content[0].text[:100]}...")
    print(f"Input Tokens: {usage.input_tokens}")
    print(f"Cache Creation Tokens (Write): {getattr(usage, 'cache_creation_input_tokens', 0)}")
    print(f"Cache Read Tokens (Hit): {getattr(usage, 'cache_read_input_tokens', 0)}")
    print("-" * 40)

# First run: Cache Miss (Creates the cache)
print("--- Run 1 (Cache Miss / Creation) ---")
query_with_cached_context("What version of the system rules is this?")

# Second run: Cache Hit (Reads from cache)
print("--- Run 2 (Cache Hit) ---")
query_with_cached_context("Summarize the main rules in one sentence.")

When you run this, the first request will show a high number of cache_creation_input_tokens. The second request (executed within 5 minutes) will show those same tokens registered under cache_read_input_tokens, billing you at the 90% discounted rate.

Implementation 2: Multi-Turn Conversations

Prompt caching is incredibly powerful for chatbots. By caching the conversation history, you only pay full price for the latest user message.

To make this work, place your cache breakpoint at the end of the conversation history. Anthropic allows up to 4 cache breakpoints per request.

from anthropic import Anthropic

client = Anthropic()

# We start a conversation and cache the history as it grows
conversation_history = [
    {"role": "user", "content": "Let's write a python script step-by-step. First, write a function to fetch data from an API."},
    {"role": "assistant", "content": "Here is a basic function using `requests`:\n\n```

python\nimport requests\ndef fetch_data(url):\n    return requests.get(url).json()\n

```"},
    # We place the cache breakpoint on the most recent message to keep the history cached
    {
        "role": "user", 
        "content": "Now, add error handling to that function.",
        "cache_control": {"type": "ephemeral"}
    }
]

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=800,
    messages=conversation_history
)

print(f"Assistant: {response.content[0].text}")
print(f"Cache Hits: {getattr(response.usage, 'cache_read_input_tokens', 0)} tokens")

Practical Takeaways for Developers

To get the most out of prompt caching, structure your API calls with these rules in mind:

Order Matters: Claude processes prompts from top to bottom. If any character changes before a cache breakpoint, the entire cache after that point is invalidated. Always put your static content (system prompts, documentation, instructions) at the very beginning, and dynamic content (user queries) at the end.
Mind the Limit: Caching only triggers if your prompt prefix meets the minimum token limit (1,024 for Sonnet/Haiku). Small prompts will not benefit from caching.
Track Your Metrics: Always log cache_read_input_tokens and cache_creation_input_tokens in your telemetry. This ensures you can audit your actual cost savings.

What will you build?

With a 90% cost reduction on large contexts, you can now build features that were previously cost-prohibitive: codebase-wide QA bots, real-time document synthesizers, and highly contextual agents.

Are you already using prompt caching in your projects? Share your cost-saving wins or ask your integration questions in the comments below!

How to Build a Production-Ready Claude Chatbot in Python

syncore — Fri, 29 May 2026 18:27:02 +0000

Anthropic's Claude 3.5 Sonnet has set a new benchmark for intelligence, speed, and cost-effectiveness. But there is a massive gulf between running a simple API script and deploying a production-ready chatbot.

In a production environment, your chatbot needs to handle state (conversation memory), stream responses in real-time to prevent user drop-off, and gracefully handle API rate limits and network failures.

In this guide, we will build a production-grade Claude chatbot in Python using the official Anthropic SDK, complete with memory management, streaming, and robust error handling.

Prerequisites

First, make sure you have the Anthropic Python SDK installed and your API key set as an environment variable.

pip install anthropic
export ANTHROPIC_API_KEY="your-api-key-here"

Step 1: Building the Core Streaming and Memory Engine

A production chatbot cannot make users wait 10 seconds for a full paragraph to generate. We must stream responses chunk-by-chunk. Additionally, we need to maintain a thread history so Claude remembers the context of the conversation.

Here is the core engine utilizing Python generators to stream tokens in real-time:

import os
from typing import Generator, List, Dict
from anthropic import Anthropic

class ClaudeChatEngine:
    def __init__(self, system_prompt: str = "You are a helpful, concise assistant."):
        # Initialize the client. It automatically looks for ANTHROPIC_API_KEY in env
        self.client = Anthropic()
        self.model = "claude-3-5-sonnet-20241022"
        self.system_prompt = system_prompt
        self.history: List[Dict[str, str]] = []

    def send_message(self, user_message: str) -> Generator[str, None, None]:
        """
        Sends a message to Claude, updates history, and yields tokens as they arrive.
        """
        # Append user message to state history
        self.history.append({"role": "user", "content": user_message})

        assistant_response = ""

        # Initiate a streaming request
        with self.client.messages.stream(
            model=self.model,
            max_tokens=1024,
            system=self.system_prompt,
            messages=self.history
        ) as stream:
            for text in stream.text_stream:
                assistant_response += text
                yield text

        # Append Claude's completed response to history to maintain context
        self.history.append({"role": "assistant", "content": assistant_response})

# Usage Example
if __name__ == "__main__":
    bot = ClaudeChatEngine()
    print("Chatbot initialized. Type 'exit' to quit.\n")

    while True:
        user_input = input("You: ")
        if user_input.lower() == "exit":
            break

        print("Claude: ", end="", flush=True)
        for token in bot.send_message(user_input):
            print(token, end="", flush=True)
        print("\n")

Step 2: Making it Production-Ready (Error Handling & Resilience)

The code above works perfectly under ideal conditions. But in production, network requests fail, APIs rate-limit you, and unexpected errors occur.

To make this production-ready, we need to wrap our API calls in a resilient layer that catches:

RateLimitError: When you exceed your tokens-per-minute (TPM) or requests-per-minute (RPM).
APIConnectionError: When network issues prevent connection to Anthropic's servers.
APIStatusError: When the API returns a non-200 HTTP code (e.g., overloaded servers or invalid requests).

Let's refactor our engine to handle these scenarios gracefully, incorporating exponential backoff for rate limits.

import time
import logging
from anthropic import (
    Anthropic, 
    APIConnectionError, 
    RateLimitError, 
    APIStatusError
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ClaudeChat")

class ResilientClaudeChatEngine(ClaudeChatEngine):
    def send_message_safe(self, user_message: str, max_retries: int = 3) -> Generator[str, None, None]:
        """
        Sends a message with exponential backoff for rate limits and robust error handling.
        """
        self.history.append({"role": "user", "content": user_message})
        assistant_response = ""
        delay = 1.0  # Initial delay for backoff

        for attempt in range(max_retries):
            try:
                with self.client.messages.stream(
                    model=self.model,
                    max_tokens=1024,
                    system=self.system_prompt,
                    messages=self.history
                ) as stream:
                    for text in stream.text_stream:
                        assistant_response += text
                        yield text

                # If we successfully finished streaming, break out of retry loop
                break

            except RateLimitError as e:
                logger.warning(f"Rate limited (429). Retrying in {delay}s... Error: {e}")
                if attempt == max_retries - 1:
                    yield "\n[Error: The system is currently busy. Please try again in a moment.]"
                    return
                time.sleep(delay)
                delay *= 2  # Exponential backoff

            except APIConnectionError as e:
                logger.error(f"Failed to connect to [Anthropic API](https://console.anthropic.com): {e}")
                yield "\n[Error: Connection issue. Please check your internet connection.]"
                return

            except APIStatusError as e:
                logger.error(f"Anthropic API returned status code {e.status_code}: {e.message}")
                yield f"\n[Error: An API error occurred (Status {e.status_code}).]"
                return

            except Exception as e:
                logger.error(f"Unexpected error: {str(e)}")
                yield "\n[Error: An unexpected system error occurred.]"
                return

        # Save the history only if we got a successful response
        if assistant_response:
            self.history.append({"role": "assistant", "content": assistant_response})

Key Takeaways for Production

Keep Context Under Control: Claude 3.5 Sonnet has a 200k context window, but sending the entire history on every message increases latency and costs. Implement a sliding window or summarization strategy if your chat sessions exceed 20-30 messages.
System Prompts: Always pass system-level instructions (like persona, guardrails, and output formatting constraints) via the system parameter, not as a user message. This ensures the model adheres strictly to your rules.
Environment Variables: Never hardcode your ANTHROPIC_API_KEY. Use env vars or secret managers (like AWS Secrets Manager or HashiCorp Vault) to inject keys at runtime.

What's Next?

You now have a production-ready core engine for a Claude chatbot. Your next step is to wrap this engine in an API layer like FastAPI to expose it to your frontend, or deploy it directly to a serverless environment.

Are you building with Claude? Let me know in the comments below what challenges you're facing with context windows or latency!