Build a Durable AI Agent in 5 Minutes

AI agents are just workflows with LLM-powered decision steps. But most agent frameworks treat them as throwaway scripts — if the process crashes mid-execution, you start over. If a tool call fails, the whole chain breaks. If you want to debug what happened, you grep through logs.

There’s a better way.

The Problem with Ephemeral Agents

A typical AI agent makes a series of tool calls: search the web, read documents, call APIs, summarize findings. Each call depends on the previous result. If any step fails — network timeout, rate limit, API error — the entire agent run is lost.

Plan → Search → Search → Summarize → Store
  ✓       ✓       ✗ (timeout)
                   ↑
         Agent dies here. Restart from scratch.

Worse, when an agent produces a wrong answer, debugging is painful. What queries did it generate? What did each search return? What did the summarizer actually receive? Without step-by-step execution history, you’re guessing.

Durable Agents: Every Step Survives

Ironflow treats agent tool calls as durable steps — each one is memoized, retried on failure, and permanently recorded:

Plan → Search → Search → Summarize → Store
  ✓       ✓       ✗ (timeout)
                   ↑
         Ironflow retries the run automatically.
         Completed steps replay from the record — they are NOT re-executed.

If the process crashes entirely, the agent resumes from the last completed step. No wasted API calls, no lost context, no starting over.

How It Works

Here’s a research agent built with Ironflow. Each named step is durable.

TypeScript

import { createFunction } from "@ironflow/node";

const researchAgent = createFunction({
  id: "research-agent",
  triggers: [{ event: "agent.research" }],
  recording: true,  // Enable time-travel debugging
}, async ({ event, step }) => {
  const topic = (event.data as { topic: string }).topic;

  // Step 1: Generate search queries (durable — only runs once)
  const plan = await step.run("plan-research", async () => {
    return {
      queries: [
        `${topic} overview`,
        `${topic} best practices`,
        `${topic} common pitfalls`,
      ],
    };
  });

  // Step 2: Execute each search. The step name is half the memoization key,
  // so it must be stable across attempts — index the loop, don't interpolate
  // a model-generated query into the name.
  const results: unknown[] = [];
  for (const [i, query] of plan.queries.entries()) {
    const result = await step.run(`search-${i}`, async () => {
      // Your search API call here
      return await searchWeb(query);
    });
    results.push(result);
  }

  // Step 3: Summarize findings (durable)
  const summary = await step.run("summarize", async () => {
    return await summarizeResults(results);
  });

  return { topic, summary, sourcesUsed: results.length };
});

Go

type ResearchRequest struct {
  Topic string `json:"topic"`
}

type ResearchPlan struct {
  Queries []string `json:"queries"`
}

var ResearchAgent = ironflow.CreateFunction(
  ironflow.FunctionConfig{
    ID:        "research-agent",
    Triggers:  []ironflow.Trigger{{Event: "agent.research"}},
    Recording: true,
  },
  func(ctx ironflow.Context) (any, error) {
    var request ResearchRequest
    if err := ctx.Event.Data(&request); err != nil {
      return nil, err
    }

    plan, err := ironflow.Run(ctx, "plan-research", func() (ResearchPlan, error) {
      return ResearchPlan{Queries: []string{
        request.Topic + " overview",
        request.Topic + " best practices",
        request.Topic + " common pitfalls",
      }}, nil
    })
    if err != nil {
      return nil, err
    }

    results := make([]SearchResult, 0, len(plan.Queries))
    for i, query := range plan.Queries {
      result, err := ironflow.Run(ctx, fmt.Sprintf("search-%d", i), func() (SearchResult, error) {
        return searchWeb(query)
      })
      if err != nil {
        return nil, err
      }
      results = append(results, result)
    }

    summary, err := ironflow.Run(ctx, "summarize", func() (Summary, error) {
      return summarizeResults(results)
    })
    if err != nil {
      return nil, err
    }

    return map[string]any{
      "topic":       request.Topic,
      "summary":     summary,
      "sourcesUsed": len(results),
    }, nil
  },
)

Python

from ironflow.worker import function


@function(id="research-agent", triggers=[{"event": "agent.research"}], recording=True)
async def research_agent(ctx):
    topic = ctx.event.data["topic"]

    # Step 1: Generate search queries (durable — only runs once)
    plan = await ctx.step.run("plan-research", lambda: {
        "queries": [f"{topic} overview", f"{topic} best practices", f"{topic} common pitfalls"],
    })

    # Step 2: Execute each search. Index the loop for a stable step name.
    results = []
    for i, query in enumerate(plan["queries"]):
        results.append(await ctx.step.run(f"search-{i}", lambda q=query: search_web(q)))

    # Step 3: Summarize findings (durable)
    summary = await ctx.step.run("summarize", lambda: summarize_results(results))

    return {"topic": topic, "summary": summary, "sourcesUsed": len(results)}

Key properties:

• Memoized: If plan-research completes but search-1 fails, the plan step is NOT re-executed. Its output is replayed from the record.
• Retried: Retry is per run, not per step. Ironflow re-invokes the function (3 attempts by default), and memoization means only the incomplete steps actually run again. In TypeScript, use a retryable SDK error or construct an IronflowError with retryable: true; a bare Error is terminal. Go treats an ordinary error as retryable and uses ironflow.WrapNonRetryable for terminal failures. Python retries an ordinary exception and raises NonRetryableError for a terminal failure.
• Recorded: Every step’s input and output is permanently stored. Time-travel through any agent run.

Agent Memory as an Event Stream

Instead of stuffing conversation history into a database column, record agent activities as events.

TypeScript

import { createProjection } from "@ironflow/node";

const agentMemory = createProjection({
  name: "agent-memory",
  events: ["agent.research"],
  initialState: () => ({ tasks: 0, topics: [] as string[] }),
  handler: (
    state: { tasks: number; topics: string[] },
    event: { name: string; data: unknown },
  ) => ({
    tasks: state.tasks + 1,
    topics: [...state.topics, (event.data as { topic: string }).topic],
  }),
});

Go

var AgentMemory = ironflow.CreateProjection(ironflow.ProjectionConfig{
  Name:   "agent-memory",
  Events: []string{"agent.research"},
  InitialState: func() map[string]any {
    return map[string]any{"tasks": 0.0, "topics": []any{}}
  },
  Handler: func(
    state map[string]any,
    event ironflow.ProjectionEvent,
    _ ironflow.ProjectionContext,
  ) (map[string]any, error) {
    topics := append([]any{}, state["topics"].([]any)...)
    topics = append(topics, event.Data["topic"])
    return map[string]any{
      "tasks":  state["tasks"].(float64) + 1,
      "topics": topics,
    }, nil
  },
})

Python

from ironflow.projection import create_projection

agent_memory = create_projection(
    name="agent-memory",
    events=["agent.research"],
    initial_state=lambda: {"tasks": 0, "topics": []},
    handler=lambda state, event, ctx: {
        "tasks": state["tasks"] + 1,
        "topics": [*state["topics"], event.data["topic"]],
    },
)

The projection automatically derives the agent’s history from events. Query it anytime:

curl http://localhost:9123/api/v1/projections/agent-memory | jq '.state.state'

Time-Travel Debugging

This is where it gets powerful. When an agent produces unexpected results, you can scrub through the entire execution:

ironflow inspect <run-id>            # TUI: browse the run's steps
ironflow inspect <run-id> --replay   # Step through the run's events frame-by-frame

Up/down (or j/k) move through the agent’s steps. You see the exact output of every step — what the planner generated, what each search returned, what the summarizer received.

No log grepping. No guessing. Just step through the timeline.

Try It

brew install sahina/tap/ironflow

Choose a Tier-1 scaffold, replace its order-processing function with the research agent above, and start the worker.

TypeScript

ironflow init my-app && cd my-app
pnpm dev

Go

ironflow init my-app --template go-quickstart && cd my-app
go run main.go

Python

There is no Python scaffold yet. Put the research agent in worker.py with Worker(functions=[research_agent]).run() at the end, then:

pip install ironflow-py
python worker.py

ironflow serve --dev runs in the foreground, so start it in another terminal, then emit from a third:

ironflow serve --dev
ironflow emit agent.research --data '{"topic":"event sourcing best practices"}'

The AI agent example shows a complete TypeScript implementation, and the getting started tutorial walks through both scaffolds end to end.

Since this post was written, both Tier-1 SDKs gained a purpose-built agent API. TypeScript exports agent(), defineTool(), and agent memory from @ironflow/node/agent. Go provides agent.Agent, agent.DefineTool, and agent.Memory, and Python provides ironflow.agent (agent, define_tool, expose_mcp). All three use the same durable steps described here. The linked example uses the TypeScript API.

Your agents deserve the same durability guarantees as your production workflows.