7 Design Principles for Production-Grade Autonomous Agents with Claude Code SDK

There's a significant gap between "getting an agent to work" and "keeping an agent running reliably in production."

Building your first Claude Code SDK agent is surprisingly straightforward. But running it on a daily schedule will reveal failure modes you didn't anticipate. I've run content automation systems in production long enough to have encountered most of them. This article documents the 7 design principles I extracted from that experience — each with implementation code.

Principle 1: Always Set a Loop Depth Limit

Claude Code SDK's query() handles tool-use loops automatically. Convenient — but this also means loops that never terminate are easy to create.

The most common infinite loop pattern: the agent repeatedly calls the same tool. It keeps trying the same fix, never escaping.

import anthropic
 
def run_agent_with_depth_limit(
    prompt: str,
    max_turns: int = 20,
    model: str = "claude-sonnet-4-6"
) -> str:
    client = anthropic.Anthropic()
    messages = [{"role": "user", "content": prompt}]
    turn_count = 0
    
    while turn_count < max_turns:
        response = client.messages.create(
            model=model,
            max_tokens=4096,
            tools=TOOLS,
            messages=messages
        )
        turn_count += 1
        
        if response.stop_reason == "end_turn":
            return extract_text(response)
        
        tool_results = execute_tools(response.content)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})
    
    raise AgentLoopLimitError(f"Agent exceeded {max_turns} turns")

Appropriate max_turns depends on task complexity. I use 10–15 for article generation and 20–30 for code review. When the limit is hit, raise an exception, log it, and skip to the next task.

Principle 2: Two Layers of Timeouts

You need both a per-call timeout and an overall loop timeout. The per-call timeout controls individual API requests; the overall timeout controls the entire agent execution.

import signal
from contextlib import contextmanager
 
@contextmanager
def timeout_context(seconds: int):
    def handler(signum, frame):
        raise TimeoutError(f"Agent timed out after {seconds} seconds")
    
    old_handler = signal.signal(signal.SIGALRM, handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)
 
async def run_agent_with_timeout(prompt: str, timeout_seconds: int = 300) -> str:
    try:
        with timeout_context(timeout_seconds):
            return await run_agent_core(prompt)
    except TimeoutError as e:
        log_timeout(prompt, timeout_seconds)
        raise AgentTimeoutError(str(e))

I set 5–10 minutes per task for scheduled runs. Any task exceeding that has a problem — stuck loop, unresponsive API — and gets force-terminated.

Principle 3: Detect Duplicate Tool Calls

90% of infinite loops follow the pattern of calling the same tool with the same arguments repeatedly. A simple guard handles most cases:

from collections import deque
from hashlib import md5
import json
 
class ToolCallDeduplicator:
    def __init__(self, window_size: int = 5):
        self.recent_calls = deque(maxlen=window_size)
    
    def is_duplicate(self, tool_name: str, tool_input: dict) -> bool:
        call_hash = md5(
            json.dumps({"name": tool_name, "input": tool_input}, sort_keys=True).encode()
        ).hexdigest()
        
        if call_hash in self.recent_calls:
            return True
        self.recent_calls.append(call_hash)
        return False
 
def execute_tools_with_dedup(tool_calls: list, dedup: ToolCallDeduplicator) -> list:
    results = []
    for call in tool_calls:
        if dedup.is_duplicate(call.name, call.input):
            results.append({
                "type": "tool_result",
                "tool_use_id": call.id,
                "content": "Error: This exact tool call was already attempted. Please try a different approach.",
                "is_error": True
            })
        else:
            results.append(execute_single_tool(call))
    return results

After adding this guard, task failures from infinite loops dropped by approximately 85%.

Principle 4: Manage Cost Budget at the Agent Level

You may have overall API cost controls, but managing the budget per agent execution is a separate concern — and a necessary one for production.

from dataclasses import dataclass
 
@dataclass
class CostBudget:
    max_dollars: float
    spent_dollars: float = 0.0
    
    INPUT_RATE = 3.0 / 1_000_000    # Sonnet 4.6 $/token
    OUTPUT_RATE = 15.0 / 1_000_000
    
    def record_usage(self, usage) -> None:
        self.spent_dollars += (
            usage.input_tokens * self.INPUT_RATE +
            usage.output_tokens * self.OUTPUT_RATE
        )
    
    @property
    def is_exceeded(self) -> bool:
        return self.spent_dollars >= self.max_dollars
 
def run_agent_with_budget(prompt: str, budget: CostBudget) -> str:
    client = anthropic.Anthropic()
    messages = [{"role": "user", "content": prompt}]
    
    while True:
        if budget.is_exceeded:
            raise BudgetExceededError(
                f"Budget exceeded: ${budget.spent_dollars:.4f} / ${budget.max_dollars}"
            )
        
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            messages=messages
        )
        budget.record_usage(response.usage)
        
        if response.stop_reason == "end_turn":
            return extract_text(response)
        # continue loop...

I set budgets of $0.05–$0.20 per task. Budget overruns get logged for monthly review — useful for identifying tasks that cost more than expected.

Principle 5: Classify Errors as Recoverable or Fatal

A common mistake is treating all errors the same way. Rate limit errors (429) and bad request errors (400) require completely different responses.

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
 
class RecoverableError(Exception):
    pass
 
class FatalError(Exception):
    pass
 
def classify_api_error(e: anthropic.APIError) -> None:
    if isinstance(e, anthropic.RateLimitError):
        raise RecoverableError(f"Rate limit: {e}")
    elif isinstance(e, anthropic.InternalServerError):
        raise RecoverableError(f"Server error: {e}")
    elif isinstance(e, anthropic.BadRequestError):
        raise FatalError(f"Bad request — likely a prompt issue: {e}")
    elif isinstance(e, anthropic.AuthenticationError):
        raise FatalError(f"Auth failed — check API key: {e}")
    else:
        raise RecoverableError(f"Unknown error: {e}")
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    retry=retry_if_exception_type(RecoverableError)
)
def call_api_with_retry(params: dict):
    client = anthropic.Anthropic()
    try:
        return client.messages.create(**params)
    except anthropic.APIError as e:
        classify_api_error(e)

FatalError propagates immediately to skip the task. RecoverableError retries with exponential backoff. This classification alone significantly reduces wasted API calls from unnecessary retries.

Principle 6: Structure Logs for Debuggability

Scheduled agent failures are hard to diagnose without structured logs. Make logs filterable from day one.

import json
import logging
from datetime import datetime, timezone
 
def create_structured_logger(task_id: str) -> logging.Logger:
    logger = logging.getLogger(task_id)
    
    class StructuredFormatter(logging.Formatter):
        def format(self, record: logging.LogRecord) -> str:
            log_data = {
                "ts": datetime.now(timezone.utc).isoformat(),
                "task_id": task_id,
                "level": record.levelname,
                "msg": record.getMessage(),
            }
            if hasattr(record, "extra"):
                log_data.update(record.extra)
            return json.dumps(log_data, ensure_ascii=False)
    
    handler = logging.StreamHandler()
    handler.setFormatter(StructuredFormatter())
    logger.addHandler(handler)
    return logger

JSON structured logs make it trivial to query with jq or CloudWatch Logs Insights: "list all budget-exceeded tasks from last week" or "calculate error rate for the search tool."

Principle 7: Design for Graceful Degradation

The most important principle. When an agent fails, the goal shouldn't be "return nothing" — it should be "return something, even if lower quality."

from enum import Enum
 
class DegradationLevel(Enum):
    FULL = "full"       # Full features (extended thinking + rich prompts)
    REDUCED = "reduced" # Reduced (no thinking + simplified prompts)
    MINIMAL = "minimal" # Minimal (Haiku + shortest prompts)
    SKIP = "skip"       # Skip (defer to next run)
 
def run_with_graceful_degradation(task: dict, max_retries: int = 3) -> dict:
    level = DegradationLevel.FULL
    
    for attempt in range(max_retries):
        try:
            if level == DegradationLevel.FULL:
                return run_full_agent(task)
            elif level == DegradationLevel.REDUCED:
                return run_reduced_agent(task)
            elif level == DegradationLevel.MINIMAL:
                return run_minimal_agent(task)
            else:
                return {"status": "skipped", "task_id": task["id"]}
                
        except (AgentLoopLimitError, BudgetExceededError):
            if level == DegradationLevel.FULL:
                level = DegradationLevel.REDUCED
            elif level == DegradationLevel.REDUCED:
                level = DegradationLevel.MINIMAL
            else:
                level = DegradationLevel.SKIP
                
        except FatalError:
            return {"status": "fatal_error", "task_id": task["id"]}
    
    return {"status": "skipped", "task_id": task["id"]}

This changes scheduled execution success from binary (0% or 100%) to a quality gradient. Full-mode failures still produce minimal-quality output rather than nothing.

Production Deployment Checklist

Before deploying a production agent, confirm all seven: loop depth limit is set; overall timeout is set; duplicate tool call detection is implemented; cost budget is managed at the agent level; errors are classified as recoverable vs. fatal; logs are structured JSON; graceful degradation is designed.

Takeaway: Production Quality Is About Failure Design

An agent's production quality isn't determined only by how well it handles the happy path. It's determined by how it fails — and how those failures are contained.

The seven principles here can be implemented independently. Start with the two that prevent the most common failures: loop depth limits and timeouts. Those two alone will meaningfully improve the reliability of your scheduled runs.

Running autonomous agents in production isn't a one-time build. It requires understanding failure modes, encoding that understanding into the design, and continuing to observe. That's what production quality means.