DEV Community: syncore

Asyncio Explained Simply: A Beginner's Guide to Python Async

syncore — Mon, 12 Oct 2026 19:18:17 +0000

4 min read · 756 words

Imagine you are cooking breakfast. You put bread in the toaster, crack two eggs into a frying pan, and brew coffee.

If you cooked synchronously, you would stand motionless in front of the toaster for two minutes. Only after the toast popped up would you turn to crack the eggs. You'd wait for the eggs to finish before grinding coffee beans. That is a terrible way to make breakfast, and it is an equally terrible way to handle I/O-bound code.

Python's asyncio solves this problem. It allows a single thread to pause an operation while waiting for an external system (like a database, an API, or a file read) and do other work in the meantime.

Here is what you need to understand asyncio without getting lost in academic jargon.

The Three Pillars of Asyncio

Before writing code, lock in these three concepts:

The Event Loop: The central engine. It tracks tasks and delegates execution. If Task A is waiting for a network response, the event loop parks Task A and resumes Task B.
Coroutines (async def): Functions declared with async def. When you call one, Python doesn't execute it immediately; it returns a coroutine object ready to be scheduled on the event loop.
Pausing Points (await): The await keyword yields execution back to the event loop. It tells Python: "Go do other work until this operation produces a result."

1. Your First Asyncio Script

Let's look at how sequential code compares to concurrent asynchronous execution using asyncio.sleep, which simulates network latency.

import asyncio
import time

async def fetch_data(source_id: int, delay: float):
    print(f"Fetching from source {source_id}...")
    await asyncio.sleep(delay)  # Yields control back to the loop
    print(f"Received data from source {source_id}!")
    return {"source": source_id, "status": "ok"}

async def main():
    start = time.perf_counter()

    # Schedule three tasks to run concurrently
    results = await asyncio.gather(
        fetch_data(1, 2.0),
        fetch_data(2, 1.0),
        fetch_data(3, 1.5),
    )

    elapsed = time.perf_counter() - start
    print(f"Results: {results}")
    print(f"Total time taken: {elapsed:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

If these tasks ran sequentially, this script would take 2.0 + 1.0 + 1.5 = 4.5 seconds. Run this code, and it finishes in roughly 2.0 seconds—the duration of the longest individual task.

asyncio.gather wraps all three coroutines, submits them to the event loop, and waits until all of them finish.

2. Real-World Async: Concurrent API Requests

The real power of asyncio shines when interacting with third-party web services. When calling remote APIs, your CPU spends 99% of its time idling, waiting on network packets.

Here is an example making multiple LLM requests concurrently using the modern Anthropic Python SDK:

import asyncio
import anthropic

# Initialize the asynchronous client
client = anthropic.AsyncAnthropic()

async def generate_summary(topic: str) -> str:
    prompt = f"Provide a one-sentence summary of the core benefit of {topic}."

    response = await client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=150,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text

async def main():
    topics = ["Docker", "Asyncio", "Redis"]

    # Launch all requests concurrently
    tasks = [generate_summary(topic) for topic in topics]
    summaries = await asyncio.gather(*tasks)

    for topic, summary in zip(topics, summaries):
        print(f"\n--- {topic} ---\n{summary}")

if __name__ == "__main__":
    asyncio.run(main())

Instead of waiting for Docker to finish before dispatching Asyncio, the event loop fires all three requests nearly simultaneously. All responses stream back as they resolve.

Three Common Traps to Avoid

1. Calling Blocking Code Inside Coroutines

Async code only yields control when you explicitly await an asynchronous call. If you drop time.sleep(5) or standard requests.get() inside an async def function, you freeze the entire event loop. No other task can run. Always use non-blocking equivalents like asyncio.sleep() or httpx.AsyncClient().

2. Forgetting the `await` Keyword

If you call an async function without await:

data = fetch_data(1, 2.0)
print(data)
# Output: <coroutine object fetch_data at 0x104...>

Python will print a coroutine object instead of your expected payload, and issue a RuntimeWarning: coroutine 'fetch_data' was never awaited.

3. Thinking Async Speeds Up Heavy Math

Asyncio does not offer parallel CPU processing. It is designed purely for I/O-bound operations (databases, microservices, file systems). If you are processing images, training machine learning models, or computing primes, reach for Python's multiprocessing module instead.

Wrapping Up

Asyncio doesn't make individual network requests faster—it stops your program from sitting idle while waiting for them. By organizing I/O around async def, await, and asyncio.gather, you can dramatically increase your application's throughput with minimal overhead.

What part of async Python trips you up the most—cancellation, task groups, or debugging? Let me know in the comments below!

FastAPI vs Flask in 2026: Which Should You Pick?

syncore — Mon, 05 Oct 2026 18:47:31 +0000

4 min read · 882 words

If you ask Python developers which backend framework to start with today, you will still spark a heated debate. For years, Flask was the undisputed champion of lightweight Python backends. Then FastAPI arrived, leveraging Python's type annotations and asynchronous architecture to challenge the status quo.

Now in 2026, the dust has settled. Flask has incorporated asynchronous capabilities, while FastAPI has evolved into an enterprise-grade powerhouse powering high-throughput microservices and generative AI backends.

So, which one should you pick for your next project? Let’s break down their modern capabilities, look at real code, and set up a pragmatic decision framework.

The Landscape in 2026: ASGI vs. WSGI

The fundamental difference between the two frameworks lies in their execution foundation:

FastAPI is built on ASGI (Asynchronous Server Gateway Interface) via Starlette and Uvicorn. It was designed from day one to handle non-blocking I/O, concurrent streaming connections, and WebSockets natively.
Flask was built on WSGI (Web Server Gateway Interface) via Werkzeug. While Flask now supports async def routes, its internal design remains rooted in synchronous paradigms. It uses an async-to-sync bridge under the hood, meaning it does not match native ASGI concurrency when under heavy I/O loads.

Furthermore, FastAPI tightly couples with Pydantic for data parsing, automatic serialization, and schema validation. Flask relies on manual parsing or community extensions like Marshmallow.

FastAPI in Action: An Async AI Gateway

A massive chunk of modern backend work involves orchestration—calling external APIs, querying databases, and processing streaming LLM responses. FastAPI thrives here.

Here is a complete, production-ready FastAPI endpoint serving an AI task using Anthropic's Python SDK:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import anthropic

app = FastAPI(title="FastAPI AI Gateway")
client = anthropic.Anthropic()

class PromptRequest(BaseModel):
    prompt: str = Field(..., min_length=5, description="Input prompt for the model")

class PromptResponse(BaseModel):
    result: str

@app.post("/analyze", response_model=PromptResponse)
async def analyze_text(payload: PromptRequest):
    try:
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1024,
            messages=[{"role": "user", "content": payload.prompt}],
        )
        output_text = response.content[0].text
        return PromptResponse(result=output_text)
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"Inference error: {exc}")

Why this shines:

Self-Documenting: Navigating to /docs instantly yields an interactive Swagger UI with verified input/output schemas.
Robust Validation: If an incoming payload lacks the prompt field or sends an integer, FastAPI rejects it before execution reaches your route.

Flask in Action: The Minimalist Approach

Flask’s greatest strength has always been non-prescriptive minimalism. You don’t need classes or Pydantic models to get an endpoint running.

Here is the equivalent endpoint written in Flask:

from flask import Flask, request, jsonify
import anthropic

app = Flask(__name__)
client = anthropic.Anthropic()

@app.post("/analyze")
def analyze_text():
    data = request.get_json(silent=True)
    if not data or "prompt" not in data or len(data["prompt"]) < 5:
        return jsonify({"error": "Field 'prompt' must be at least 5 characters"}), 400

    try:
        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=1024,
            messages=[{"role": "user", "content": data["prompt"]}],
        )
        return jsonify({"result": response.content[0].text}), 200
    except Exception as exc:
        return jsonify({"error": f"Inference error: {exc}"}), 500

if __name__ == "__main__":
    app.run(port=5000)

Why this works:

Simplicity: No need to understand type hints or schema definitions if you just want to process a payload and return JSON.
Predictability: Flask’s simple request context is easy to debug, test, and inject into legacy systems.

Direct Comparison

Feature	FastAPI	Flask
Architecture	Native ASGI (Uvicorn / Starlette)	WSGI with async route wrappers
Data Validation	Built-in via Pydantic	Manual or third-party extensions
API Documentation	Automatic OpenAPI & Swagger	Requires extensions (e.g., Flasgger)
Concurrency	High out-of-the-box performance	Medium (limited by thread workers)
Learning Curve	Moderate (Requires modern type hints)	Very gentle

When to Choose FastAPI

Choose FastAPI if:

You are building an API-first service: If you are feeding frontends (React, Vue, mobile apps) or building microservices, the automated OpenAPI docs alone save dozens of hours.
You deal with high I/O concurrency: When your application coordinates multiple databases, external third-party APIs, or streaming pipelines, ASGI handles high connection counts with low overhead.
You want strict type safety: FastAPI catches bugs during development instead of during runtime errors in production.

When to Choose Flask

Choose Flask if:

You need server-side rendering (SSR): Flask paired with Jinja2 is still one of the fastest, cleanest ways to ship traditional multi-page web applications.
You are building a quick script or internal tool: If you need a 30-line micro-endpoint inside a data science utility, Flask introduces almost zero conceptual overhead.
You depend on a mature, legacy WSGI stack: Many enterprise environments have heavily customized WSGI middleware and monitoring agents that are already configured for Flask.

The Verdict

In 2026, FastAPI is the clear default for new REST and GraphQL APIs. Its integration with modern Python type hints, automatic schema generation, and high-performance async runtime make it the standard for contemporary microservices.

However, Flask is far from obsolete. It remains the gold standard for quick prototypes, classic server-rendered web applications, and teams prioritizing simplicity over strict schema contracts.

What does your stack look like this year? Have you migrated your legacy Flask apps to FastAPI, or are you sticking with the battle-tested WSGI veteran? Let us know in the comments below!

10 Python One-Liners That Will Blow Your Mind (With Real Examples)

syncore — Mon, 28 Sep 2026 17:51:57 +0000

3 min read · 525 words

Python is famous for letting you express complex ideas in just a few lines of code. But some one-liners go beyond clean syntax—they completely change how you think about solving problems.

Whether you're processing data, handling files, or automating tedious workflows, here are 10 Python one-liners that pack a serious punch, plus a bonus look at how to supercharge your Python scripts using the latest Anthropic SDK.

1. Start a Quick Web Server

Need to share files from your current directory instantly? No need to install Express or Flask.

python3 -m http.server 8080

Takeaway: Instantly spins up a static file server in the current directory.

2. Flatten a Nested List

Dealing with deeply nested lists from an API response? Unpack them recursively in one go.

flatten = lambda l: [item for sublist in l for item in (flatten(sublist) if isinstance(sublist, list) else [sublist])]
print(flatten([1, [2, [3, 4], 5], 6]))  # [1, 2, 3, 4, 5, 6]

3. Read a File Into a List of Lines (Without Newlines)

Forget .readlines() leaving trailing \n characters everywhere.

lines = [line.strip() for line in open('data.txt')]

4. Transpose a 2D Matrix

Flipping rows and columns is a classic interview problem. Python's zip unpacks it effortlessly.

matrix = [[1, 2, 3], [4, 5, 6]]
transposed = list(map(list, zip(*matrix)))
print(transposed)  # [[1, 4], [2, 5], [3, 6]]

5. Simple In-Memory Profiling

Want to measure how long a code block takes without importing timeit boilerplate?

import time; t = time.time(); [i**2 for i in range(10_000_000)]; print(f"Took: {time.time() - t:.4f}s")

6. Remove Duplicates While Preserving Order

set() destroys order. Use dict.fromkeys() to keep things deterministic.

unique_items = list(dict.fromkeys([1, 3, 2, 1, 4, 2]))
print(unique_items)  # [1, 3, 2, 4]

7. Merge Two Dictionaries (Python 3.9+)

The merge operator | makes dictionary combination cleaner than ever.

merged = {"a": 1, "b": 2} | {"b": 99, "c": 4}
print(merged)  # {'a': 1, 'b': 99, 'c': 4}

8. Generate a Random Password

No external dependencies required for basic string generation.

import random, string
password = "".join(random.choices(string.ascii_letters + string.digits + string.punctuation, k=16))

9. Check if a String is a Palindrome

Strip spaces, lowercase, and compare it reversed in a single slice.

is_palindrome = lambda s: (c := ''.join(s.lower().split())) == c[::-1]
print(is_palindrome("A man a plan a canal Panama"))  # True

10. Query Claude Directly from Python

Want to add AI capabilities to your scripts? Here is how to make a clean API call using the current anthropic SDK. Notice how we use claude-opus-5 with adaptive thinking enabled and zero deprecated parameters.

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    thinking={"type": "adaptive"},
    output_config={"effort": "medium"},
    messages=[{"role": "user", "content": "Explain quantum computing in one sentence."}],
)

for block in response.content:
    if block.type == "text":
        print(block.text)

Conclusion

One-liners are fantastic for readability and quick scripts, but remember the golden rule of Python: readability counts. Don't sacrifice maintainability just to fit everything onto a single line.

What is your favorite Python one-line trick? Drop it in the comments below!

Multi-Agent Workflows with Claude: Patterns and Pitfalls

syncore — Mon, 21 Sep 2026 16:04:50 +0000

3 min read · 604 words

Multi-agent systems are currently all the rage. Splitting a monolithic task—like building a web app or writing a research paper—into specialized agents (e.g., a Planner, a Coder, a Reviewer) sounds like the ultimate scaling trick.

When applied to modern LLMs, multi-agent architectures can be powerful, but they also introduce silent failure modes, runaway token usage, and frustrating debugging sessions.

Here is how to build practical, production-ready multi-agent workflows using claude-opus-5 and claude-sonnet-5, along with the patterns you should follow and the pitfalls to avoid.

The Architecture: Router-Worker Pattern

Instead of creating a tangled web where every agent talks to every other agent, stick to deterministic orchestrations. The Router-Worker pattern is one of the most reliable architectures. A primary router analyzes the user request and delegates sub-tasks to specialized worker agents.

Let's implement a clean, two-tier workflow using the official Python SDK: a Research Agent that gathers data, followed by a Synthesis Agent that writes the final report.

import anthropic

client = anthropic.Anthropic()

def run_researcher(topic: str) -> str:
    """Worker Agent 1: Gathers technical details."""
    print(f"🤖 [Researcher] Investigating: {topic}")

    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=8000,
        output_config={"effort": "medium"},
        messages=[
            {
                "role": "user",
                "content": f"Provide 3 key technical challenges and solutions for: {topic}"
            }
        ]
    )
    return response.content[0].text

def run_synthesizer(topic: str, research_data: str) -> str:
    """Worker Agent 2: Synthesizes findings into a final document."""
    print(f"🤖 [Synthesizer] Formatting final report...")

    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=16000,
        output_config={"effort": "high"},
        messages=[
            {
                "role": "user",
                "content": f"Topic: {topic}\n\nResearch Data:\n{research_data}\n\nWrite a polished markdown report."
            }
        ]
    )
    return response.content[0].text

# Execute the workflow
topic = "Scaling WebSockets in Distributed Systems"
raw_research = run_researcher(topic)
final_report = run_synthesizer(topic, raw_research)

print("\n--- FINAL OUTPUT ---\n")
print(final_report[:500] + "\n[...truncated for display...]")

Note: Notice how we omit parameters like temperature, top_p, and top_k. Modern API standards require us to steer behavior purely through prompt engineering and output_config={"effort": "..."}.

Structured JSON Outputs for Inter-Agent Communication

When agents pass data to each other, brittle string parsing will break your pipeline. Instead of letting agents reply in free-form text, enforce strict JSON schemas using output_config={"format": {...}}.

Here is how you can ensure an evaluator agent always returns a parseable evaluation object before passing it downstream:

import json
import anthropic

client = anthropic.Anthropic()

eval_schema = {
    "type": "object",
    "properties": {
        "score": {"type": "integer", "description": "Score from 1 to 10"},
        "approved": {"type": "boolean"},
        "feedback": {"type": "string"}
    },
    "required": ["score", "approved", "feedback"]
}

def evaluate_code(code_snippet: str) -> dict:
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=4000,
        output_config={"format": {"type": "json_schema", "schema": eval_schema}},
        messages=[
            {
                "role": "user",
                "content": f"Review this code for security vulnerabilities:\n\n{code_snippet}"
            }
        ]
    )

    # Guaranteed to match the JSON schema
    return json.loads(response.content[0].text)

result = evaluate_code("eval(user_input)")
print(result)

Major Pitfalls to Avoid

Infinite Agent Loops: If Agent A reviews Agent B's work, and Agent B fixes Agent A's feedback, they can get stuck looping forever. Always set a hard step limit (e.g., maximum 3 iterations) in your orchestration code.
Over-Engineering: Don't use five agents when one well-prompted call to claude-opus-5 will do. Every handoff introduces latency, token costs, and a new point of failure.
Ignoring Context Windows: While models like claude-opus-5 boast massive 1M token contexts, passing bloated conversation histories between multiple workers degrades reasoning focus. Summarize intermediate outputs before handing them off.

Summary Checklist

Model Selection: Use claude-sonnet-5 for high-volume worker tasks, claude-opus-5 for deep reasoning/synthesis, and claude-haiku-4-5 for fast routing or validation.
Control Depth: Use output_config={"effort": "low"|"medium"|"high"} instead of legacy parameters.
Enforce Schemas: Use native JSON formatting configs rather than relying on regex to parse agent conversations.

What multi-agent patterns are you building? Drop a comment below or share your experiences!

How to Add Conversational Memory to Your Claude Chatbot

syncore — Mon, 14 Sep 2026 16:01:19 +0000

3 min read · 612 words

Building a stateless LLM chatbot is easy. You send a prompt, you get a response, and the model instantly forgets everything that just happened. But real user applications demand context. Users expect the AI to remember what they said two messages ago.

In this tutorial, you’ll learn how to add robust conversational memory to a Python chatbot using the Anthropic SDK. We'll use claude-opus-5 to manage multi-turn conversations cleanly without bloating your context window.

Why LLMs Need Application-Level Memory

The Anthropic API is stateless. Every call to client.messages.create() is a completely fresh interaction. To create the illusion of memory, your application code must maintain a history array of alternating user and assistant messages and send the full history with every new request.

The Basic Chat Loop with Memory

Let's start by implementing a simple CLI chatbot that keeps track of the conversation in a local Python list.

Make sure you have the latest SDK installed:

pip install anthropic

Here is a working implementation:

import os
import anthropic

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

def run_chat_loop():
    # Initialize message history
    messages = []

    print("🤖 Claude Chatbot Initialized. Type 'exit' to quit.\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (KeyboardInterrupt, EOFError):
            break

        if user_input.lower() in ["exit", "quit"]:
            print("Goodbye!")
            break

        if not user_input:
            continue

        # Append the user's message to history
        messages.append({"role": "user", "content": user_input})

        try:
            # Send the complete conversation history to Claude
            response = client.messages.create(
                model="claude-opus-5",
                max_tokens=4000,
                messages=messages
            )

            # Extract assistant's reply
            assistant_reply = response.content[0].text
            print(f"\nClaude: {assistant_reply}\n")

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

        except Exception as e:
            print(f"\n⚠️ Error: {e}\n")

if __name__ == "__main__":
    run_chat_loop()

Handling Long Conversations (Pruning & Summarization)

Models like claude-opus-5 feature a massive 1M token context window, meaning you can carry out exceptionally long sessions. However, letting history grow infinitely increases latency and token costs.

For production apps, you should implement a rolling window or summarize old turns. Here is how you can cap your history to the last $N$ turns to keep things fast and cost-effective:

import anthropic

client = anthropic.Anthropic()

def chat_with_rolling_window(messages, new_user_message, max_history_turns=10):
    # Append the latest user message
    messages.append({"role": "user", "content": new_user_message})

    # Keep system instructions separate if you use them, 
    # but for raw message history, slice the tail.
    # Each turn consists of 1 user + 1 assistant message (2 items).
    max_items = max_history_turns * 2
    if len(messages) > max_items:
        # Always ensure the history starts with a 'user' message
        trimmed_messages = messages[-max_items:]
        if trimmed_messages[0]["role"] == "assistant":
            trimmed_messages = trimmed_messages[1:]
    else:
        trimmed_messages = messages

    response = client.messages.create(
        model="claude-sonnet-5", # Great choice for high-volume chat tasks
        max_tokens=4000,
        messages=trimmed_messages
    )

    assistant_reply = response.content[0].text

    # Update original history reference
    messages.append({"role": "assistant", "content": assistant_reply})

    return assistant_reply, messages

Pro-Tips for Production Chatbots

Drop Temperature/Top-P: Modern configurations omit temperature, top_p, and top_k. If you need to steer style or determinism, rely entirely on clear system prompts.
Use the Right Model: Use claude-opus-5 for deep reasoning tasks, and switch to claude-sonnet-5 for high-throughput, latency-sensitive applications.
Structured Outputs: If your chatbot needs to collect user data (like booking details) into memory, enforce JSON schema responses using output_config={"format": {...}} instead of attempting assistant-turn prefilling.

Conclusion

Adding memory to your Claude chatbot is as straightforward as maintaining a clean array of turns and passing it back on every request. By combining a rolling window strategy with the blazing speed and massive context of current Claude models, you can build seamless conversational experiences that scale.

What kind of AI assistant are you building? Let me know in the comments below!

Claude Opus 5 vs Sonnet 5 vs Haiku 4.5: Which Model Should You Actually Use?

syncore — Mon, 07 Sep 2026 15:14:20 +0000

3 min read · 584 words

Choosing the right AI model for your application shouldn't feel like guessing at a candy store. With the release of Anthropic's latest generation, picking the wrong model means either burning cash on unnecessary compute or frustrating users with slow, heavy responses.

Let's break down the current lineup (claude-opus-5, claude-sonnet-5, and claude-haiku-4-5), look at their pricing and context limits, and write some actual code to put them to work using the latest Python SDK.

The Current Lineup at a Glance

Model ID	Input/Output Cost (per 1M tokens)	Context Window	Best Suited For
`claude-opus-5`	$5 / $25	1,000,000	Complex reasoning, default heavy-lifting
`claude-sonnet-5`	$3 / $15	1,000,000	High-volume production workflows
`claude-haiku-4-5`	$1 / $5	200,000	Fast tasks, classification, triage

A few critical API rules have changed in this generation that you need to know before writing code:

No sampling parameters: temperature, top_p, and top_k are removed. Control output behavior purely through prompt engineering.
Extended Thinking: Use thinking={"type": "adaptive"} instead of old budget token configurations.
Reasoning Depth: Control depth using output_config={"effort": "low"|"medium"|"high"}.
Structured Output: Assistant-turn prefilling is gone; use output_config={"format": {...}} instead.

1. Claude Opus 5: The Heavy Lifter

claude-opus-5 is your default choice for deep reasoning, architectural design, complex code refactoring, and multi-step problem solving. With a massive 1-million-token context window and competitive pricing ($5/$25 per 1M tokens), it handles massive codebases or dense legal documents with ease.

Here is how to run a deep analysis task using Opus 5 with adaptive thinking enabled:

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    thinking={"type": "adaptive"},
    output_config={"effort": "high"},
    messages=[
        {
            "role": "user",
            "content": "Analyze this legacy auth module for security vulnerabilities and propose a modern OAuth2 migration plan.",
        }
    ],
)

print(response.content)

2. Claude Sonnet 5: The High-Volume Workhorse

If you're running high-throughput pipelines, customer service automation, or content generation where volume is high and margins matter, claude-sonnet-5 is your go-to model. At $3/$15 per 1M tokens with the same 1-million-token context window as Opus, it strikes the optimal balance between speed, capability, and cost.

Use Sonnet 5 when you need fast, reliable execution across thousands of daily requests.

3. Claude Haiku 4-5: Speed and Simplicity

claude-haiku-4-5 is built for speed. Priced at just $1/$5 per 1M tokens with a 200K context window, it’s ideal for intent classification, data extraction, and quick chat triage where latency is your primary bottleneck.

Here is an example using Haiku 4.5 with structured JSON output to categorize incoming support tickets instantly:

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1024,
    output_config={
        "format": {
            "type": "json_object",
            "schema": {
                "type": "object",
                "properties": {
                    "category": {"type": "string"},
                    "priority": {"type": "string", "enum": ["low", "medium", "high"]},
                    "summary": {"type": "string"},
                },
                "required": ["category", "priority", "summary"],
            },
        }
    },
    messages=[
        {
            "role": "user",
            "content": "Classify this ticket: 'My billing page is throwing a 500 error when I try to update my credit card.'",
        }
    ],
)

print(response.content)

Summary: Which one should you pick?

Use claude-opus-5 when accuracy and deep reasoning are non-negotiable.
Use claude-sonnet-5 as your daily driver for most production apps, agents, and pipelines.
Use claude-haiku-4-5 when you need instant responses, classification, or high-volume data munging on a tight budget.

Drop a comment below with how you're using the new Claude models in your stack!

Build a Fast RAG Pipeline with Claude and Local Embeddings

syncore — Mon, 24 Aug 2026 10:02:34 +0000

3 min read · 588 words

Retrieval-Augmented Generation (RAG) is the quickest way to ground Claude in your private data without fine-tuning. But running a full RAG pipeline often means wrestling with heavy external vector databases or pricey embedding APIs.

In this article, we’ll build a lean, lightning-fast RAG system using Python, sentence-transformers for local embeddings, and claude-opus-5 for answer generation. No complex cloud infrastructure required—everything runs locally until we query the model.

Prerequisites

You'll need Python 3.10+ and the official Anthropic SDK, along with a few data science libraries for handling local vector search:

pip install anthropic sentence-transformers numpy

Make sure your API key is set in your environment:

export ANTHROPIC_API_KEY="your-api-key-here"

Step 1: Generate Local Embeddings

We'll start by taking a small knowledge base of text snippets, converting them into vector embeddings using a lightweight model from Hugging Face, and storing them in memory.

import numpy as np
from sentence_transformers import SentenceTransformer

# 1. Load a fast, lightweight local embedding model
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")

# 2. Define our local knowledge base
documents = [
    "Project Apollo uses Python 3.11 and FastAPI for the backend services.",
    "The deployment pipeline for staging is triggered automatically via GitHub Actions.",
    "Database migrations are managed using Alembic. Never run raw DDL in production.",
    "To reset the local Redis cache, run 'redis-cli flushall' in your terminal."
]

# 3. Generate embeddings for our documents
doc_embeddings = embedding_model.encode(documents)

Step 2: Retrieve Relevant Context and Query Claude

When a user asks a question, we embed the query, compute cosine similarities against our document embeddings, and pull the top matches. Then, we pass those snippets directly to claude-opus-5.

Remember: current Claude models rely purely on prompt steering (no temperature, top_p, or top_k), and use thinking={"type": "adaptive"} for extended reasoning.

import anthropic

client = anthropic.Anthropic()

def search_docs(query: str, top_k: int = 2):
    # Embed the incoming query
    query_vector = embedding_model.encode(query)

    # Compute cosine similarity
    similarities = np.dot(doc_embeddings, query_vector) / (
        np.linalg.norm(doc_embeddings, axis=1) * np.linalg.norm(query_vector)
    )

    # Get top_k indices sorted by highest similarity
    top_indices = np.argsort(similarities)[::-1][:top_k]
    return [documents[i] for i in top_indices]

def ask_rag(user_query: str):
    # Retrieve relevant context locally
    retrieved_chunks = search_docs(user_query)
    context = "\n---\n".join(retrieved_chunks)

    # Construct the grounded prompt
    prompt = f"""You are a helpful technical assistant. Answer the user's question using ONLY the provided context. If you don't know the answer based on the context, say so.

Context:
{context}

Question:
{user_query}"""

    # Call Claude using the current SDK and API rules
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=4000,
        thinking={"type": "adaptive"},
        output_config={"effort": "medium"},
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content

# Test the RAG pipeline
answer = ask_rag("How do we handle database migrations?")
print(answer)

Production Considerations

While this in-memory NumPy approach works great for prototypes and small codebases, scaling up involves a few best practices:

Swap to a Vector DB: For thousands or millions of documents, replace the NumPy array with a persistent vector store like Chroma, FAISS, or pgvector.
Chunking Strategy: Instead of whole sentences, split large Markdown or PDF files into 500-token chunks with 50-token overlaps to preserve context.
Model Selection: Use claude-sonnet-5 if you're running high-volume automated Q&A pipelines where throughput is critical.

Conclusion

Building effective RAG doesn't require massive cloud stacks. By combining local embeddings with claude-opus-5, you get a secure, private, and extremely fast question-answering system that runs right inside your application workflow.

Have you built a RAG pipeline with Claude yet? Drop a comment below with your favorite embedding model or questions about the new API parameters!

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!

DEV Community: syncore

Asyncio Explained Simply: A Beginner's Guide to Python Async

The Three Pillars of Asyncio

1. Your First Asyncio Script

2. Real-World Async: Concurrent API Requests

Three Common Traps to Avoid

1. Calling Blocking Code Inside Coroutines

2. Forgetting the await Keyword

3. Thinking Async Speeds Up Heavy Math

Wrapping Up

FastAPI vs Flask in 2026: Which Should You Pick?

The Landscape in 2026: ASGI vs. WSGI

FastAPI in Action: An Async AI Gateway

Why this shines:

Flask in Action: The Minimalist Approach

Why this works:

Direct Comparison

When to Choose FastAPI

When to Choose Flask

The Verdict

10 Python One-Liners That Will Blow Your Mind (With Real Examples)

1. Start a Quick Web Server

2. Flatten a Nested List

3. Read a File Into a List of Lines (Without Newlines)

4. Transpose a 2D Matrix

5. Simple In-Memory Profiling

6. Remove Duplicates While Preserving Order

7. Merge Two Dictionaries (Python 3.9+)

8. Generate a Random Password

9. Check if a String is a Palindrome

10. Query Claude Directly from Python

Conclusion

Multi-Agent Workflows with Claude: Patterns and Pitfalls

The Architecture: Router-Worker Pattern

Structured JSON Outputs for Inter-Agent Communication

Major Pitfalls to Avoid

Summary Checklist

How to Add Conversational Memory to Your Claude Chatbot

Why LLMs Need Application-Level Memory

The Basic Chat Loop with Memory

Handling Long Conversations (Pruning & Summarization)

Pro-Tips for Production Chatbots

Conclusion

Claude Opus 5 vs Sonnet 5 vs Haiku 4.5: Which Model Should You Actually Use?

The Current Lineup at a Glance

1. Claude Opus 5: The Heavy Lifter

2. Claude Sonnet 5: The High-Volume Workhorse

3. Claude Haiku 4-5: Speed and Simplicity

Summary: Which one should you pick?

Build a Fast RAG Pipeline with Claude and Local Embeddings

Prerequisites

Step 1: Generate Local Embeddings

Step 2: Retrieve Relevant Context and Query Claude

Production Considerations

Conclusion

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

The Fundamental Rule

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

Best Practices for Sonnet:

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

Best Practices for Opus:

Quick Comparison Cheat Sheet

Practical Takeaways

What's Your Default Model?

Streaming Responses with Claude: A Complete Developer Guide

How Claude Streaming Works

Option 1: Streaming in Python

Prerequisites

Code Example: Python Console Stream

Key Elements

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

Prerequisites

Code Example: Express SSE Server

How the Frontend Consumes This

Understanding Stream Events

Essential Best Practices

Conclusion

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

Why Use the Batch API?

Step 1: Preparing Your Batch File

2. Forgetting the `await` Keyword