DEV Community: Pamui Afrika

LOOP ENGINEERING: TECHNICAL BLUEPRINT

Pamui Afrika — Thu, 25 Jun 2026 14:52:17 +0000

Complete Architecture, Design Patterns & Implementation Guide for AI Systems

Executive Overview
Core Loop Architecture
Loop Components & Subsystems
State Management
Control Flow & Termination Logic
Error Handling & Recovery
Context Management
Verification & Validation Systems
Cost Optimization & Token Management
Advanced Patterns
Implementation Examples
Monitoring & Observability

EXECUTIVE OVERVIEW

Definition

Loop engineering is the practice of designing self-prompting autonomous systems where humans define objectives, constraints and verification rules—while the system autonomously iterates through cycles of task generation, execution, observation and refinement until goals are achieved.

Paradigm Shift

PROMPT ENGINEERING (Era 1)
User → [Write Perfect Prompt] → LLM → Answer → Copy/Paste

LOOP ENGINEERING (Era 2)
User → [Design Loop System] → [Autonomous Iterations]
                              ├─ Generate Task
                              ├─ Execute
                              ├─ Verify Result
                              ├─ Feedback
                              └─ Retry/Next (until goal met)

Why Loops > Prompts for Complex Tasks

Dimension	Prompt Engineering	Loop Engineering
Task Length	Single pass	Iterative multi-step
Failure Recovery	Manual retry	Automatic with backoff
State Management	User memory	External persistence
Verification	Manual check	Automated validation
Tool Usage	Static composition	Dynamic, context-aware
Context Window	Single instance	Rolling, persistent
Cost Control	Unpredictable	Bounded with rules

CORE LOOP ARCHITECTURE

1. The Canonical Loop Cycle

Every loop execution follows this deterministic sequence:

┌─────────────────────────────────────────────────────────┐
│                    LOOP INITIALIZATION                  │
│  • Load state from storage                              │
│  • Validate preconditions                               │
│  • Inject context into LLM prompt                       │
└──────────────────┬──────────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────────┐
│              PHASE 1: PLANNING (LLM Call)               │
│                                                         │
│  Input:  [Goal] + [Current State] + [Context]           │
│  LLM:    Generate next action(s)                        │
│  Output: {action, params, reasoning, confidence}        │
│                                                         │
│  Cost: ~0.5K-5K tokens depending on context size        │
└──────────────────┬──────────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────────┐
│           PHASE 2: EXECUTION (Tool/Code Run)            │
│                                                         │
│  • Parse LLM output (JSON schema validation)            │
│  • Execute tool with timeout protection                 │
│  • Capture stdout/stderr/errors                         │
│  • Record wall-clock time & resource usage              │
│                                                         │
│  Cost: $0 (deterministic execution)                     │
└──────────────────┬──────────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────────┐
│        PHASE 3: OBSERVATION (State Update)              │
│                                                         │
│  • Capture execution output                             │
│  • Update external state stores                         │
│  • Extract metrics (success/failure, delta)             │
│  • Persist iteration record                             │
│                                                         │
│  Cost: $0 (I/O operations)                              │
└──────────────────┬──────────────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────────┐
│       PHASE 4: VERIFICATION (Goal Check)                │
│                                                         │
│  Evaluate: Goal Met? → EXIT LOOP                        │
│            Recoverable? → RETRY Phase 1                 │
│            Dead End? → ERROR → Phase 5                  │
│            New State? → LOOP again                      │
│                                                         │
│  Cost: $0 (deterministic logic)                         │
└──────────────────┬──────────────────────────────────────┘
                   │
     ┌─────────────┴──────────┬──────────────┐
     │                        │              │
     ▼                        ▼              ▼
[GOAL MET]          [CONTINUE LOOP]    [ERROR/TIMEOUT]
   EXIT                  Loop back          Phase 5
                         to Planning

2. Loop State Machine

[INIT]
  │
  ├─→ [PLANNING] ──(LLM Call)──→ [PARSING] ──(Validate)──→ [EXECUTION]
  │                                                            │
  │                                                            ▼
  │                        ┌──────────────────────────────→ [OBSERVATION]
  │                        │                                    │
  │                        │                                    ▼
  │                        │              ┌─────────────────[VERIFICATION]
  │                        │              │                    │
  └──────────────────(Loop Control)───────┤                    ├──→ [SUCCESS/EXIT]
                                           │                    │
                                           └──(Retry Logic)─────┤
                                                            │
                                           ┌────────────────┤
                                           │                │
                                      [RECOVERY]      [TIMEOUT/FATAL]
                                           │                │
                                           └────→ [CLEANUP]←─┘
                                                    │
                                                    ▼
                                               [FINAL_STATE]

3. Minimal Loop Implementation (Pseudocode)

class AutonomousLoop:
    def __init__(self, goal: str, llm_client, tools: Dict[str, Callable]):
        self.goal = goal
        self.llm = llm_client
        self.tools = tools
        self.state = {"iterations": 0, "history": []}
        self.max_iterations = 20
        self.timeout = 300  # seconds

    def run(self) -> LoopResult:
        start_time = time.time()

        while self.state["iterations"] < self.max_iterations:
            # PHASE 1: PLANNING
            prompt = self._build_prompt()
            response = self.llm.complete(prompt)
            action = self._parse_action(response)

            # PHASE 2: EXECUTION
            try:
                result = self._execute_action(action, timeout=30)
            except Exception as e:
                result = {"error": str(e), "type": type(e).__name__}

            # PHASE 3: OBSERVATION
            self.state["history"].append({
                "iteration": self.state["iterations"],
                "action": action,
                "result": result,
                "timestamp": time.time()
            })

            # PHASE 4: VERIFICATION
            verification = self._verify_goal_met(result)
            if verification["goal_met"]:
                return LoopResult(success=True, state=self.state)

            # Check for unrecoverable conditions
            if self._is_fatal_error(result) or (time.time() - start_time) > self.timeout:
                return LoopResult(success=False, state=self.state, reason="fatal_error")

            self.state["iterations"] += 1

        return LoopResult(success=False, state=self.state, reason="max_iterations")

    def _build_prompt(self) -> str:
        context = self._serialize_state()
        return f"""
Goal: {self.goal}

Current State:
{context}

Available Tools: {list(self.tools.keys())}

Respond ONLY in JSON format:
{{
    "thought": "reasoning",
    "tool": "tool_name",
    "parameters": {{}},
    "confidence": 0.0-1.0
}}
"""

    def _execute_action(self, action: Dict, timeout: int):
        tool_name = action["tool"]
        tool = self.tools[tool_name]
        return tool(**action["parameters"], timeout=timeout)

    def _verify_goal_met(self, result: Dict) -> Dict:
        # Implement domain-specific verification
        return {"goal_met": result.get("status") == "success"}

LOOP COMPONENTS & SUBSYSTEMS

1. Planning Engine (LLM Interface)

Responsibilities:

Convert current state to optimized prompt
Call LLM with appropriate temperature/max_tokens
Parse structured output (JSON schema validation)
Handle token budget constraints

Implementation:

class PlanningEngine:
    def __init__(self, model: str = "claude-sonnet-4-6", max_tokens: int = 2000):
        self.model = model
        self.max_tokens = max_tokens
        self.client = Anthropic()

    def generate_plan(self, 
                     goal: str, 
                     state: Dict, 
                     context: str,
                     tools_available: List[str]) -> Dict:
        """
        Generate next action(s) from LLM

        Args:
            goal: Loop objective
            state: Current execution state
            context: Historical context + observations
            tools_available: Available tool names

        Returns:
            Parsed action dict with validation
        """

        # Temperature varies by task criticality
        temperature = 0.3 if state["retry_count"] > 2 else 0.7

        prompt = f"""You are an autonomous AI agent executing this goal:
GOAL: {goal}

Current Execution State:
{json.dumps(state, indent=2)}

Recent Context:
{context}

Available Tools:
{', '.join(tools_available)}

You MUST respond with ONLY valid JSON (no markdown, no preamble):
{{
    "thought": "brief reasoning about next step",
    "action": "tool_name or 'finish'",
    "parameters": {{}},
    "confidence": 0.0-1.0,
    "justification": "why this action progresses toward goal"
}}
"""

        message = self.client.messages.create(
            model=self.model,
            max_tokens=self.max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )

        response_text = message.content[0].text

        # Robust JSON parsing with fallback
        try:
            action = json.loads(response_text)
        except json.JSONDecodeError:
            # Try to extract JSON from markdown code blocks
            match = re.search(r'```

(?:json)?\s*(.*?)\s*

```', response_text, re.DOTALL)
            if match:
                action = json.loads(match.group(1))
            else:
                raise ValueError(f"Failed to parse LLM response: {response_text}")

        # Validate schema
        self._validate_action_schema(action, tools_available)

        return action

    def _validate_action_schema(self, action: Dict, available_tools: List[str]):
        required_keys = {"thought", "action", "parameters"}
        if not required_keys.issubset(action.keys()):
            raise ValueError(f"Missing required keys. Got: {action.keys()}")

        if action["action"] not in available_tools and action["action"] != "finish":
            raise ValueError(f"Unknown tool: {action['action']}")

        if not isinstance(action.get("parameters"), dict):
            raise ValueError("parameters must be a dict")

2. Execution Engine (Tool/Action Runner)

Responsibilities:

Execute planned actions safely
Manage resource limits (timeout, memory)
Capture output and errors
Provide structured execution feedback

Implementation:

class ExecutionEngine:
    def __init__(self, tools: Dict[str, Callable], timeout: int = 30):
        self.tools = tools
        self.timeout = timeout
        self.execution_log = []

    def execute(self, action: Dict) -> ExecutionResult:
        """
        Safely execute a planned action with timeout protection
        """
        tool_name = action["action"]
        parameters = action.get("parameters", {})

        if tool_name == "finish":
            return ExecutionResult(
                status="success",
                output={"message": "Loop completed successfully"},
                tool=tool_name,
                duration=0
            )

        if tool_name not in self.tools:
            return ExecutionResult(
                status="error",
                error=f"Unknown tool: {tool_name}",
                tool=tool_name,
                duration=0
            )

        tool = self.tools[tool_name]
        start_time = time.time()

        try:
            # Execute with timeout using signal or ThreadPoolExecutor
            result = self._execute_with_timeout(tool, parameters)

            return ExecutionResult(
                status="success",
                output=result,
                tool=tool_name,
                duration=time.time() - start_time,
                parameters=parameters
            )

        except TimeoutError:
            return ExecutionResult(
                status="timeout",
                error=f"Tool '{tool_name}' exceeded {self.timeout}s timeout",
                tool=tool_name,
                duration=time.time() - start_time
            )

        except Exception as e:
            return ExecutionResult(
                status="error",
                error=str(e),
                error_type=type(e).__name__,
                tool=tool_name,
                duration=time.time() - start_time
            )

    def _execute_with_timeout(self, func: Callable, params: Dict, timeout: int = None):
        """Execute function with timeout protection"""
        timeout = timeout or self.timeout

        from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout

        with ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(func, **params)
            try:
                return future.result(timeout=timeout)
            except FutureTimeout:
                raise TimeoutError(f"Execution exceeded {timeout}s")

3. Verification Engine (Goal & State Validation)

Responsibilities:

Check if goal is achieved
Validate state consistency
Detect loops/infinite recursion
Assess progress toward objective

Implementation:

class VerificationEngine:
    def __init__(self, goal: str):
        self.goal = goal
        self.goal_verifier = None  # Domain-specific implementation

    def verify_goal_met(self, state: Dict, last_result: ExecutionResult) -> VerificationResult:
        """
        Determine if the loop goal has been achieved

        Returns:
            VerificationResult with:
            - goal_met: bool
            - confidence: float [0-1]
            - reason: str
            - suggested_action: str (retry, continue, abort)
        """

        # Check for explicit success markers
        if last_result.status == "success":
            if self._check_goal_success_criteria(state, last_result):
                return VerificationResult(
                    goal_met=True,
                    confidence=0.95,
                    reason="Goal success criteria met",
                    suggested_action="exit"
                )

        # Check for fatal errors
        if last_result.status == "error":
            if self._is_fatal_error(last_result):
                return VerificationResult(
                    goal_met=False,
                    confidence=1.0,
                    reason=f"Fatal error: {last_result.error}",
                    suggested_action="abort"
                )

        # Check for infinite loops (same action repeating)
        if self._detect_infinite_loop(state):
            return VerificationResult(
                goal_met=False,
                confidence=0.9,
                reason="Infinite loop detected",
                suggested_action="abort"
            )

        # Progress check - are we making forward progress?
        if self._detect_no_progress(state):
            return VerificationResult(
                goal_met=False,
                confidence=0.7,
                reason="No progress for N iterations",
                suggested_action="retry_different_approach"
            )

        # Default: continue looping
        return VerificationResult(
            goal_met=False,
            confidence=0.5,
            reason="Goal not yet achieved, continuing",
            suggested_action="continue"
        )

    def _check_goal_success_criteria(self, state: Dict, result: ExecutionResult) -> bool:
        """
        Domain-specific implementation
        Examples:
        - Check if output matches expected format
        - Verify file exists with correct content
        - Validate test suite passes
        """
        # This is implemented per-domain
        if self.goal_verifier:
            return self.goal_verifier(state, result)
        return False

    def _detect_infinite_loop(self, state: Dict) -> bool:
        """Detect if same action repeats N times"""
        history = state.get("history", [])
        if len(history) < 3:
            return False

        # Check last 3 actions
        last_three = [h["action"]["action"] for h in history[-3:]]
        return last_three[0] == last_three[1] == last_three[2]

    def _detect_no_progress(self, state: Dict) -> bool:
        """Check if state has advanced in last N iterations"""
        history = state.get("history", [])
        if len(history) < 5:
            return False

        # Measure state deltas
        recent_deltas = self._compute_state_deltas(history[-5:])
        return all(delta < 0.1 for delta in recent_deltas)  # All changes < 10%

    def _compute_state_deltas(self, history: List) -> List[float]:
        """Compute change magnitude for each iteration"""
        deltas = []
        for i in range(1, len(history)):
            prev_state = history[i-1].get("result", {})
            curr_state = history[i].get("result", {})
            delta = self._state_diff_magnitude(prev_state, curr_state)
            deltas.append(delta)
        return deltas

    def _state_diff_magnitude(self, state1: Dict, state2: Dict) -> float:
        """Compute normalized difference between states (0-1)"""
        # Simple heuristic: ratio of different keys to total keys
        all_keys = set(state1.keys()) | set(state2.keys())
        if not all_keys:
            return 0.0
        different = sum(1 for k in all_keys if state1.get(k) != state2.get(k))
        return different / len(all_keys)

STATE MANAGEMENT

1. State Store Architecture

Loops require external, persistent state because LLMs are stateless. State must survive loop iterations and be accessible to the planning engine.

State Layers:

┌─────────────────────────────────────────────────────┐
│            LAYER 1: WORKING STATE                    │
│  (In-memory, fast access)                           │
│  • Current iteration number                         │
│  • Last execution result                            │
│  • Short-term flags                                 │
│                                                      │
│  Storage: Redis, in-memory dict                     │
│  TTL: Duration of loop execution                    │
└─────────────────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────┐
│            LAYER 2: PERSISTENT STATE                 │
│  (Database, durable)                                │
│  • Full execution history                           │
│  • Artifacts (generated files)                      │
│  • Metrics & performance data                       │
│                                                      │
│  Storage: PostgreSQL, S3, Git                       │
│  TTL: Indefinite                                    │
└─────────────────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────┐
│            LAYER 3: CONTEXT MEMORY                   │
│  (Long-form reference for LLM)                      │
│  • Previous iterations summaries                    │
│  • Relevant code/docs                               │
│  • Error patterns & solutions                       │
│                                                      │
│  Storage: Vector DB, File System                    │
│  TTL: Loop-scoped or permanent                      │
└─────────────────────────────────────────────────────┘

2. State Schema

@dataclass
class LoopState:
    """Complete state snapshot for a loop execution"""

    # Metadata
    loop_id: str                          # UUID
    goal: str                            # Loop objective
    created_at: float                    # Unix timestamp

    # Execution State
    iterations: int                      # Current iteration count
    max_iterations: int                  # Termination limit
    status: str                          # "running", "success", "failed", "timeout"

    # History
    history: List[IterationRecord]       # All iteration results

    # Artifacts
    artifacts: Dict[str, str]            # {"file_path": "content", ...}
    generated_files: List[str]           # Files created during loop

    # Metrics
    total_tokens_used: int               # Sum of all LLM tokens
    total_cost: float                    # USD cost estimate
    wall_clock_duration: float           # Seconds

    # Context
    context_window: str                  # Serialized state for next LLM call
    error_log: List[str]                 # Error messages for recovery

    def serialize(self) -> str:
        """Serialize to JSON for storage"""
        return json.dumps(asdict(self), default=str)

    @classmethod
    def deserialize(cls, data: str) -> 'LoopState':
        """Load from JSON"""
        return cls(**json.loads(data))

    def to_context_string(self) -> str:
        """Serialize state for LLM prompt injection"""
        return f"""
EXECUTION STATE:
- Iteration: {self.iterations}/{self.max_iterations}
- Status: {self.status}
- Tokens Used: {self.total_tokens_used}
- Cost: ${self.total_cost:.4f}
- Duration: {self.wall_clock_duration:.1f}s

RECENT HISTORY:
{self._serialize_recent_history()}

ARTIFACTS CREATED:
{json.dumps(self.generated_files, indent=2)}

ERROR LOG:
{json.dumps(self.error_log[-5:], indent=2)}  # Last 5 errors
"""

    def _serialize_recent_history(self, limit: int = 5) -> str:
        recent = self.history[-limit:]
        return "\n".join([
            f"  [{r.iteration}] {r.action['action']} → {r.result['status']}"
            for r in recent
        ])

@dataclass
class IterationRecord:
    """Single iteration record"""
    iteration: int
    timestamp: float
    action: Dict                    # {"action": "...", "parameters": {...}}
    result: ExecutionResult
    tokens_used: int
    cost: float
    duration: float

3. State Persistence

class StateStore:
    """Unified state persistence layer"""

    def __init__(self, backend: str = "postgres"):
        self.backend = backend
        if backend == "postgres":
            self.db = PostgresBackend()
        elif backend == "redis":
            self.db = RedisBackend()
        elif backend == "sqlite":
            self.db = SQLiteBackend()

    def save_state(self, state: LoopState) -> None:
        """Persist state after each iteration"""
        self.db.insert("loop_states", {
            "loop_id": state.loop_id,
            "goal": state.goal,
            "iteration": state.iterations,
            "status": state.status,
            "state_json": state.serialize(),
            "updated_at": datetime.utcnow().isoformat(),
            "tokens_used": state.total_tokens_used,
            "cost": state.total_cost
        })

    def load_state(self, loop_id: str) -> LoopState:
        """Restore state from storage"""
        record = self.db.query_one(
            "SELECT state_json FROM loop_states WHERE loop_id = %s ORDER BY updated_at DESC",
            (loop_id,)
        )
        return LoopState.deserialize(record["state_json"])

    def save_iteration(self, loop_id: str, record: IterationRecord) -> None:
        """Append iteration record"""
        self.db.insert("iteration_records", {
            "loop_id": loop_id,
            "iteration_num": record.iteration,
            "action": json.dumps(record.action),
            "result": json.dumps(asdict(record.result)),
            "tokens_used": record.tokens_used,
            "cost": record.cost,
            "duration": record.duration,
            "created_at": datetime.fromtimestamp(record.timestamp).isoformat()
        })

    def get_history(self, loop_id: str, limit: int = 20) -> List[IterationRecord]:
        """Fetch iteration history"""
        records = self.db.query(
            """SELECT * FROM iteration_records 
               WHERE loop_id = %s 
               ORDER BY iteration_num DESC 
               LIMIT %s""",
            (loop_id, limit)
        )
        return [IterationRecord(**r) for r in records]

CONTROL FLOW & TERMINATION LOGIC

1. Termination Conditions

A loop must have multiple, explicit exit criteria:

class TerminationController:
    """Manages loop exit conditions"""

    def __init__(self, 
                 max_iterations: int = 20,
                 timeout_seconds: int = 300,
                 max_cost_usd: float = 10.0):
        self.max_iterations = max_iterations
        self.timeout = timeout_seconds
        self.max_cost = max_cost_usd
        self.iteration_timeout = 30  # Per-iteration limit

    def should_continue(self, state: LoopState) -> Tuple[bool, str]:
        """
        Determine if loop should continue

        Returns: (should_continue: bool, reason: str)
        """

        # Check 1: Goal achieved
        if state.status == "success":
            return (False, "Goal achieved")

        # Check 2: Max iterations exceeded
        if state.iterations >= self.max_iterations:
            return (False, f"Max iterations ({self.max_iterations}) reached")

        # Check 3: Total timeout exceeded
        if state.wall_clock_duration > self.timeout:
            return (False, f"Total timeout ({self.timeout}s) exceeded")

        # Check 4: Cost limit exceeded
        if state.total_cost >= self.max_cost:
            return (False, f"Cost limit (${self.max_cost}) exceeded")

        # Check 5: Fatal error occurred
        if state.status == "fatal_error":
            return (False, "Fatal error encountered")

        # Check 6: Unrecoverable state
        if self._is_unrecoverable(state):
            return (False, "Unrecoverable state detected")

        # Otherwise, continue
        return (True, "Continue looping")

    def _is_unrecoverable(self, state: LoopState) -> bool:
        """Check for states that cannot recover"""

        # Pattern: Same error repeating N times
        if len(state.error_log) >= 3:
            recent_errors = state.error_log[-3:]
            if all(e == recent_errors[0] for e in recent_errors):
                return True

        # Pattern: Cycling between two states infinitely
        if len(state.history) >= 4:
            actions = [h.action["action"] for h in state.history[-4:]]
            if actions[0] == actions[2] and actions[1] == actions[3]:
                return True

        return False

2. Retry & Backoff Strategy

class RetryController:
    """Intelligent retry logic with exponential backoff"""

    def __init__(self, 
                 max_retries: int = 3,
                 backoff_base: float = 2.0,
                 jitter: bool = True):
        self.max_retries = max_retries
        self.backoff_base = backoff_base
        self.jitter = jitter
        self.retry_counts = {}  # {action_id: count}

    def should_retry(self, state: LoopState, last_result: ExecutionResult) -> bool:
        """Determine if action should be retried"""

        # Never retry if goal is achieved
        if state.status == "success":
            return False

        # Only retry on recoverable errors
        if last_result.status not in ["timeout", "error"]:
            return False

        # Check retry limit
        action_id = state.history[-1].action.get("action")
        retries = self.retry_counts.get(action_id, 0)
        if retries >= self.max_retries:
            return False

        # Don't retry fatal errors
        if self._is_fatal_error(last_result):
            return False

        return True

    def get_backoff_delay(self, retry_count: int) -> float:
        """Calculate exponential backoff with optional jitter"""
        delay = self.backoff_base ** retry_count

        if self.jitter:
            delay *= (0.5 + random.random())  # 50-150% of computed delay

        return delay

    def _is_fatal_error(self, result: ExecutionResult) -> bool:
        """Identify errors that shouldn't be retried"""
        fatal_errors = [
            "KeyError", "ValueError", "AttributeError",  # Code bugs
            "PermissionError", "FileNotFoundError",       # Resource errors
            "SyntaxError", "ImportError"                  # Config errors
        ]
        return result.error_type in fatal_errors

ERROR HANDLING & RECOVERY

1. Error Classification

class ErrorClassification:
    """Categorize errors for appropriate handling"""

    TRANSIENT = {
        "timeout": "Exceeded execution time limit",
        "rate_limit": "API rate limit exceeded",
        "network": "Network connectivity issue",
        "memory": "Temporary memory pressure"
    }

    RECOVERABLE = {
        "invalid_tool": "Tool doesn't exist, try different tool",
        "malformed_params": "Parameter format incorrect, fix and retry",
        "missing_dependency": "Required file/tool missing, create it first"
    }

    FATAL = {
        "permission_denied": "Insufficient permissions, cannot proceed",
        "syntax_error": "Code has syntax errors, manual fix required",
        "resource_exhausted": "Cannot allocate required resources",
        "auth_failed": "Authentication failure, credentials invalid"
    }

    @classmethod
    def classify(cls, error: Exception, context: Dict) -> Tuple[str, str]:
        """
        Classify error and suggest recovery action

        Returns: (category, suggested_action)
        """
        error_name = type(error).__name__
        error_msg = str(error).lower()

        # Match against patterns
        for category, errors in [
            ("transient", cls.TRANSIENT),
            ("recoverable", cls.RECOVERABLE),
            ("fatal", cls.FATAL)
        ]:
            for error_key, description in errors.items():
                if error_key in error_msg or error_key in error_name.lower():
                    return (category, error_key)

        # Default classification
        if "timeout" in error_msg:
            return ("transient", "timeout")
        if "error" in error_msg:
            return ("recoverable", "unknown_error")

        return ("fatal", "unclassified")

class ErrorRecoveryStrategy:
    """Map error classifications to recovery actions"""

    RECOVERY_ACTIONS = {
        "timeout": {
            "action": "retry_with_longer_timeout",
            "max_retries": 2,
            "delay_seconds": 5
        },
        "rate_limit": {
            "action": "backoff_and_retry",
            "max_retries": 3,
            "delay_seconds": 30
        },
        "invalid_tool": {
            "action": "replan_with_available_tools",
            "suggest_alternatives": True,
            "max_retries": 1
        },
        "missing_dependency": {
            "action": "create_dependency_first",
            "max_retries": 2
        },
        "syntax_error": {
            "action": "abort_loop",
            "log_error": True
        }
    }

    @staticmethod
    def handle_error(error: Exception, 
                    error_type: str, 
                    state: LoopState) -> Dict:
        """
        Handle error and return recovery strategy
        """
        strategy = ErrorRecoveryStrategy.RECOVERY_ACTIONS.get(
            error_type, 
            {"action": "abort_loop"}
        )

        return {
            "should_retry": strategy.get("max_retries", 0) > 0,
            "action": strategy["action"],
            "delay": strategy.get("delay_seconds", 0),
            "retry_limit": strategy.get("max_retries", 0),
            "log_context": {
                "error": str(error),
                "iteration": state.iterations,
                "cost_so_far": state.total_cost
            }
        }

2. Circuit Breaker Pattern

class CircuitBreaker:
    """Prevent cascading failures with circuit breaker pattern"""

    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # "closed", "open", "half_open"

    def record_success(self):
        """Mark successful execution"""
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self):
        """Mark failed execution"""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            raise CircuitBreakerOpen(
                f"Circuit opened after {self.failure_threshold} failures"
            )

    def can_execute(self) -> bool:
        """Check if action should be attempted"""

        if self.state == "closed":
            return True

        if self.state == "open":
            # Try to transition to half-open if timeout passed
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half_open"
                return True
            return False

        if self.state == "half_open":
            return True

        return False

    def on_half_open_success(self):
        """Reset after successful test in half-open state"""
        self.failure_count = 0
        self.state = "closed"

    def on_half_open_failure(self):
        """Return to open state"""
        self.state = "open"
        self.last_failure_time = time.time()

CONTEXT MANAGEMENT

1. Rolling Context Window

Since LLMs have finite context windows, loops must manage what information flows to the planning engine:

class ContextManager:
    """Manage context injection for LLM planning"""

    def __init__(self, max_context_tokens: int = 8000):
        self.max_context_tokens = max_context_tokens
        self.tokenizer = Tokenizer()  # Claude tokenizer

    def build_planning_context(self, 
                              state: LoopState,
                              goal: str,
                              tools: List[str]) -> str:
        """
        Build optimized context for planning LLM call

        Strategy:
        1. Goal (fixed)
        2. Current iteration state (variable)
        3. Recent history summaries (compressed)
        4. Relevant artifacts (sampled)
        5. Error context (if applicable)
        """

        sections = []
        tokens_used = 0

        # SECTION 1: Goal (Always included, high priority)
        goal_section = f"OBJECTIVE:\n{goal}\n\n"
        sections.append(("goal", goal_section))
        tokens_used += self.tokenizer.count_tokens(goal_section)

        # SECTION 2: Current State
        state_section = f"""CURRENT STATE (Iteration {state.iterations}/{state.max_iterations}):
- Status: {state.status}
- Cost So Far: ${state.total_cost:.4f}
- Duration: {state.wall_clock_duration:.1f}s
- Available Tools: {', '.join(tools)}
"""
        sections.append(("state", state_section))
        tokens_used += self.tokenizer.count_tokens(state_section)

        # SECTION 3: Recent History (Compressed)
        history_section = self._compress_history(state.history[-10:])
        sections.append(("history", history_section))
        tokens_used += self.tokenizer.count_tokens(history_section)

        # SECTION 4: Artifacts (Sampled if too large)
        if tokens_used < self.max_context_tokens * 0.7:
            artifacts_section = self._sample_artifacts(state.artifacts)
            sections.append(("artifacts", artifacts_section))
            tokens_used += self.tokenizer.count_tokens(artifacts_section)

        # SECTION 5: Error Context (If errors occurred)
        if state.error_log:
            errors_section = f"""RECENT ERRORS:
{chr(10).join(state.error_log[-3:])}
"""
            sections.append(("errors", errors_section))
            tokens_used += self.tokenizer.count_tokens(errors_section)

        # Trim if over budget
        if tokens_used > self.max_context_tokens:
            sections = self._trim_sections(sections, self.max_context_tokens)

        return "\n".join([section for _, section in sections])

    def _compress_history(self, history: List[IterationRecord]) -> str:
        """Summarize history in compact format"""
        lines = ["HISTORY:"]
        for record in history[-10:]:
            action = record.action.get("action", "unknown")
            status = record.result.get("status", "unknown")
            lines.append(f"  [{record.iteration}] {action} → {status}")
        return "\n".join(lines) + "\n"

    def _sample_artifacts(self, artifacts: Dict[str, str]) -> str:
        """Include relevant artifacts, truncate if needed"""
        section = "GENERATED ARTIFACTS:\n"
        for path, content in artifacts.items():
            # Include first 500 chars
            preview = content[:500] + ("..." if len(content) > 500 else "")
            section += f"\n{path}:\n{preview}\n"
        return section

    def _trim_sections(self, sections: List[Tuple], max_tokens: int) -> List[Tuple]:
        """Drop least important sections to fit budget"""
        priority = ["goal", "state", "history", "artifacts", "errors"]

        sorted_sections = sorted(
            sections,
            key=lambda x: priority.index(x[0]) if x[0] in priority else 999
        )

        trimmed = []
        tokens = 0
        for name, section in sorted_sections:
            section_tokens = self.tokenizer.count_tokens(section)
            if tokens + section_tokens <= max_tokens:
                trimmed.append((name, section))
                tokens += section_tokens

        return trimmed

2. Memory Types in Loops

class LoopMemory:
    """Multi-tier memory system for loops"""

    def __init__(self):
        self.short_term = {}        # Iteration-level state
        self.working = {}            # Task-level artifacts
        self.long_term = {}          # Loop-lifetime knowledge
        self.external = {}           # Files, Git, databases

    # SHORT-TERM (lasts 1-2 iterations)
    # Used for: immediate action results, flags

    # WORKING (lasts duration of loop)
    # Used for: generated code, intermediate outputs

    # LONG-TERM (persists across loops)
    # Used for: learned patterns, solutions to recurring problems

    # EXTERNAL (persists indefinitely)
    # Used for: final artifacts, version control, audit log

VERIFICATION & VALIDATION SYSTEMS

1. Multi-Layer Verification

class VerificationFramework:
    """Multi-layer verification for robust goal achievement"""

    def __init__(self):
        self.assertions = []
        self.validators = []

    def register_assertion(self, name: str, predicate: Callable[[Dict], bool]):
        """Register a verifiable assertion about success"""
        self.assertions.append({"name": name, "check": predicate})

    def register_validator(self, name: str, validator: Callable[[Dict], bool]):
        """Register a validator that must pass"""
        self.validators.append({"name": name, "check": validator})

    def verify_goal_achieved(self, state: LoopState, result: Dict) -> VerificationReport:
        """
        Multi-layer verification:
        Layer 1: Assertions (conditions that must be true)
        Layer 2: Validators (functional validation)
        Layer 3: Integration tests (end-to-end)
        """

        report = VerificationReport()

        # LAYER 1: Assertions
        assertion_results = []
        for assertion in self.assertions:
            try:
                passed = assertion["check"](result)
                assertion_results.append({
                    "assertion": assertion["name"],
                    "passed": passed,
                    "weight": 1.0
                })
            except Exception as e:
                assertion_results.append({
                    "assertion": assertion["name"],
                    "passed": False,
                    "error": str(e)
                })

        # LAYER 2: Validators
        validator_results = []
        for validator in self.validators:
            try:
                passed = validator["check"](result)
                validator_results.append({
                    "validator": validator["name"],
                    "passed": passed,
                    "required": True
                })
            except Exception as e:
                validator_results.append({
                    "validator": validator["name"],
                    "passed": False,
                    "error": str(e)
                })

        # LAYER 3: Integration Check
        integration_passed = self._run_integration_test(result)

        # Aggregate
        all_assertions_pass = all(a["passed"] for a in assertion_results)
        all_validators_pass = all(v["passed"] for v in validator_results)

        report.success = all_assertions_pass and all_validators_pass and integration_passed
        report.confidence = self._compute_confidence(
            assertion_results, validator_results, integration_passed
        )
        report.details = {
            "assertions": assertion_results,
            "validators": validator_results,
            "integration": integration_passed
        }

        return report

    def _run_integration_test(self, result: Dict) -> bool:
        """Run end-to-end integration test"""
        # Domain-specific implementation
        try:
            # Example: Run test suite, verify artifact quality, etc.
            return True
        except:
            return False

    def _compute_confidence(self, assertions, validators, integration) -> float:
        """Compute confidence [0-1]"""
        scores = []

        # Assertion confidence
        assertion_pass_rate = sum(1 for a in assertions if a["passed"]) / len(assertions)
        scores.append(assertion_pass_rate * 0.4)

        # Validator confidence
        validator_pass_rate = sum(1 for v in validators if v["passed"]) / len(validators)
        scores.append(validator_pass_rate * 0.4)

        # Integration confidence
        scores.append(integration * 0.2)

        return sum(scores)

@dataclass
class VerificationReport:
    success: bool
    confidence: float  # [0-1]
    details: Dict

COST OPTIMIZATION & TOKEN MANAGEMENT

1. Token Budget System

class TokenBudgetManager:
    """Track and optimize token usage across loop"""

    def __init__(self, budget_usd: float = 10.0, model: str = "claude-sonnet-4-6"):
        self.budget_usd = budget_usd
        self.model = model
        self.pricing = self._load_pricing()
        self.spent = 0.0
        self.token_log = []

    def _load_pricing(self) -> Dict:
        """Current pricing (June 2026)"""
        return {
            "claude-opus-4-6": {"input": 0.015/1000, "output": 0.045/1000},
            "claude-sonnet-4-6": {"input": 0.003/1000, "output": 0.015/1000},
            "claude-haiku-4-5": {"input": 0.0008/1000, "output": 0.004/1000}
        }

    def can_afford_call(self, estimated_input_tokens: int) -> bool:
        """Check if next LLM call fits budget"""
        estimated_cost = self._estimate_cost(estimated_input_tokens, 2000)
        return self.spent + estimated_cost <= self.budget_usd

    def record_call(self, input_tokens: int, output_tokens: int) -> float:
        """Log token usage and return cost"""
        cost = self._compute_cost(input_tokens, output_tokens)
        self.spent += cost
        self.token_log.append({
            "input": input_tokens,
            "output": output_tokens,
            "cost": cost,
            "timestamp": time.time()
        })
        return cost

    def _estimate_cost(self, input_tokens: int, output_tokens: int) -> float:
        """Estimate cost before call"""
        pricing = self.pricing[self.model]
        return (input_tokens * pricing["input"] + 
                output_tokens * pricing["output"])

    def _compute_cost(self, input_tokens: int, output_tokens: int) -> float:
        """Compute actual cost"""
        return self._estimate_cost(input_tokens, output_tokens)

    def get_remaining_budget(self) -> float:
        return self.budget_usd - self.spent

    def get_budget_report(self) -> Dict:
        return {
            "total_budget": self.budget_usd,
            "spent": self.spent,
            "remaining": self.get_remaining_budget(),
            "percent_used": (self.spent / self.budget_usd) * 100,
            "total_tokens": sum(t["input"] + t["output"] for t in self.token_log),
            "call_count": len(self.token_log)
        }

class TokenOptimizer:
    """Optimize token usage through compression and caching"""

    @staticmethod
    def compress_history(history: List[IterationRecord], ratio: float = 0.5) -> str:
        """Compress history summaries to reduce context size"""
        # Keep first and last iterations completely
        # Summarize middle iterations
        if len(history) <= 4:
            return "\n".join(f"[{r.iteration}] {r.action['action']}" for r in history)

        compressed = []
        compressed.append(f"[{history[0].iteration}] {history[0].action['action']}")

        # Compress middle
        middle = history[1:-1]
        summary = f"... {len(middle)} intermediate iterations (actions: {set(m.action['action'] for m in middle)}) ..."
        compressed.append(summary)

        compressed.append(f"[{history[-1].iteration}] {history[-1].action['action']}")

        return "\n".join(compressed)

    @staticmethod
    def cache_results(result: Dict, key: str) -> None:
        """Cache successful results to avoid recomputation"""
        import hashlib
        hash_key = hashlib.md5(json.dumps(result).encode()).hexdigest()
        # Store in Redis or cache system
        pass

ADVANCED PATTERNS

1. Parallel Loop Execution

class ParallelLoopCoordinator:
    """Coordinate multiple parallel loops with shared state"""

    def __init__(self, max_parallel: int = 3):
        self.max_parallel = max_parallel
        self.loops = {}
        self.shared_state = {}
        self.lock = threading.Lock()

    def spawn_loop(self, loop_id: str, goal: str) -> 'AutonomousLoop':
        """Create new autonomous loop"""
        if len(self.loops) >= self.max_parallel:
            raise RuntimeError("Max parallel loops reached")

        loop = AutonomousLoop(goal)
        self.loops[loop_id] = loop
        return loop

    def run_parallel(self) -> Dict[str, LoopResult]:
        """Execute all loops in parallel"""
        futures = {}
        with ThreadPoolExecutor(max_workers=self.max_parallel) as executor:
            for loop_id, loop in self.loops.items():
                futures[loop_id] = executor.submit(loop.run)

        results = {}
        for loop_id, future in futures.items():
            results[loop_id] = future.result()

        return results

    def coordinate_state(self, loop_id: str, state_update: Dict) -> None:
        """Share state between parallel loops"""
        with self.lock:
            self.shared_state.update(state_update)

2. Nested Loops

class NestedLoopController:
    """Support hierarchical/nested loop execution"""

    def __init__(self, parent_goal: str):
        self.parent_goal = parent_goal
        self.subloops = []
        self.parent_state = {}

    def create_subloop(self, subgoal: str) -> 'AutonomousLoop':
        """Create nested loop for subtask"""
        subloop = AutonomousLoop(subgoal)
        subloop.parent_state = self.parent_state
        self.subloops.append(subloop)
        return subloop

    def run_hierarchical(self) -> LoopResult:
        """Execute parent loop with subloop management"""
        for subloop in self.subloops:
            result = subloop.run()
            if not result.success:
                return LoopResult(success=False, reason="subloop_failed")

        return LoopResult(success=True)

3. Adaptive Loop Configuration

class AdaptiveLoopController:
    """Adjust loop parameters based on performance"""

    def __init__(self, initial_config: Dict):
        self.config = initial_config
        self.metrics = []

    def adapt_parameters(self, state: LoopState) -> None:
        """Dynamically adjust loop behavior based on metrics"""

        # Measure success rate
        success_rate = self._compute_success_rate(state)

        # Adjust temperature based on performance
        if success_rate < 0.3:
            # Increase determinism for struggling loops
            self.config["temperature"] = 0.3
        elif success_rate > 0.8:
            # Can afford more exploration
            self.config["temperature"] = 0.8

        # Adjust max_iterations based on progress
        if self._is_making_progress(state):
            self.config["max_iterations"] = min(
                self.config["max_iterations"] + 5, 50
            )
        else:
            self.config["max_iterations"] = max(
                self.config["max_iterations"] - 5, 10
            )

IMPLEMENTATION EXAMPLES

Example 1: Code Generation Loop

def create_code_generation_loop(
    requirements: str,
    language: str = "python",
    budget: float = 5.0
) -> AutonomousLoop:
    """Create a loop that generates and tests code"""

    def generate_tool(specs: str) -> Dict:
        """Generate code based on specs"""
        # Implementation: Call LLM to generate code
        pass

    def test_tool(code: str) -> Dict:
        """Execute generated code and test it"""
        try:
            exec_globals = {}
            exec(code, exec_globals)
            return {"status": "success", "output": exec_globals}
        except Exception as e:
            return {"status": "error", "error": str(e)}

    def lint_tool(code: str) -> Dict:
        """Lint and validate code quality"""
        # Run pylint, mypy, etc.
        pass

    loop = AutonomousLoop(
        goal=requirements,
        llm_client=Anthropic(),
        tools={
            "generate": generate_tool,
            "test": test_tool,
            "lint": lint_tool
        }
    )

    loop.budget_manager = TokenBudgetManager(budget)
    return loop

Example 2: Data Processing Loop

def create_data_processing_loop(
    input_file: str,
    processing_spec: str
) -> AutonomousLoop:
    """Create a loop that processes data iteratively"""

    def load_data_tool(path: str) -> Dict:
        import pandas as pd
        return {"data": pd.read_csv(path).to_dict()}

    def transform_tool(data: Dict, transformation: str) -> Dict:
        # Apply transformation
        pass

    def validate_tool(data: Dict) -> Dict:
        # Validate data quality
        pass

    def save_tool(data: Dict, path: str) -> Dict:
        import pandas as pd
        pd.DataFrame(data).to_csv(path)
        return {"status": "success", "path": path}

    loop = AutonomousLoop(
        goal=processing_spec,
        tools={
            "load": load_data_tool,
            "transform": transform_tool,
            "validate": validate_tool,
            "save": save_tool
        }
    )

    return loop

MONITORING & OBSERVABILITY

1. Loop Metrics & Telemetry

class LoopTelemetry:
    """Comprehensive monitoring for loop execution"""

    def __init__(self, loop_id: str):
        self.loop_id = loop_id
        self.metrics = {
            "iterations": 0,
            "successes": 0,
            "failures": 0,
            "retries": 0,
            "total_tokens": 0,
            "total_cost": 0.0,
            "start_time": time.time(),
            "errors": []
        }

    def record_iteration(self, success: bool, tokens: int, cost: float) -> None:
        self.metrics["iterations"] += 1
        if success:
            self.metrics["successes"] += 1
        else:
            self.metrics["failures"] += 1
        self.metrics["total_tokens"] += tokens
        self.metrics["total_cost"] += cost

    def record_error(self, error: Exception) -> None:
        self.metrics["errors"].append({
            "type": type(error).__name__,
            "message": str(error),
            "timestamp": time.time()
        })

    def get_metrics(self) -> Dict:
        duration = time.time() - self.metrics["start_time"]
        return {
            **self.metrics,
            "duration_seconds": duration,
            "success_rate": self.metrics["successes"] / max(self.metrics["iterations"], 1),
            "cost_per_iteration": self.metrics["total_cost"] / max(self.metrics["iterations"], 1),
            "tokens_per_iteration": self.metrics["total_tokens"] / max(self.metrics["iterations"], 1)
        }

class LoopDashboard:
    """Real-time dashboard for loop monitoring"""

    def __init__(self):
        self.loops = {}

    def register_loop(self, loop_id: str, telemetry: LoopTelemetry):
        self.loops[loop_id] = telemetry

    def render_dashboard(self) -> str:
        """Generate text-based dashboard"""
        output = "LOOP EXECUTION DASHBOARD\n"
        output += "=" * 80 + "\n\n"

        for loop_id, telemetry in self.loops.items():
            metrics = telemetry.get_metrics()
            output += f"Loop: {loop_id}\n"
            output += f"  Iterations: {metrics['iterations']}\n"
            output += f"  Success Rate: {metrics['success_rate']:.1%}\n"
            output += f"  Total Cost: ${metrics['total_cost']:.4f}\n"
            output += f"  Duration: {metrics['duration_seconds']:.1f}s\n"
            output += f"  Tokens/Iteration: {metrics['tokens_per_iteration']:.0f}\n"
            output += "\n"

        return output

2. Logging & Audit Trail

class LoopLogger:
    """Comprehensive audit logging for compliance and debugging"""

    def __init__(self, loop_id: str, log_file: str = None):
        self.loop_id = loop_id
        self.logger = self._setup_logger(log_file)

    def _setup_logger(self, log_file: str):
        import logging
        logger = logging.getLogger(f"loop.{self.loop_id}")

        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )

        # File handler
        if log_file:
            fh = logging.FileHandler(log_file)
            fh.setLevel(logging.DEBUG)
            fh.setFormatter(formatter)
            logger.addHandler(fh)

        # Console handler
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)
        ch.setFormatter(formatter)
        logger.addHandler(ch)

        return logger

    def log_iteration(self, iteration: int, action: Dict, result: Dict) -> None:
        self.logger.info(
            f"[ITERATION {iteration}] Action: {action['action']} → "
            f"Status: {result.get('status')}"
        )

    def log_error(self, error: Exception, context: Dict) -> None:
        self.logger.error(
            f"ERROR: {type(error).__name__}: {str(error)}\n"
            f"Context: {json.dumps(context, indent=2)}"
        )

    def log_state_snapshot(self, state: LoopState) -> None:
        self.logger.debug(
            f"STATE SNAPSHOT:\n{json.dumps(json.loads(state.serialize()), indent=2)}"
        )

PRODUCTION DEPLOYMENT CHECKLIST

[ ] Token budget enforcement implemented
[ ] Circuit breaker protection active
[ ] Cost limits enforced per-loop
[ ] Comprehensive error classification & recovery
[ ] State persistence configured (DB backend)
[ ] Monitoring/telemetry in place
[ ] Retry logic with exponential backoff
[ ] Verification framework implemented
[ ] Context management optimized
[ ] Logging/audit trail enabled
[ ] Graceful degradation strategies
[ ] Performance benchmarking completed
[ ] Cost projections validated
[ ] Runaway loop detection active

CONCLUSION

Loop engineering shifts AI development from reactive prompting to proactive system design. The key is building robust infrastructure around iterative autonomy: state management, verification, cost control and error recovery.

The loop becomes the unit of work, not the LLM call.

Send Your First SMS in Python with MojaWave

Pamui Afrika — Tue, 23 Jun 2026 10:19:06 +0000

If you build software in Tanzania or anywhere in East Africa, sending an SMS shouldn't be the hard part of your project. MojaWave gives you one REST endpoint that delivers straight to Vodacom, Tigo, Airtel, and Halotel.

In this post we'll send a single "Hello World" SMS in Python — no SDK required, just the requests library. Total time: about two minutes.

What you'll need

Python 3.8+
A MojaWave API key — grab one free here →
A phone number to text yourself When you sign up you get two kinds of keys:

Prefix	Environment	Use it for
`sk_test_mw_`	Sandbox	Development. No real SMS, no charges.
`mw_`	Live	Real messages to real phones.

Start with the sandbox key so you can run this with zero cost, then swap to your live key when you're ready.

Step 1 — Install requests

pip install requests

Step 2 — Write the script

Create a file called hello_sms.py:

import os
import requests

# Load your key from an environment variable — never hardcode it
API_KEY = os.environ["MOJAWAVE_API_KEY"]

response = requests.post(
    "https://api.mojawave.com/v1/sms/send",
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    },
    json={
        "to": "+255712345678",      # your number in E.164 format
        "from": "MojaWave",          # sender ID (or an approved one on your account)
        "message": "Hello World from MojaWave 🌊",
    },
)

data = response.json()

if response.ok:
    sms = data["data"]
    print(f"Sent ✅  id={sms['id']}  status={sms['status']}")
else:
    print(f"Failed ❌  {response.status_code}: {data}")

Security note: Your API key controls real spending. Keep it in an environment variable or secrets manager — never commit it to Git or ship it in client-side code.

Step 3 — Run it

Set your key and run:

export MOJAWAVE_API_KEY="sk_test_mw_your_key_here"
python hello_sms.py

You'll see something like:

Sent ✅  id=89b82624-f1a2-4f5e-85b5-102e79a06779  status=sent

That's it — your first SMS is on the wire.

What the API sends back

A successful call returns 201 Created with a JSON envelope:

{
  "success": true,
  "data": {
    "id": "89b82624-f1a2-4f5e-85b5-102e79a06779",
    "type": "sms",
    "to": "+255712345678",
    "status": "sent",
    "segments": 1,
    "credits_cost": 1,
    "queued_at": "2026-04-05T12:03:04.485Z",
    "sent_at": "2026-04-05T12:04:04.393Z"
  }
}

A few fields worth knowing:

status moves through queued → sent → delivered (or failed).
segments — messages over 160 characters are split automatically; each segment costs a credit.
credits_cost — how many credits this send used. ## Prefer the SDK?

If you'd rather not handle raw HTTP, the official Python SDK wraps the same call:

import mojawave

client = mojawave.Client(api_key="mw_your_key")
sms = client.sms.send(
    to="+255712345678",
    from_="MojaWave",
    message="Hello World from MojaWave 🌊",
)
print(sms.id, sms.status)

Where to go next

Once your Hello World lands, these are the natural next steps:

Delivery receipts — add a webhook_url to get real-time message.delivered callbacks (verified with HMAC-SHA256).
Bulk send — POST /v1/sms/bulk blasts one message to up to 10,000 recipients with per-recipient personalization.
Check your balance — GET /v1/credits returns your remaining SMS and email credits.
AI integration — the mojawave-mcp package exposes every endpoint as a tool for Claude, ChatGPT, Gemini, Cursor, and any MCP client, so you can send SMS in plain English. Full reference: mojawave.com/docs

Get your key

If you skipped ahead — here's where you start:

👉 Get your free MojaWave API key

Building something with SMS in East Africa? Drop a comment with what you're working on.

DEV Community: Pamui Afrika

LOOP ENGINEERING: TECHNICAL BLUEPRINT

Complete Architecture, Design Patterns & Implementation Guide for AI Systems

TABLE OF CONTENTS

EXECUTIVE OVERVIEW

Definition

Paradigm Shift

Why Loops > Prompts for Complex Tasks

CORE LOOP ARCHITECTURE

1. The Canonical Loop Cycle

2. Loop State Machine

3. Minimal Loop Implementation (Pseudocode)

LOOP COMPONENTS & SUBSYSTEMS

1. Planning Engine (LLM Interface)

2. Execution Engine (Tool/Action Runner)

3. Verification Engine (Goal & State Validation)

STATE MANAGEMENT

1. State Store Architecture

2. State Schema

3. State Persistence

CONTROL FLOW & TERMINATION LOGIC

1. Termination Conditions

2. Retry & Backoff Strategy

ERROR HANDLING & RECOVERY

1. Error Classification

2. Circuit Breaker Pattern

CONTEXT MANAGEMENT

1. Rolling Context Window

2. Memory Types in Loops

VERIFICATION & VALIDATION SYSTEMS

1. Multi-Layer Verification

COST OPTIMIZATION & TOKEN MANAGEMENT

1. Token Budget System

ADVANCED PATTERNS

1. Parallel Loop Execution

2. Nested Loops

3. Adaptive Loop Configuration

IMPLEMENTATION EXAMPLES

Example 1: Code Generation Loop

Example 2: Data Processing Loop

MONITORING & OBSERVABILITY

1. Loop Metrics & Telemetry

2. Logging & Audit Trail

PRODUCTION DEPLOYMENT CHECKLIST

CONCLUSION

Send Your First SMS in Python with MojaWave

What you'll need

Step 1 — Install requests

Step 2 — Write the script

Step 3 — Run it

What the API sends back

Where to go next

Get your key