Most "AI agents" shipping right now are demos wearing production paint. They answer questions fluently and break the moment they touch a workflow with money, state, or consequences.

The agent illusion

Multi-agent frameworks benchmark beautifully. Five specialist LLMs cooperate, the demo plays cleanly, the README has a diagram with arrows. Then someone wires the thing to a real billing system and it issues three refunds for the same chargeback because a tool call retried on a flaky network.

That gap is the actual job most agent tutorials skip.

When I review AI code, the same pattern keeps appearing. The LLM call is fine. The agent loop is fine. What is missing is the layer underneath: state, idempotency, audit, and a tool surface the agent cannot use to hurt you.

The unsexy layer

Systems answer four questions before the agent does anything:

  1. What did the agent already do? Persisted state, not "look at the conversation history."
  2. What happens if this action runs twice? Idempotency keys on every external effect.
  3. Who approved this? An audit log a human can read during a postmortem.
  4. Can I roll this back? A clear inverse for every irreversible operation, or a freeze before execution.

None of this is glamorous. It is also what separates a system that is a toy demo from one that can run mostly unsupervised in production.

The shape of that contract in code:

class ExpenseAction(BaseModel):
    idempotency_key: str
    requested_by: str
    requested_at: datetime
    approval_required: bool = True
    dry_run: bool = True
    payload: ExpensePayload

def submit(action: ExpenseAction, repo: ExpenseRepo) -> Result:
    if repo.find_by_key(action.idempotency_key):
        return Result.duplicate()
    if action.dry_run:
        return Result.preview(action.plan())
    if action.approval_required and not action.is_approved():
        return Result.pending_approval()
    repo.persist(action)
    return Result.ok(action.execute())

The agent does not call the side effect. It builds a typed plan. A function decides whether to run it.

State that survives retries

Agents need state management that works across restarts and network failures. The Telegram expense bot we build in our Agentic AI cohort program, uses context.user_data to track multi-step flows:

async def handle_expense_text(self, update, context):
    text = update.message.text
    result = self._preprocessor.preprocess(text)

    if not result.is_valid:
        await update.message.reply_text(f"Invalid: {result.error}")
        return ConversationHandler.END

    response = self._build_service().classify(result.text).response

    # Store state for the callback handler
    context.user_data["expense_description"] = result.text
    context.user_data["classification_response"] = response

    keyboard = build_category_confirmation_keyboard(
        suggested_category=response.category,
        all_categories=[c.value for c in ExpenseCategory],
    )

    await update.message.reply_text(
        f"I categorized this as {response.category} ({response.total_amount} {response.currency}). Confirm or pick another category:",
        reply_markup=keyboard,
    )

    return ConversationState.WAITING_FOR_CATEGORY

async def handle_category_selection(self, update, context):
    query = update.callback_query
    await query.answer()

    # Retrieve state from previous handler
    description = context.user_data.get("expense_description")
    response = context.user_data.get("classification_response")

    if description is None or response is None:
        await query.edit_message_text("Session expired. Send expense again.")
        return ConversationHandler.END

    _, category = query.data.split(":", 1)

    self._build_service().persist_with_category(
        expense_description=description,
        category_name=category,
        response=response,
        telegram_user_id=update.effective_user.id,
    )

    await query.edit_message_text(f"Saved as {category}!")
    return ConversationHandler.END

The .get() with defensive error handling is what saves you when the bot restarts mid-conversation. No silent corruption, no half-written database rows. The user just has to resend their expense description and pick the category again. This is the work of production agents.

Tools the agent cannot trust

LLMs are undeterministic and hallucinate. Design your tool surface for mistrust:

  • Narrow scopes. read_expense and flag_expense are two tools, not one tool with a mode flag the LLM can flip.
  • Dry-run by default. Every write tool returns a plan first. The agent opts in to execute. You get human-in-the-loop (HITL) for free.
  • Schema-validated inputs. Pydantic at the tool boundary so a malformed argument cannot reach your database.
  • Explicit confirmation for anything destructive. The agent proposes, a human taps approve.

The agent is not the brain of your application. It is a planner that we acknowledge is fallible. The real logic lives in the tools, and the agent's job is to call them with valid inputs and ask for help when it is unsure.

Input validation before the LLM sees anything

Validate at system boundaries before user input reaches your tools. This prevents XSS, length attacks, and malformed data from consuming tokens:

from dataclasses import dataclass, field
import re

XSS_PATTERNS = ("<script", "javascript:", "onerror=", "onload=")
CURRENCY_SYMBOLS = {"$": "USD", "€": "EUR", "£": "GBP", "¥": "JPY"}
AMOUNT_PATTERN = re.compile(r"\d+([.,]\d+)?")

@dataclass
class PreprocessingResult:
    text: str
    is_valid: bool
    warnings: list[str] = field(default_factory=list)
    error: str | None = None

class InputPreprocessor:
    def preprocess(self, text: str) -> PreprocessingResult:
        text = text.strip()

        if len(text) < 3:
            return PreprocessingResult(text, False, error="Input too short")
        if len(text) > 500:
            return PreprocessingResult(text, False, error="Input too long")

        if any(pattern in text.lower() for pattern in XSS_PATTERNS):
            return PreprocessingResult(text, False, error="Invalid characters")

        for symbol, code in CURRENCY_SYMBOLS.items():
            text = text.replace(symbol, code)

        warnings = []
        if not AMOUNT_PATTERN.search(text):
            warnings.append("No amount detected")

        return PreprocessingResult(text, True, warnings)

This runs before the LLM call, returning error messages without burning tokens or risking injection.

Human-in-the-loop as a design pattern

Production agents are not fully autonomous. They classify, extract, or suggest, then wait for a human to confirm. Confidence scores guide when to ask:

from dataclasses import dataclass

@dataclass(frozen=True)
class ClassificationResult:
    response: ExpenseCategorizationResponse
    persisted: bool

def process_with_hitl(result: ClassificationResult, threshold: float = 0.8) -> str:
    if result.response.confidence >= threshold:
        return result.response.category

    print(
        f"Low confidence ({result.response.confidence:.0%}): '{result.response.category}' — {result.response.reason}"
    )
    user_input = input(
        f"Accept '{result.response.category}'? (Enter to confirm, or type a category): "
    ).strip()

    if not user_input:
        return result.response.category
    return user_input

In the Telegram bot, this becomes an inline keyboard. The bot states its category guess and asks the human to confirm or pick a different one, with the AI suggestion highlighted.

Telegram expense bot confidence-based keyboards

The pattern: AI proposes, human disposes. This surfaces in the service layer we built in prior weeks:

@dataclass
class ClassificationService:
    assistant: Assistant
    expense_repo: ExpenseRepository

    def classify(self, description: str) -> ClassificationResult:
        messages = self._build_messages(description)
        response = self.assistant.completion(messages)
        return ClassificationResult(response=response, persisted=False)

    def persist_with_category(
        self,
        expense_description: str,
        category_name: str,
        response: ExpenseCategorizationResponse,
        telegram_user_id: int | None = None,
    ) -> None:
        """Store the user's chosen category, not the AI guess."""
        expense = Expense(
            amount=response.total_amount,
            currency=response.currency,
            category=ExpenseCategory(category_name),
            description=expense_description,
            telegram_user_id=telegram_user_id,
        )
        self.expense_repo.add(expense)

The persist_with_category method accepts the human's decision. The database stores what the user confirmed, not what the model guessed. As the ExpenseCategorizationResponse captures the AI's original category and confidence, we can analyze overrides later to identify model weaknesses.

Dependency injection for testable agents

The service layer pattern separates business logic from LLM provider details. Inject dependencies rather than hardcoding them:

from unittest.mock import create_autospec
from decimal import Decimal

def test_classify_calls_assistant():
    # No real OpenAI call, no .env file, no network
    mock_assistant = create_autospec(Assistant)
    mock_assistant.completion.return_value = ExpenseCategorizationResponse(
        category="Food",
        total_amount=Decimal("5.50"),
        currency=Currency.USD,
        confidence=0.95,
        cost=Decimal("0.001"),
    )

    mock_repo = create_autospec(ExpenseRepository)
    service = ClassificationService(assistant=mock_assistant, expense_repo=mock_repo)
    result = service.classify("Coffee at Starbucks $5.50")

    mock_assistant.completion.assert_called_once()
    assert result.response.category == "Food"
    assert result.persisted is False

Because the service receives its dependencies rather than creating them, you can test classification logic without burning API credits or waiting on network calls. This is a key strategy to test the interface at the service layer, not the LLM provider.

The same service powers the CLI, Telegram bot, and REST API. Change providers (OpenAI to Anthropic) or add caching by swapping the Assistant implementation. Business logic stays untouched.

Speed vs safety

The tradeoff is iteration speed vs execution safety.

Put the LLM behind a typed service boundary and you can swap models without touching business logic. Store actions as events instead of overwriting state, and your audit log writes itself. I wrote about why event sourcing pays off.

Agentic loops with typed tool results

The tool-use loop needs to handle partial results, retries, and tool failures. Here is the pattern from the warm up exercises you can do on our Agentic Cohort page:

from typing import cast
import anthropic
from anthropic.types import (
    MessageParam,
    TextBlock,
    ToolUseBlock,
    ToolResultBlockParam,
)

# TOOLS defined with JSON schema for get_exchange_rate(from_currency, to_currency)

def answer_with_tools(question: str, client: anthropic.Anthropic) -> str:
    messages: list[MessageParam] = [{"role": "user", "content": question}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=512,
            tools=TOOLS,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return cast(TextBlock, response.content[0]).text

        if response.stop_reason != "tool_use":
            # anything other than tool_use here means no tool calls to process — looping would spin forever
            raise RuntimeError(f"Unexpected stop reason: {response.stop_reason}")

        tool_uses = [
            cast(ToolUseBlock, b) for b in response.content if b.type == "tool_use"
        ]
        tool_results: list[ToolResultBlockParam] = [
            {
                "type": "tool_result",
                "tool_use_id": b.id,
                "content": str(get_exchange_rate(**cast(dict[str, str], b.input))),
            }
            for b in tool_uses
        ]

        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

The loop continues until stop_reason == "end_turn". Tool results are typed, preventing schema drift between the tool definition and implementation.

In production, wrap get_exchange_rate() in a try/except and return error results to the LLM when tools fail. The agent can retry, pick a different tool, or surface the error to the user.

The fix is separation of concerns, typed interfaces, and a well-defined contract between the agent and its tools.

Keep reading