⬡ API & SDK/2026-03-14Advanced

Claude API Streaming & Tool Use in Production — Patterns for Parallel Calls, Error Handling, and Retry Strategies

Master production-grade streaming and tool use patterns with Claude API. Learn parallel tool calling, intelligent error handling, resilient retry strategies, and resource optimization.

Claude API¹¹⁵ Streaming⁴ Tool Use⁸ Production Patterns Error Handling⁵

✦ Premium Article

What Separates a Prototype from Production

Building reliable, production-grade systems with Claude API requires mastering streaming responses and tool use—especially when building agentic workflows that depend on multiple sequential and parallel tool calls. The difference between a prototype and a production system lies in thoughtful error handling, intelligent retry strategies, and resource optimization.

💡

The most common production failures with Claude API don't stem from Claude's core performance, but from inadequate error handling, poor retry strategies, and inefficient resource management. This guide focuses on the operational patterns that make the difference between a prototype and a bulletproof production system.

Streaming Fundamentals for Production

Understanding Stream Lifecycle

Streaming in production requires understanding the complete lifecycle of a stream, not just the happy path:

import anthropic
import json
from typing import Generator
 
client = anthropic.Anthropic()
 
def stream_with_lifecycle_tracking():
    """
    Complete streaming lifecycle including:
    - Connection establishment
    - Token streaming
    - Error states
    - Cleanup
    """
    stream_metadata = {
        "tokens_received": 0,
        "errors": [],
        "time_to_first_token": None,
        "total_duration": None
    }
 
    try:
        with client.messages.stream(
            model="claude-opus-4-6",
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": "Analyze the following data and provide insights"
                }
            ]
        ) as stream:
            # Track time to first token (crucial for UX)
            first_token_received = False
 
            for text in stream.text_stream:
                if not first_token_received:
                    stream_metadata["time_to_first_token"] = stream.get_final_message().id
                    first_token_received = True
 
                stream_metadata["tokens_received"] += len(text.split())
                # Process/yield tokens as they arrive
                yield text
 
            # Capture final message metadata
            final_message = stream.get_final_message()
            stream_metadata["usage"] = {
                "input_tokens": final_message.usage.input_tokens,
                "output_tokens": final_message.usage.output_tokens
            }
 
    except anthropic.APIError as e:
        stream_metadata["errors"].append({
            "type": str(type(e).__name__),
            "message": str(e)
        })
        raise
    finally:
        # Always log stream metadata for monitoring
        log_stream_metrics(stream_metadata)

Buffering Strategies for Real-time Systems

Different systems require different buffering approaches:

import asyncio
from collections import deque
from typing import AsyncGenerator
 
class StreamBuffer:
    """Intelligent buffering for streaming responses."""
 
    def __init__(self, buffer_size: int = 5, flush_timeout: float = 0.1):
        self.buffer = deque(maxlen=buffer_size)
        self.flush_timeout = flush_timeout
 
    async def stream_with_buffer(
        self,
        messages: list
    ) -> AsyncGenerator[str, None]:
        """
        Stream with adaptive buffering:
        - Small buffers for low-latency applications
        - Large buffers for throughput optimization
        - Time-based flushing for consistency
        """
        client = anthropic.AsyncAnthropic()
        accumulated_text = ""
        last_flush = asyncio.get_event_loop().time()
 
        async with client.messages.stream(
            model="claude-opus-4-6",
            max_tokens=2048,
            messages=messages
        ) as stream:
            async for text in stream:
                self.buffer.append(text)
                accumulated_text += text
 
                current_time = asyncio.get_event_loop().time()
                should_flush = (
                    len(self.buffer) == self.buffer.maxlen or
                    (current_time - last_flush) > self.flush_timeout
                )
 
                if should_flush:
                    # Emit buffered content
                    yield accumulated_text
                    accumulated_text = ""
                    last_flush = current_time
 
            # Flush remaining content
            if accumulated_text:
                yield accumulated_text

⚠️

Never buffer indefinitely. Always implement time-based flush limits to prevent memory growth and ensure responsiveness, even if no new tokens arrive.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦The ±50% jitter backoff pattern that cut my 529 overloaded_error final-failure rate from 2.1% to 0.18%

✦How I route parallel tool calls by result size to hit 5.1s avg latency and a 0.4% timeout rate

✦Why conversation checkpointing — built before optimization — cut my re-run cost by ~60%

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Parallel Tool Calling Patterns

Coordinating Multiple Tool Calls

Claude can suggest multiple tool calls in a single response. Production systems must handle concurrent execution properly:

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Any
 
class ToolExecutor:
    """Execute tool calls safely with proper concurrency management."""
 
    def __init__(self, max_concurrent: int = 5):
        self.max_concurrent = max_concurrent
        self.executor = ThreadPoolExecutor(max_workers=max_concurrent)
 
    async def execute_parallel_tools(
        self,
        tool_calls: list[dict]
    ) -> list[dict]:
        """
        Execute multiple tool calls in parallel with:
        - Concurrency limits
        - Error isolation
        - Timeout protection
        - Result tracking
        """
        results = []
        semaphore = asyncio.Semaphore(self.max_concurrent)
 
        async def execute_single_tool(tool_call: dict) -> dict:
            async with semaphore:
                try:
                    # Execute with timeout
                    result = await asyncio.wait_for(
                        self._execute_tool(tool_call),
                        timeout=30.0
                    )
                    return {
                        "tool_use_id": tool_call["id"],
                        "content": result,
                        "error": None
                    }
                except asyncio.TimeoutError:
                    return {
                        "tool_use_id": tool_call["id"],
                        "content": None,
                        "error": "Tool execution timeout (30s)"
                    }
                except Exception as e:
                    return {
                        "tool_use_id": tool_call["id"],
                        "content": None,
                        "error": f"Tool execution failed: {str(e)}"
                    }
 
        # Execute all tools concurrently
        tasks = [execute_single_tool(tc) for tc in tool_calls]
        results = await asyncio.gather(*tasks)
 
        return results
 
    async def _execute_tool(self, tool_call: dict) -> Any:
        """Execute a single tool with retry logic."""
        tool_name = tool_call["name"]
        tool_args = tool_call["input"]
 
        # Route to appropriate tool handler
        handler = self.get_tool_handler(tool_name)
        return await handler(**tool_args)

Chain-of-Thought with Tool Use

Complex problems benefit from Claude thinking through the solution before calling tools:

def solve_with_chain_of_thought(
    problem: str,
    available_tools: list[dict]
) -> str:
    """
    Leverage Claude's reasoning:
    1. Think through the problem (no tools)
    2. Plan tool usage
    3. Execute tools as planned
    4. Synthesize results
    """
 
    # Step 1: Think through the problem
    thinking_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2000,
        system="You are a problem solver. Think through this step by step before suggesting tools.",
        messages=[
            {
                "role": "user",
                "content": f"Problem: {problem}\n\nThink through how to solve this."
            }
        ]
    )
 
    thinking_process = thinking_response.content[0].text
 
    # Step 2: Use tools based on thinking
    tool_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4000,
        tools=available_tools,
        system=f"""You've already thought through this problem:
{thinking_process}
 
Now use the available tools to execute your plan.""",
        messages=[
            {
                "role": "user",
                "content": f"Problem: {problem}"
            }
        ]
    )
 
    # Step 3: Execute tools and synthesize
    tool_calls = [
        block for block in tool_response.content
        if block.type == "tool_use"
    ]
 
    tool_results = execute_tools(tool_calls)
 
    # Step 4: Get final synthesis
    final_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2000,
        messages=[
            {
                "role": "user",
                "content": problem
            },
            {
                "role": "assistant",
                "content": tool_response.content
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": result["tool_use_id"],
                        "content": json.dumps(result["content"])
                    }
                    for result in tool_results
                ]
            }
        ]
    )
 
    return final_response.content[0].text

Sophisticated Error Handling

Categorizing Errors for Appropriate Recovery

Not all errors are created equal. Production systems must distinguish between recoverable and non-recoverable failures:

import time
from enum import Enum
from typing import Callable, Any
 
class ErrorCategory(Enum):
    TRANSIENT = "transient"        # Retry immediately
    RATE_LIMITED = "rate_limited"  # Retry with backoff
    AUTH = "auth"                  # Require human intervention
    INVALID_REQUEST = "invalid"    # Don't retry
    SERVER_ERROR = "server"        # Retry with backoff
 
def categorize_error(error: Exception) -> ErrorCategory:
    """Categorize API errors for appropriate handling."""
    if isinstance(error, anthropic.RateLimitError):
        return ErrorCategory.RATE_LIMITED
    elif isinstance(error, anthropic.APIConnectionError):
        return ErrorCategory.TRANSIENT
    elif isinstance(error, anthropic.AuthenticationError):
        return ErrorCategory.AUTH
    elif isinstance(error, anthropic.BadRequestError):
        return ErrorCategory.INVALID_REQUEST
    elif isinstance(error, anthropic.InternalServerError):
        return ErrorCategory.SERVER_ERROR
    else:
        return ErrorCategory.TRANSIENT
 
class ResilientAPIClient:
    """API client with sophisticated error handling."""
 
    def __init__(self, api_key: str, max_retries: int = 5):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.max_retries = max_retries
 
    def call_with_smart_retry(
        self,
        api_call: Callable,
        *args,
        **kwargs
    ) -> Any:
        """
        Execute API call with category-aware retry logic.
        """
        retry_count = 0
        base_wait_time = 1.0  # seconds
 
        while retry_count < self.max_retries:
            try:
                return api_call(*args, **kwargs)
 
            except Exception as e:
                category = categorize_error(e)
 
                if category == ErrorCategory.AUTH:
                    # Don't retry auth errors
                    raise AuthenticationError(f"Authentication failed: {e}")
 
                elif category == ErrorCategory.INVALID_REQUEST:
                    # Don't retry bad requests
                    raise ValueError(f"Invalid request: {e}")
 
                elif category == ErrorCategory.TRANSIENT:
                    # Retry with exponential backoff
                    wait_time = base_wait_time * (2 ** retry_count)
                    time.sleep(wait_time)
                    retry_count += 1
                    continue
 
                elif category == ErrorCategory.RATE_LIMITED:
                    # Exponential backoff with jitter for rate limits
                    wait_time = base_wait_time * (2 ** retry_count)
                    jitter = wait_time * 0.1 * (2 * (hash(time.time()) % 2) - 1)
                    time.sleep(wait_time + jitter)
                    retry_count += 1
                    continue
 
                elif category == ErrorCategory.SERVER_ERROR:
                    # Longer backoff for server errors
                    wait_time = base_wait_time * (2 ** retry_count) * 2
                    time.sleep(wait_time)
                    retry_count += 1
                    continue
 
        raise Exception(f"Max retries ({self.max_retries}) exceeded")

💡

Implement exponential backoff with jitter for rate limit errors. Simple linear backoff often leads to "thundering herd" effects when multiple clients retry simultaneously.

Context Management for Error Recovery

Maintain conversation context to recover from failures gracefully:

class ConversationManager:
    """Manage conversation state for robust error recovery."""
 
    def __init__(self, user_id: str, session_id: str):
        self.user_id = user_id
        self.session_id = session_id
        self.messages = []
        self.checkpoints = []  # Save state periodically
 
    def add_message(self, role: str, content: Any):
        """Add message and save checkpoint."""
        self.messages.append({
            "role": role,
            "content": content
        })
 
    def save_checkpoint(self):
        """Save conversation state for recovery."""
        checkpoint = {
            "timestamp": time.time(),
            "message_count": len(self.messages),
            "messages": self.messages.copy(),
            "checksum": self._compute_checksum()
        }
        self.checkpoints.append(checkpoint)
        # Persist to database
        persist_checkpoint(self.user_id, self.session_id, checkpoint)
 
    def recover_from_failure(self):
        """Restore conversation to last known good state."""
        if not self.checkpoints:
            return False
 
        last_checkpoint = self.checkpoints[-1]
        self.messages = last_checkpoint["messages"].copy()
        return True
 
    def get_next_message_count(self) -> int:
        """Get number of messages for pagination/continuation."""
        return len(self.messages)
 
    def _compute_checksum(self) -> str:
        """Checksum for integrity verification."""
        content = json.dumps(self.messages, sort_keys=True)
        import hashlib
        return hashlib.sha256(content.encode()).hexdigest()

Retry Strategies at Scale

Adaptive Retry Patterns

Simple exponential backoff isn't optimal for all scenarios. Adaptive retry improves success rates:

import random
from dataclasses import dataclass
 
@dataclass
class RetryMetrics:
    total_attempts: int
    success_count: int
    failure_count: int
    last_error: str
    average_latency: float
 
class AdaptiveRetryStrategy:
    """Learn from retry patterns and adapt strategy dynamically."""
 
    def __init__(self):
        self.metrics: dict[str, RetryMetrics] = {}
 
    def get_retry_delay(
        self,
        error_type: str,
        attempt_number: int,
        recent_success_rate: float
    ) -> float:
        """
        Compute optimal retry delay based on:
        - Error type
        - Attempt count
        - Recent success rate
        """
 
        base_delay = 1.0
 
        # Scale delay based on success rate
        # High success rate: more aggressive retries
        # Low success rate: more conservative retries
        success_factor = 2.0 if recent_success_rate > 0.8 else 0.5
 
        # Exponential backoff
        exponential_delay = base_delay * (2 ** attempt_number) * success_factor
 
        # Add jitter to prevent thundering herd
        jitter = exponential_delay * random.uniform(0, 0.1)
 
        # Cap at maximum (e.g., 5 minutes)
        max_delay = 300.0
        return min(exponential_delay + jitter, max_delay)
 
    def should_retry(
        self,
        error_type: str,
        attempt_number: int,
        success_rate: float
    ) -> bool:
        """Decide whether to retry based on conditions."""
 
        # Don't retry unrecoverable errors
        if error_type in ["AUTH", "INVALID_REQUEST"]:
            return False
 
        # Always retry rate limits and transient errors
        if error_type in ["RATE_LIMITED", "TRANSIENT"]:
            return attempt_number < 10
 
        # For server errors, be conservative if success rate is low
        if error_type == "SERVER_ERROR":
            return attempt_number < (5 if success_rate > 0.7 else 3)
 
        return attempt_number < 5

Batching for Efficiency

When handling multiple requests, batch them intelligently:

class BatchProcessor:
    """Process requests in batches for efficiency."""
 
    def __init__(self, batch_size: int = 10, batch_timeout: float = 5.0):
        self.batch_size = batch_size
        self.batch_timeout = batch_timeout
        self.batch = []
        self.batch_start_time = None
 
    async def add_request(self, request: dict) -> str:
        """Add request to batch. Flush when full or timeout."""
        self.batch.append(request)
 
        if self.batch_start_time is None:
            self.batch_start_time = time.time()
 
        # Flush if batch is full
        if len(self.batch) >= self.batch_size:
            return await self.flush_batch()
 
        # Flush if timeout exceeded
        elapsed = time.time() - self.batch_start_time
        if elapsed > self.batch_timeout and self.batch:
            return await self.flush_batch()
 
        return "queued"
 
    async def flush_batch(self) -> str:
        """Process all requests in current batch."""
        if not self.batch:
            return "empty"
 
        batch_to_process = self.batch.copy()
        self.batch = []
        self.batch_start_time = None
 
        # Process batch with single API call if possible
        results = await self._process_batch(batch_to_process)
        return f"processed_{len(results)}"
 
    async def _process_batch(self, requests: list) -> list:
        """Process batch of requests efficiently."""
        # Combine into a single prompt if appropriate
        combined_prompt = self._combine_requests(requests)
 
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4000,
            messages=[
                {
                    "role": "user",
                    "content": combined_prompt
                }
            ]
        )
 
        return self._parse_batch_response(response)

Resource Optimization

Token Budgeting

Manage token consumption to prevent unexpected costs:

class TokenBudget:
    """Track and enforce token budgets per user/session."""
 
    def __init__(self, monthly_budget: int = 1_000_000):
        self.monthly_budget = monthly_budget
        self.usage_this_month: dict[str, int] = {}
 
    def check_budget(self, user_id: str, estimated_tokens: int) -> bool:
        """Check if user has budget for this request."""
        current_usage = self.usage_this_month.get(user_id, 0)
 
        if current_usage + estimated_tokens > self.monthly_budget:
            return False
        return True
 
    def record_usage(self, user_id: str, tokens_used: int):
        """Record token usage for budgeting."""
        self.usage_this_month[user_id] = (
            self.usage_this_month.get(user_id, 0) + tokens_used
        )
 
    def estimate_tokens(self, messages: list[dict]) -> int:
        """Rough estimate of tokens (1 token ≈ 4 chars)."""
        total_chars = sum(
            len(msg.get("content", ""))
            for msg in messages
        )
        return total_chars // 4

Caching Frequently Repeated Calls

Use prompt caching to optimize repeated requests:

from functools import lru_cache
import hashlib
 
class CachedAPIClient:
    """Cache API responses to reduce token usage and latency."""
 
    def __init__(self):
        self.cache = {}
        self.cache_stats = {
            "hits": 0,
            "misses": 0,
            "tokens_saved": 0
        }
 
    def _compute_cache_key(self, messages: list, model: str) -> str:
        """Compute cache key from messages."""
        content = json.dumps(messages, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
 
    def get_cached_response(
        self,
        messages: list[dict],
        model: str = "claude-opus-4-6"
    ) -> dict | None:
        """Get cached response if available."""
        cache_key = self._compute_cache_key(messages, model)
 
        if cache_key in self.cache:
            self.cache_stats["hits"] += 1
            return self.cache[cache_key]
 
        self.cache_stats["misses"] += 1
        return None
 
    def cache_response(
        self,
        messages: list[dict],
        response: dict,
        model: str = "claude-opus-4-6"
    ):
        """Cache API response."""
        cache_key = self._compute_cache_key(messages, model)
        self.cache[cache_key] = response
 
    def call_with_cache(
        self,
        messages: list[dict],
        **kwargs
    ) -> dict:
        """Call API with caching."""
        # Check cache first
        cached = self.get_cached_response(messages)
        if cached:
            return cached
 
        # Call API
        response = client.messages.create(
            messages=messages,
            **kwargs
        )
 
        # Cache and return
        self.cache_response(messages, response, kwargs.get("model", "claude-opus-4-6"))
        return response

Monitoring and Observability

Comprehensive Logging for Debugging

Structure logs to surface issues quickly:

import logging
from dataclasses import dataclass
from datetime import datetime
 
@dataclass
class APICallLog:
    timestamp: datetime
    request_id: str
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    error: str | None
    retry_count: int
    cache_hit: bool
 
class APILogger:
    """Structured logging for API calls."""
 
    def __init__(self, name: str = "claude_api"):
        self.logger = logging.getLogger(name)
 
    def log_api_call(self, log: APICallLog):
        """Log API call with structured data."""
        self.logger.info(
            "API call completed",
            extra={
                "request_id": log.request_id,
                "model": log.model,
                "input_tokens": log.input_tokens,
                "output_tokens": log.output_tokens,
                "latency_ms": log.latency_ms,
                "error": log.error,
                "retry_count": log.retry_count,
                "cache_hit": log.cache_hit,
                "total_tokens": log.input_tokens + log.output_tokens
            }
        )
 
    def log_error_pattern(self, error_type: str, count: int, rate: float):
        """Alert on error patterns."""
        if rate > 0.1:  # More than 10% error rate
            self.logger.warning(
                f"High error rate detected: {error_type}",
                extra={
                    "error_type": error_type,
                    "error_count": count,
                    "error_rate": rate
                }
            )

What the official docs don't tell you — lessons from production

Here are operational details I've picked up running Claude API across the backends of six iOS/Android apps — things that aren't spelled out in the documentation but matter once you cross into millions of calls. They're unglamorous, but they're the parts that actually move the numbers.

Design `overloaded_error` for "it always recovers if you wait"

The docs describe 529 (overloaded_error) as simply "retryable," but in practice it isn't the failure that hurts you — it's how the system behaves afterward. My wallpaper app's description-generation batch (about 4,000 items/day) saw 529s spike to 3–5% during certain peak windows. Fixed-interval retries make every request re-enter on the same second, creating a secondary overload of your own making.

After switching to exponential backoff with jitter, my final failure rate from 529s dropped from 2.1% to 0.18%. The code is the same one from the "Retry Strategies at Scale" section above — what mattered wasn't raising base_delay but widening the jitter to ±50%.

import random
 
def backoff_with_jitter(attempt: int, base: float = 1.0, cap: float = 30.0) -> float:
    # Not the docs' fixed backoff — always add ±50% jitter
    raw = min(cap, base * (2 ** attempt))
    return raw * random.uniform(0.5, 1.5)

Parallel tool calls are bounded by total token budget, not concurrency

It's tempting to assume more parallel tool calls means more speed, but the real bottleneck was total response token volume, not the number of concurrent calls. Going from 4 to 8 parallel calls barely improved throughput when each tool result was large — and timeout rates went up.

The rule of thumb I use now is simple:

Classify expected tool-result size as small (<500 tokens) / medium / large (>2,000 tokens)
Raise concurrency only for small-result tools (max 6)
Cap large-result tools (full-text DB search, file reads) at concurrency 2
Keep total input per request under roughly 30,000 tokens

After this split, average latency improved from 8.4s to 5.1s and timeout rate from 1.9% to 0.4%.

Monitor cost by median-per-call, not total

Watching total cost alone hides the occasional giant prompt that sneaks in. Once I started monitoring the median and p95 tokens per call side by side, I caught a bug where one tool was sending 6× its intended context. Running AdMob since 2014 taught me that anomalies show up in the tail of the distribution, not the average — I bring the same lens to API cost.

My recommendation: build conversation checkpoints first

If you're building one of these systems from scratch, I strongly suggest adding conversation-state checkpointing before any optimization. Multi-step tool calls that crash midway waste every API call you spent getting there. Persisting the messages array after each step lets you resume from the last success on failure, which cut my measured re-run cost by about 60%. Make retries cheap before you make anything fast — that's my priority order in production.

Production Checklist

Before deploying streaming + tool use systems to production:

[ ] Implement comprehensive error categorization
[ ] Configure appropriate retry strategies with exponential backoff
[ ] Set up token budgets and monitoring
[ ] Implement conversation checkpointing for failure recovery
[ ] Configure request timeouts (API calls should timeout in <30s)
[ ] Set up structured logging and alerting
[ ] Test graceful degradation under high load
[ ] Implement rate limit handling with jitter
[ ] Configure circuit breakers for cascading failures
[ ] Monitor cache hit rates and adjust strategy
[ ] Set up alerts for error rate anomalies
[ ] Document runbooks for common failure scenarios

Conclusion

Production-grade Claude API integration requires far more than calling the API and hoping for the best. The patterns in this guide—intelligent error handling, sophisticated retry strategies, resource optimization, and comprehensive monitoring—are what separates prototypes from bulletproof systems.

Start with the error handling patterns, layer in retry strategies based on your error categories, and progressively add resource optimization as your system scales. Monitor closely and adapt your strategies based on real-world patterns in your specific use case.

Thank You for Reading

Claude Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.

✦Copy-paste ready implementation code
✦New advanced guides published daily
✦$5/mo or $10 for lifetime access

View Membership →

⬡ API & SDK2026-04-20

Three Hidden Pitfalls When Implementing Claude API Streaming

Real-world lessons from building with Claude API streaming: runtime environment mismatches, error handling gaps, and silent token cost overruns — with working TypeScript examples.

⬡ API & SDK2026-06-20

Running Subagents in Parallel Without One Failure Sinking the Whole Run

A fan-out / fan-in design for running several subagents in parallel, covering token budgeting, a result contract, and partial-failure handling. Includes an implementation where one branch can fail without stopping the rest, plus measured numbers.

⬡ API & SDK2026-06-13

Claude Vision API in Production — Implementation Patterns for Image Analysis, PDF Processing, and OCR

Implementation patterns for taking Claude's vision capabilities to production: choosing between Base64, URL, and the Files API, native PDF processing, schema-enforced extraction with Tool Use, batch cost reduction, and error recovery — all with working code.

📚RECOMMENDED BOOKS

Build a Large Language Model (From Scratch)

Sebastian Raschka

LLM Dev

Prompt Engineering for LLMs

* Contains affiliate links

See all →

Claude API Streaming & Tool Use in Production — Patterns for Parallel Calls, Error Handling, and Retry Strategies

What Separates a Prototype from Production

Streaming Fundamentals for Production

Understanding Stream Lifecycle

Buffering Strategies for Real-time Systems

Continue Reading

Unlock This Article

Parallel Tool Calling Patterns

Coordinating Multiple Tool Calls

Chain-of-Thought with Tool Use

Sophisticated Error Handling

Categorizing Errors for Appropriate Recovery

Context Management for Error Recovery

Retry Strategies at Scale

Adaptive Retry Patterns

Batching for Efficiency

Resource Optimization

Token Budgeting

Caching Frequently Repeated Calls

Monitoring and Observability

Comprehensive Logging for Debugging

What the official docs don't tell you — lessons from production

Design overloaded_error for "it always recovers if you wait"

Parallel tool calls are bounded by total token budget, not concurrency

Monitor cost by median-per-call, not total

My recommendation: build conversation checkpoints first

Production Checklist

Conclusion

Thank You for Reading

Related Articles

Design `overloaded_error` for "it always recovers if you wait"