DEV Community

MUHAMMAD MUSADIQ
MUHAMMAD MUSADIQ

Posted on

Handling API Failures: How to Code Dynamic Re-Planning Loops for Autonomous Agents

Autonomous agents break the moment an external API returns a 500 error, hits a rate limit, or sends unexpected JSON. If your agent relies on rigid, linear tool execution, a single failed endpoint can bring down the entire system, waste tokens, and leave the user hanging. To fix this, you need a dynamic replanning loop that feeds API errors back into the LLM's context, updates the agent's state machine, and selects an alternate path forward without crashing.

I learned this the hard way while building an automated scraper agent that kept dying whenever a target site changed its schema or rate-limited my requests. Once I shifted from hardcoded fallbacks to state-aware re-planning, my pipeline success rate jumped dramatically.

Here is how to code a resilient re-planning loop from scratch.

  • Python 3.10 or higher.
  • An OpenAI API key (or any provider supporting JSON mode and tool calls).
  • pip install openai pydantic installed in your environment.

Step 1: Define the Agent State Machine

First, create a state class to track task goals, completed steps, failed tools, and error context. This state gives the model full visibility into what went wrong so it does not repeat the same bad API call.

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional

@dataclass
class AgentState:
    goal: str
    completed_steps: List[str] = field(default_factory=list)
    failed_attempts: List[Dict[str, Any]] = field(default_factory=list)
    available_tools: List[str] = field(default_factory=list)
    is_complete: bool = False
    final_output: Optional[str] = None
Enter fullscreen mode Exit fullscreen mode

Step 2: Safe Tool Execution Wrapper

Never let a raw HTTP error bubble up and kill your application process. Wrap your API integrations so they catch exceptions and standardise the error payload.

import requests

def call_weather_api(city: str) -> dict:
    # Example tool that simulates an API failure
    if city.lower() == "bad_city":
        raise requests.exceptions.HTTPError("500 Internal Server Error: Service Unavailable")
    return {"temp": 72, "condition": "Sunny"}

def safe_execute_tool(tool_name: str, tool_func, **kwargs) -> dict:
    try:
        result = tool_func(**kwargs)
        return {"status": "success", "result": result}
    except Exception as e:
        return {
            "status": "error", 
            "tool": tool_name, 
            "args": kwargs, 
            "error_message": str(e)
        }
Enter fullscreen mode Exit fullscreen mode

This structural isolation is critical when building complex workflow automation platforms where third-party integrations fail constantly.

Step 3: Build the Re-Planning Loop Logic

Now, assemble the core engine. When a tool call succeeds, record the result and move forward. When it fails, append the failure details to failed_attempts and prompt the LLM to either try a different tool or modify its parameters.

import json
from openai import OpenAI

client = OpenAI()

def run_agent_loop(goal: str, max_retries: int = 3):
    state = AgentState(
        goal=goal,
        available_tools=["call_weather_api", "search_web_backup"]
    )

    attempts = 0
    while not state.is_complete and attempts < max_retries:
        attempts += 1

        # Construct the context window dynamically based on state
        system_prompt = (
            "You are an autonomous agent with dynamic re-planning abilities. "
            "Your goal is to complete the request. If a tool fails, evaluate the error message "
            "and decide on an alternate step or tool. Do not repeat failed tool calls with identical arguments."
        )

        user_prompt = f"""
Goal: {state.goal}
Completed Steps: {state.completed_steps}
Failed Attempts: {json.dumps(state.failed_attempts)}
Available Tools: {state.available_tools}

Respond with a JSON object containing:
1. "thought": Your reasoning on what to do next based on past failures.
2. "action": The tool name to call, or "FINISH".
3. "args": Key-value arguments for the tool.
4. "final_response": Summary string if action is "FINISH".
"""

        response = client.chat.completions.create(
            model="gpt-4o",
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ]
        )

        plan = json.loads(response.choices[0].message.content)
        print(f"\n[Attempt {attempts}] LLM Thought: {plan['thought']}")

        action = plan.get("action")
        if action == "FINISH":
            state.is_complete = True
            state.final_output = plan.get("final_response")
            break

        # Execute selected action
        if action == "call_weather_api":
            res = safe_execute_tool("call_weather_api", call_weather_api, **plan.get("args", {}))
            if res["status"] == "success":
                state.completed_steps.append(f"Fetched weather: {res['result']}")
            else:
                state.failed_attempts.append(res)
                print(f"[Warning] Tool failure recorded: {res['error_message']}")

        elif action == "search_web_backup":
            # Mock fallback tool
            state.completed_steps.append("Fallback tool executed: Found estimated weather online.")

    return state
Enter fullscreen mode Exit fullscreen mode

This basic structure forms the core pattern for robust AI agent development, keeping your processes alive even when web endpoints go down.

Step 4: Run the Agent with a Failing Trigger

Test the loop by passing an input that triggers a simulated API crash.

if __name__ == "__main__":
    # Test with an invalid city that forces an API error on step 1
    final_state = run_agent_loop(
        goal="Get the current weather for bad_city. If that fails, find a backup answer."
    )

    print("\n--- Final Agent Summary ---")
    print("Success:", final_state.is_complete)
    print("Completed Steps:", final_state.completed_steps)
    print("Final Output:", final_state.final_output)
Enter fullscreen mode Exit fullscreen mode

Expected Outcomes

When you execute the code above, you should see output similar to this:

[Attempt 1] LLM Thought: I will call call_weather_api for bad_city.
[Warning] Tool failure recorded: 500 Internal Server Error: Service Unavailable

[Attempt 2] LLM Thought: The call_weather_api tool failed with a 500 error for bad_city. I will switch to the backup search tool instead.

--- Final Agent Summary ---
Success: True
Completed Steps: ['Fallback tool executed: Found estimated weather online.']
Final Output: Retrieved approximate weather using backup web search after primary API failed.
Enter fullscreen mode Exit fullscreen mode

Instead of crashing on the first exception, the agent inspects the failure context, updates its internal plan, and picks an alternate route to achieve the goal.

Setting up production systems like this usually takes fine-tuning, especially when dealing with multi-modal tools or strict rate limits. If you are building complex agent architecture and need extra engineering capacity, Gaper can match you with vetted developers to help build out your infrastructure fast.

Top comments (0)