Why 88% of AI Agents Fail in Production: The 5-Layer Harness Architecture Is the Answer

# Example of MCP-based tool registration (Claude Agent SDK)
from anthropic import Anthropic
 
client = Anthropic()
 
tools = [
    {
        "name": "read_file",
        "description": "Reads and returns the contents of a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Absolute path of the file to read"
                }
            },
            "required": ["path"]
        }
    },
    {
        "name": "run_tests",
        "description": "Runs the specified test suite and returns the results",
        "input_schema": {
            "type": "object",
            "properties": {
                "test_path": {
                    "type": "string",
                    "description": "Path to the test file or directory"
                }
            },
            "required": ["test_path"]
        }
    }
]
 
response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=4096,
    tools=tools,
    messages=[{"role": "user", "content": "Read src/auth.py and run the tests"}]
)

# PEV pattern implementation example
async def pev_loop(task: str, max_retries: int = 3) -> dict:
    plan = await agent.plan(task)        # 1. Create a plan
 
    for attempt in range(max_retries):
        result = await agent.execute(plan)   # 2. Execute
 
        # Delegate verification to an independent function, not the agent itself
        verification = await verifier.check(
            task=task,
            plan=plan,
            result=result
        )                                     # 3. Verify
 
        if verification.passed:
            return {"status": "success", "result": result}
 
        # On verification failure, incorporate feedback, revise the plan, and retry
        plan = await agent.replan(
            original_task=task,
            failed_result=result,
            feedback=verification.feedback
        )
 
    return {"status": "failed", "last_result": result}

# Hierarchical memory management example
# mem0: Python library that provides long-term memory for agents (pip install mem0ai)
from mem0 import Memory
 
memory = Memory()
 
class StatefulAgent:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.session_history = []  # Short-term: maintained only within the session
 
    async def process(self, message: str) -> str:
        # Retrieve context relevant to the current message from long-term memory
        long_term = memory.search(
            query=message,
            user_id=self.user_id,
            limit=5
        )
 
        # Synthesize short-term + long-term memory to build the system prompt
        context = self._build_context(
            session=self.session_history,
            long_term=long_term
        )
 
        response = await llm.complete(system=context, user=message)
 
        # Update short-term memory
        self.session_history.append({"role": "user", "content": message})
        self.session_history.append({"role": "assistant", "content": response})
 
        # Save important information (project conventions, user preferences, etc.) to long-term memory
        memory.add(
            messages=[
                {"role": "user", "content": message},
                {"role": "assistant", "content": response}
            ],
            user_id=self.user_id
        )
 
        return response
 
    def _build_context(self, session: list, long_term: list) -> str:
        # Insert long-term memory at the beginning of the system prompt
        memory_context = "\n".join(f"- {m['memory']}" for m in long_term)
        return f"Past memories:\n{memory_context}\n\n"

# High-risk action guardrail implementation
from enum import Enum
from typing import Callable
 
class RiskLevel(Enum):
    LOW      = "low"
    MEDIUM   = "medium"
    HIGH     = "high"
    CRITICAL = "critical"
 
# It's best to define risk level and policy together at the time of tool registration
ACTION_POLICIES = {
    "read_file":         RiskLevel.LOW,
    "write_file":        RiskLevel.MEDIUM,
    "delete_file":       RiskLevel.CRITICAL,
    "send_http_request": RiskLevel.HIGH,
    "run_shell_command": RiskLevel.HIGH,
}
 
async def guarded_execute(
    action: str,
    params: dict,
    human_approval_fn: Callable
) -> dict:
    risk = ACTION_POLICIES.get(action, RiskLevel.HIGH)
 
    if risk == RiskLevel.CRITICAL:
        # CRITICAL actions require explicit human approval
        approved = await human_approval_fn(
            action=action,
            params=params,
            reason=f"Risk level CRITICAL: approval required to execute {action}"
        )
        if not approved:
            return {"status": "rejected", "reason": "Not approved by user"}
 
    elif risk == RiskLevel.HIGH:
        # HIGH can be auto-approved after a policy check
        if not policy_check(action, params):  # policy check helper
            return {"status": "blocked", "reason": "Policy violation"}
 
    return await execute_action(action, params)  # actual tool execution

# Agent tracing with LangSmith
# LangSmith: observability tool for tracking agent execution flow, token usage, and failure points
# pip install langsmith
from langsmith import traceable
 
@traceable(name="agent-main-loop", tags=["production"])
async def agent_loop(task: str) -> str:
    """@traceable automatically traces the entire function execution."""
    plan = await create_plan(task)
 
    results = []
    for step in plan.steps:
        result = await execute_step(step)
        results.append(result)
 
    return synthesize_results(results)
 
# Applying it to individual steps records the tokens and time consumed per step
@traceable(name="execute-step")
async def execute_step(step) -> dict:
    result = await run_tool(step.tool, step.params)  # actual tool call
    return {"step_type": step.type, "result": result}

import asyncio
from anthropic import Anthropic
 
client = Anthropic()
 
async def planner_agent(brief: str) -> dict:
    """Short brief → complete product spec (returns JSON)"""
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=8192,
        system="""You are a product architect.
Expand the simple requirements provided into a technical spec.
Always return in JSON format:
{"features": [...], "api_contracts": {...}, "test_criteria": [...]}""",
        messages=[{"role": "user", "content": brief}]
    )
    return parse_spec(response.content[0].text)  # JSON parsing helper
 
async def generator_agent(spec: dict, sprint_idx: int) -> dict:
    """Spec → sprint-unit code implementation"""
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=8192,
        system="You are a senior engineer. Implement code according to the specification.",
        messages=[{
            "role": "user",
            "content": f"Implement Sprint {sprint_idx}:\n{spec['features'][sprint_idx]}"
        }]
    )
    return {"code": response.content[0].text, "sprint": sprint_idx}
 
async def evaluator_agent(code: dict, test_criteria: list) -> dict:
    """Run tests, analyze results, and determine pass/fail"""
    # run_tests: helper that calls actual test tools like pytest / Playwright
    test_results = await run_tests(code["code"], test_criteria)
 
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=2048,
        system="Analyze the test results and provide a pass/fail judgment with specific feedback.",
        messages=[{"role": "user", "content": f"Test results:\n{test_results}"}]
    )
    return {
        "passed": "PASS" in response.content[0].text,
        "feedback": response.content[0].text
    }
 
async def three_agent_harness(brief: str) -> str:
    spec = await planner_agent(brief)      # Planning
 
    final_code = []
    for sprint_idx in range(len(spec["features"])):
        retries = 0
        while retries < 3:
            code = await generator_agent(spec, sprint_idx)   # Implementation
            evaluation = await evaluator_agent(              # Layer 2: verification loop
                code, spec["test_criteria"]
            )
 
            if evaluation["passed"]:
                final_code.append(code)
                break
 
            # Incorporate feedback into the spec for use in the next attempt
            spec["features"][sprint_idx] += f"\nFeedback: {evaluation['feedback']}"
            retries += 1
 
    return assemble_final_code(final_code)  # Assemble per-sprint code into final deliverable

// Full 5-layer harness setup in TypeScript
import Anthropic from "@anthropic-ai/sdk";
import { Client as LangSmithClient } from "langsmith";
import { traceable } from "langsmith/traceable"; // Layer 5: Observability
import { Memory } from "mem0ai";                 // Layer 3: Context & Memory
 
const client = new Anthropic();
const langsmith = new LangSmithClient();
const memory = new Memory();
 
// Layer 4: Guardrails — list of high-risk operations to block
const BLOCKED_OPERATIONS = ["force_push", "delete_branch_main", "drop_table"];
 
// Wrapping with traceable() automatically records all executions to LangSmith
const codingAgentHarness = traceable(
  async (task: string, repoContext: string) => {
    // Layer 3: Load project long-term memory
    const projectMemory = await memory.search({ query: task, limit: 10 });
 
    // Layer 4: Pre-task risk check
    const isBlocked = BLOCKED_OPERATIONS.some(op => task.includes(op));
    if (isBlocked) {
      return { status: "blocked", reason: "Policy-violating operation detected" };
    }
 
    // Layer 1: Tool definitions (Tool Orchestration)
    const tools: Anthropic.Tool[] = [
      {
        name: "read_file",
        description: "Reads and returns the contents of a file",
        input_schema: {
          type: "object" as const,
          properties: {
            path: { type: "string", description: "Path of the file to read" }
          },
          required: ["path"]
        }
      },
      {
        name: "write_file",
        description: "Writes content to a file",
        input_schema: {
          type: "object" as const,
          properties: {
            path:    { type: "string", description: "Path of the file to write" },
            content: { type: "string", description: "Content to write to the file" }
          },
          required: ["path", "content"]
        }
      },
      {
        name: "run_tests",
        description: "Runs the test suite and returns the results",
        input_schema: {
          type: "object" as const,
          properties: {
            test_path: { type: "string", description: "Path to the test directory" }
          },
          required: ["test_path"]
        }
      }
    ];
 
    let passed = false;
    let attempts = 0;
 
    // Layer 2: PEV verification loop
    while (!passed && attempts < 3) {
      await client.messages.create({
        model: "claude-opus-4-7",
        max_tokens: 8192,
        tools,
        system: buildSystemPrompt(projectMemory, repoContext), // inject memory
        messages: [{ role: "user", content: task }]
      });
 
      // Verify with an independent function, not the agent itself
      const testResult = await runTestSuite();
      passed = testResult.passed;
      attempts++;
    }
 
    // Layer 3: Save task result to long-term memory
    await memory.add({ task, result: "completed", attempts });
 
    return { status: passed ? "success" : "failed", attempts };
  },
  { name: "coding-agent", tags: ["production"] }
);