➡️ Sequential workflow: one step at a time

Some jobs only make sense in order. You can't analyse data you haven't collected, or review a report that hasn't been written. A sequential workflow splits a big task into a chain of smaller steps, and each step uses the output of the step before it.

What kind of thing is this?

Sequential processing is a general design pattern, not a Claude API setting. You build it with ordinary code that calls Claude several times. Anthropic calls it prompt chaining in Building effective agents.

What is it?

In a sequential workflow you break a task into fixed steps that run in a set order. Step 1's output becomes step 2's input, and so on, like passing a baton in a relay race. Each step is a small, focused job, often a single call to Claude with its own instructions.

Step
One focused piece of work, such as "list the key facts".
Dependency
When one step needs another step's result before it can start.
Gate (check)
A test between steps that checks the output is good enough to continue.
Checkpoint
Saved progress, so a failed run can restart from the last good step instead of from the beginning.
Retry
Trying a failed step again, usually a limited number of times.

Why is it important?

Real-world analogy

Preparing a report at work

First you collect the data, then you analyse it to find what matters, then you write a draft, and finally a colleague reviews it before it goes to your manager. Doing these out of order makes no sense: you can't review a draft that doesn't exist. If the review finds problems, the draft goes back for fixes, but you don't start the research again.

How does it work?

  1. Split the task into steps that each have one clear goal.
  2. Write instructions for each step.
  3. Run step 1 with the original input.
  4. Check the output (the gate). If it fails, retry the step or stop.
  5. Save a checkpoint and pass the output to the next step.
  6. Repeat until the last step produces the final result.

Workflow diagram

Simple overview

Each arrow carries the previous step's result forward. If the review finds problems, only the draft step runs again.

Detailed: what happens at every step

If a step still fails after the allowed retries, the workflow stops and asks a person to look, rather than passing bad work along.

Try it: run the pipeline

Interactive demoA simulation in your browser. No AI is called.

    Practical example: research-and-report agent

    A manager pastes in a page of messy market notes and wants a clean one-page report.

    StepReceivesInstructionCheck before moving on
    1. ResearchRaw notes"List the 5 most important facts, using only the notes."At least 3 bullet points
    2. AnalysisFact list"Identify 2 trends and 1 risk."Contains TREND and RISK
    3. DraftAnalysis"Write a 200-word report for a manager."Long enough
    4. ReviewDraft"Fix unclear or unsupported sentences; return the final report."Long enough

    In this example, "research" means pulling facts out of the notes you supply. To research the live web, you would give the first step a web search tool. See Tool.

    Code example

    Runnable Python. It needs pip install anthropic, the ANTHROPIC_API_KEY environment variable on your computer or server, and a notes.txt file.

    
    # sequential_report.py: research, then analysis, then draft, then review.
    import json
    import pathlib
    import anthropic
    
    client = anthropic.Anthropic(max_retries=3)  # the SDK retries network and rate-limit errors for us
    MODEL = "claude-opus-5-5"
    CHECKPOINT = pathlib.Path("checkpoint.json")
    
    def ask(instructions: str, material: str) -> str:
        """One focused Claude call: this step's instructions + the previous step's output."""
        response = client.messages.create(
            model=MODEL,
            max_tokens=16000,
            messages=[{"role": "user",
                       "content": f"{instructions}\n\n<material>\n{material}\n</material>"}],
        )
        return "".join(b.text for b in response.content if b.type == "text")
    
    # Each step: (name, instructions, a simple quality check for its output)
    STEPS = [
        ("research", "List the 5 most important facts in this material as bullet points. Use only the material.",
         lambda out: len(out.strip().splitlines()) >= 3),
        ("analysis", "Using these facts, identify 2 trends and 1 risk. Label them TREND and RISK.",
         lambda out: "TREND" in out and "RISK" in out),
        ("draft", "Write a 200-word report for a manager from this analysis. Start with a one-sentence summary.",
         lambda out: len(out.split()) > 80),
        ("review", "Improve unclear sentences and remove unsupported claims. Return only the final report.",
         lambda out: len(out.split()) > 80),
    ]
    
    def run_pipeline(source_text: str) -> str:
        # If an earlier run stopped part-way, pick up where it left off.
        state = json.loads(CHECKPOINT.read_text()) if CHECKPOINT.exists() else {"done": []}
        material = state.get("last_output", source_text)
    
        for name, instructions, check in STEPS:
            if name in state["done"]:
                print(f"[done] {name} (skipped: already done, loaded from checkpoint)")
                continue
            for attempt in (1, 2):                     # each step gets at most two tries
                output = ask(instructions, material)
                if check(output):
                    break
                print(f"[retry] {name}: output failed its check (attempt {attempt})")
            else:                                       # runs only if both tries failed
                raise RuntimeError(f"Step '{name}' failed twice. Stopping so a person can look.")
    
            material = output                           # the baton passes to the next step
            state = {"done": state["done"] + [name], "last_output": material}
            CHECKPOINT.write_text(json.dumps(state))    # save progress after every step
            print(f"[done] {name}")
    
        CHECKPOINT.unlink()                             # finished, so clear the checkpoint
        return material
    
    if __name__ == "__main__":
        print(run_pipeline(pathlib.Path("notes.txt").read_text()))
    

    Expected output (wording varies):

    
    [done] research
    [retry] analysis: output failed its check (attempt 1)
    [done] analysis
    [done] draft
    [done] review
    Summary: E-scooter demand grew strongly this year, but battery costs are a rising risk.
    ...
    

    Step-by-step explanation

    1. anthropic.Anthropic(max_retries=3): the official SDK automatically retries temporary problems such as network errors and rate limits.
    2. ask() makes one focused call. The previous result is wrapped in <material> tags so Claude can clearly tell the instructions from the data.
    3. STEPS lists the chain in order. Each entry has a name, instructions and a small check function (a gate).
    4. At the start, run_pipeline loads the checkpoint if one exists, so finished steps are skipped.
    5. The inner for attempt in (1, 2) loop is the retry: if the check fails, the step runs once more.
    6. The for … else block runs only when every attempt failed. The pipeline then stops on purpose instead of passing bad work forward.
    7. material = output is the most important line: it's how one step's result passes to the next.
    8. After each step, progress is written to checkpoint.json. When everything is done, the checkpoint is deleted.

    Passing results between steps

    Error handling, retries and checkpoints

    🔁 Retries

    Use the SDK's built-in retries for network problems. Add your own limited retry for "the output didn't pass the check", but never retry forever.

    🚦 Gates

    Simple code checks such as length, required words or valid JSON are cheap and catch most problems early.

    💾 Checkpoints

    Save after each step so a crash at step 4 doesn't mean paying for steps 1–3 again.

    🛑 Fail safely

    When a step keeps failing, stop and tell a person. See Human in the Loop.

    Use cases: when to use it, and when not to

    👍 Good fit

    • Research → analyse → write → review
    • Translate → then adapt the tone for a market
    • Write an outline → check it → write the full text
    • Extract data → validate → save

    👎 Poor fit

    • Steps that don't depend on each other: use Parallel to save time
    • Requests that need different handling: use Route
    • Open-ended problems where you can't list the steps in advance: use an agent loop or Orchestrate

    Advantages and limitations

    ✅ Advantages

    • Simple to build and understand
    • Higher quality from focused steps
    • Easy to test each step on its own

    ⚠️ Limitations

    • Slower: each step waits for the one before
    • More API calls means more cost
    • An early mistake can carry through every later step

    🚫 Common mistakes

    • No checks between steps
    • Passing everything to every step
    • Unlimited retries
    • Making steps sequential that could run in parallel

    Best practices

    Knowledge check

    1. What makes a workflow "sequential"?

    2. What is a "gate" between steps?

    3. Why save a checkpoint after each step?

    4. You need summaries of three unrelated documents. Is sequential the best choice?

    5. A step fails its check twice. What should a well-built workflow do?

    Mini project

    Build a three-step "blog post" chain.

    1. Step 1 Outline: "Create a 5-point outline about {topic}." Check: at least 5 lines.
    2. Step 2 Write: "Write a 300-word post from this outline." Check: more than 200 words.
    3. Step 3 Title: "Suggest 3 titles for this post." Check: exactly 3 lines.
    4. Reuse run_pipeline from the example by changing only the STEPS list.
    5. Bonus: stop the program after step 2, run it again, and confirm it resumes from the checkpoint.

    Key takeaways

    Further learning