Autonomous Agent Workflows

Real-World Scenario

A research analyst asks an agent to “research our top 5 competitors and produce a SWOT analysis for each.” This requires 30+ tool calls, intermediate synthesis, parallel research, and structured output — far beyond a simple ReAct loop. An autonomous workflow agent decomposes the task, runs competitor research in parallel, synthesizes findings, and produces a formatted report — recovering gracefully if one research step fails.

Workflow Agent with Checkpointing

import anthropic
import json
import sqlite3
from dataclasses import dataclass, asdict, field
from datetime import datetime
from typing import Callable, Any
from pathlib import Path

client = anthropic.Anthropic()

@dataclass
class WorkflowStep:
    id:          str
    description: str
    status:      str  = "pending"   # pending, running, done, failed, skipped
    result:      Any  = None
    error:       str  = ""
    started_at:  str  = ""
    finished_at: str  = ""

@dataclass
class WorkflowState:
    workflow_id: str
    task:        str
    steps:       list[WorkflowStep] = field(default_factory=list)
    created_at:  str = ""
    final_result: str = ""

class WorkflowCheckpointer:
    """Persist workflow state to SQLite for resumability."""

    def __init__(self, db_path: str = "./workflow_state.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS workflows (
                id      TEXT PRIMARY KEY,
                state   TEXT NOT NULL,
                updated TEXT NOT NULL
            )
        """)
        self.conn.commit()

    def save(self, state: WorkflowState) -> None:
        now = datetime.now().isoformat()
        self.conn.execute(
            "INSERT OR REPLACE INTO workflows (id, state, updated) VALUES (?, ?, ?)",
            (state.workflow_id, json.dumps(asdict(state)), now)
        )
        self.conn.commit()

    def load(self, workflow_id: str) -> WorkflowState | None:
        row = self.conn.execute(
            "SELECT state FROM workflows WHERE id = ?", (workflow_id,)
        ).fetchone()
        if not row:
            return None
        data = json.loads(row[0])
        data["steps"] = [WorkflowStep(**s) for s in data["steps"]]
        return WorkflowState(**data)


class AutonomousWorkflowAgent:
    def __init__(
        self,
        tools: list[dict],
        tool_implementations: dict[str, Callable],
        system: str = "",
        checkpointer: WorkflowCheckpointer | None = None,
    ):
        self.tools          = tools
        self.impls          = tool_implementations
        self.system         = system
        self.checkpointer   = checkpointer or WorkflowCheckpointer()

    def _plan_steps(self, task: str) -> list[WorkflowStep]:
        """Use the LLM to decompose the task into concrete steps."""
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="You are a workflow planner. Decompose tasks into 3-8 concrete, sequential steps.",
            messages=[{
                "role": "user",
                "content": f"""Decompose this task into concrete steps:
{task}

Available tools: {[t['name'] for t in self.tools]}

Return JSON: {{"steps": [{{"id": "step_1", "description": "..."}}]}}"""
            }]
        )
        text = response.content[0].text
        try:
            start = text.index("{")
            data  = json.loads(text[start:text.rindex("}") + 1])
            return [WorkflowStep(id=s["id"], description=s["description"])
                    for s in data.get("steps", [])]
        except Exception:
            return [WorkflowStep(id="step_1", description=task)]

    def _execute_step(self, step: WorkflowStep, context: str) -> str:
        """Execute a single workflow step."""
        messages = [{
            "role": "user",
            "content": f"""Complete this step:
Step: {step.description}
Context from previous steps: {context[:2000] if context else "None"}

Use tools as needed. When done, provide a concise summary of what you did and found."""
        }]

        for _ in range(10):  # max iterations per step
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                system=self.system,
                tools=self.tools,
                messages=messages,
            )

            text_blocks = [b.text for b in response.content if hasattr(b, "text")]

            if response.stop_reason == "end_turn":
                return text_blocks[0] if text_blocks else "Step complete."

            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type != "tool_use":
                    continue
                impl   = self.impls.get(block.name)
                result = impl(block.input) if impl else f"Tool {block.name} not found"
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

            messages.append({"role": "user", "content": tool_results})

        return "Step timed out."

    def run(
        self,
        task: str,
        workflow_id: str | None = None,
        resume: bool = False,
    ) -> str:
        import uuid
        workflow_id = workflow_id or str(uuid.uuid4())[:8]

        # Try to resume from checkpoint
        state = self.checkpointer.load(workflow_id) if resume else None

        if state is None:
            print(f"[{workflow_id}] Planning steps...")
            steps = self._plan_steps(task)
            state = WorkflowState(
                workflow_id=workflow_id,
                task=task,
                steps=steps,
                created_at=datetime.now().isoformat(),
            )
            self.checkpointer.save(state)
            print(f"[{workflow_id}] {len(steps)} steps planned")
        else:
            print(f"[{workflow_id}] Resuming from checkpoint "
                  f"({sum(s.status == 'done' for s in state.steps)} steps already done)")

        # Execute each pending step
        context_parts = []
        for step in state.steps:
            if step.status == "done":
                context_parts.append(f"{step.description}: {step.result}")
                continue

            print(f"[{workflow_id}] Running: {step.description}")
            step.status     = "running"
            step.started_at = datetime.now().isoformat()
            self.checkpointer.save(state)

            try:
                result = self._execute_step(step, "\n".join(context_parts))
                step.status      = "done"
                step.result      = result
                step.finished_at = datetime.now().isoformat()
                context_parts.append(f"{step.description}: {result}")
                print(f"[{workflow_id}] Done: {result[:100]}...")
            except Exception as e:
                step.status = "failed"
                step.error  = str(e)
                print(f"[{workflow_id}] FAILED: {e}")

            self.checkpointer.save(state)

        # Synthesize final result
        print(f"[{workflow_id}] Synthesizing final result...")
        synthesis = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""Original task: {task}

Step results:
{chr(10).join(context_parts)}

Synthesize a comprehensive final answer."""
            }]
        )
        state.final_result = synthesis.content[0].text
        self.checkpointer.save(state)
        return state.final_result

Parallel Sub-Agent Execution

import anthropic
import concurrent.futures
import json
from dataclasses import dataclass

client = anthropic.Anthropic()

@dataclass
class SubAgentResult:
    agent_id: str
    task:     str
    result:   str
    success:  bool
    error:    str = ""

def run_sub_agent(agent_id: str, task: str, system: str, max_tokens: int = 1024) -> SubAgentResult:
    """Run a single sub-agent and return its result."""
    try:
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",  # cheaper model for sub-agents
            max_tokens=max_tokens,
            system=system,
            messages=[{"role": "user", "content": task}],
        )
        return SubAgentResult(
            agent_id=agent_id,
            task=task,
            result=response.content[0].text,
            success=True,
        )
    except Exception as e:
        return SubAgentResult(
            agent_id=agent_id, task=task, result="", success=False, error=str(e)
        )


def parallel_research_agent(
    main_task: str,
    subtasks: list[dict],  # [{"id": "...", "task": "...", "system": "..."}]
    max_workers: int = 5,
) -> str:
    """Run multiple research sub-agents in parallel, then synthesize."""
    print(f"Dispatching {len(subtasks)} parallel sub-agents...")

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(
                run_sub_agent,
                st["id"],
                st["task"],
                st.get("system", "You are a research assistant. Be concise and factual."),
            ): st["id"]
            for st in subtasks
        }

        results = []
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            status = "✓" if result.success else "✗"
            print(f"  {status} [{result.agent_id}] completed")
            results.append(result)

    # Sort by original order
    result_map = {r.agent_id: r for r in results}
    ordered    = [result_map[st["id"]] for st in subtasks if st["id"] in result_map]

    # Synthesize
    context = "\n\n".join(
        f"[{r.agent_id}] {r.task}\n{r.result if r.success else f'FAILED: {r.error}'}"
        for r in ordered
    )

    synthesis = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""Main task: {main_task}

Research findings from parallel agents:
{context}

Synthesize a comprehensive answer. Note any conflicting or missing information."""
        }]
    )
    return synthesis.content[0].text


# Example: analyze competing AI assistants in parallel
result = parallel_research_agent(
    main_task="Compare the top 3 cloud providers for ML workloads",
    subtasks=[
        {
            "id": "aws",
            "task": "Summarize AWS ML services: SageMaker, Bedrock, key strengths and weaknesses for production ML.",
        },
        {
            "id": "gcp",
            "task": "Summarize GCP ML services: Vertex AI, TPUs, key strengths and weaknesses for production ML.",
        },
        {
            "id": "azure",
            "task": "Summarize Azure ML services: Azure ML, OpenAI integration, key strengths and weaknesses.",
        },
    ],
)
print(result[:500])

Human-in-the-Loop with Async Approval

import anthropic
import json
from typing import Callable

client = anthropic.Anthropic()

RISKY_ACTIONS = {
    "delete_file":      "HIGH",
    "send_email":       "HIGH",
    "update_database":  "MEDIUM",
    "create_file":      "LOW",
    "read_file":        "LOW",
}

def requires_approval(tool_name: str, risk_level: str = "MEDIUM") -> bool:
    action_risk = RISKY_ACTIONS.get(tool_name, "LOW")
    levels = {"LOW": 1, "MEDIUM": 2, "HIGH": 3}
    return levels.get(action_risk, 0) >= levels.get(risk_level, 2)


def get_human_approval(tool_name: str, inputs: dict) -> bool:
    """Prompt for human approval (sync version — in production use async/queue)."""
    print(f"\n⚠️  APPROVAL REQUIRED")
    print(f"   Action: {tool_name}")
    print(f"   Inputs: {json.dumps(inputs, indent=4)}")
    response = input("   Approve? (y/n): ").strip().lower()
    return response == "y"


def human_in_loop_agent(
    task: str,
    tools: list[dict],
    tool_implementations: dict[str, Callable],
    approval_threshold: str = "MEDIUM",
    auto_approve_callback: Callable | None = None,
) -> str:
    messages = [{"role": "user", "content": task}]

    for _ in range(20):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return next((b.text for b in response.content if hasattr(b, "text")), "Done.")

        messages.append({"role": "assistant", "content": response.content})
        tool_results = []

        for block in response.content:
            if block.type != "tool_use":
                continue

            # Check if this action needs approval
            if requires_approval(block.name, approval_threshold):
                if auto_approve_callback:
                    approved = auto_approve_callback(block.name, block.input)
                else:
                    approved = get_human_approval(block.name, block.input)

                if not approved:
                    result = f"Action '{block.name}' was rejected by the user. Please suggest an alternative approach."
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
                    continue

            impl   = tool_implementations.get(block.name)
            result = impl(block.input) if impl else "Tool not found"
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result,
            })

        messages.append({"role": "user", "content": tool_results})

    return "Max iterations reached."