Automate multi-step workflows — Computer Use and Automation

Lesson updated November 3, 2026 per OpenAI’s Computer Use integration guidelines: the agent no longer handles login; command validation and branches that commit the user (reviews, new orders) require human confirmation; the orchestrator now checks the success_text_contains criterion it previously ignored, flags failed steps instead of skipping them, and never replays a step that creates data. Source: Confirmation and consent.

Orchestrating complex tasks

Real-world workflows rarely consist of a single action. Populating a CRM involves navigating between pages, finding a contact, updating their information, and then creating an opportunity—four chained operations, each of which can fail for its own reasons. Handing this entire block to an agent in a single prompt works sometimes, but you’ll never know where it broke down. This lesson teaches you to structure these multi-step automations so that every link is observable and replayable.

Step-based architecture

Start by breaking the workflow into atomic steps, each with a clear objective and a success criterion. This criterion is what distinguishes a to-do list from a true workflow: without it, your orchestrator can’t tell whether the agent actually succeeded or stopped on an error page. In the example below, opening the module is verified by the URL, and the search by the presence of the contact’s name on screen. Login isn’t included here: your code handled that beforehand, without exposing the password to the model (Lesson 8). The final step has "retry": False because it creates data: replaying it after a partial failure could create a duplicate.

workflow = [
    {
        "name": "open_contacts",
        "task": "Open the Contacts module in the CRM.",
        "success_url_contains": "/contacts",
    },
    {
        "name": "search_contact",
        "task": "Search for contact 'Jean Dupont' in the search bar.",
        "success_text_contains": "Jean Dupont",
    },
    {
        "name": "update_record",
        "task": "Open Jean Dupont's profile and update their phone number: 06 98 76 54 32.",
    },
    {
        "name": "create_opportunity",
        "task": "Create an opportunity: 'Cloud Migration Q2 2026', amount 50000 EUR, probability 60%.",
        "retry": False,  # non-replayable step: creates data
    },
]

The orchestrator

Now you need to execute this list. The orchestrator walks through the steps sequentially, wrapping each in a retry loop: a step that raises an exception or fails its success criterion is retried up to max_retries times, unless marked non-replayable. This detail matters in production, because many visual agent failures are transient—a slow page, an incomplete render at screenshot time. A step that fails definitively stops the workflow, since subsequent steps depend on it. Finally, the orchestrator returns a structured log indicating, for each step, its status and the number of attempts consumed: your first diagnostic tool.

def run_workflow(workflow: list, page, max_retries: int = 2):
    """Executes a multi-step workflow with retries."""
    results = []

    for step in workflow:
        step_name = step["name"]
        print(f"\n--- Step: {step_name} ---")
        attempts = max_retries + 1 if step.get("retry", True) else 1
        error = "success criterion not met"

        for attempt in range(attempts):
            try:
                run_agent(step["task"], page=page)

                # Optional checks: URL, then visible text
                url_check = step.get("success_url_contains")
                if url_check and url_check not in page.url:
                    print("  Expected URL not found")
                    continue
                text_check = step.get("success_text_contains")
                if text_check and text_check not in page.locator("body").inner_text():
                    print("  Expected text not found")
                    continue

                print(f"  Success (attempt {attempt + 1})")
                results.append({"step": step_name, "status": "success", "attempts": attempt + 1})
                break

            except Exception as e:
                print(f"  Error: {e}")
                error = str(e)
        else:
            # No attempt succeeded: subsequent steps depend on this one
            results.append({"step": step_name, "status": "failed", "error": error})
            break

    return results

Passing data between steps

Often, one step produces data needed by the next: the price read on a cart page must be verified at the checkout step; the case number created in step 2 serves as a reference in step 5. A shared context solves this need without turning your code into a tangle of global variables. The example below intentionally stops before payment: OpenAI classifies confirming a financial transaction as an action requiring user validation at the moment of execution. The class below stores key-value pairs and exposes an inject method that replaces {key} markers in a prompt template, letting you write your tasks as parameterized templates.

class WorkflowContext:
    """Stores data shared between steps."""

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

    def set(self, key: str, value):
        self.data[key] = value

    def get(self, key: str, default=None):
        return self.data.get(key, default)

    def inject(self, template: str) -> str:
        """Replaces {key} variables in the template."""
        task = template
        for key, value in self.data.items():
            task = task.replace("{" + key + "}", str(value))
        return task


# Usage example
ctx = WorkflowContext()

# Step 1: agent reads a price
extracted_price = run_agent(
    "Read the total price displayed and return only the number.",
    page=page
)
ctx.set("price", extracted_price.strip())

# Step 2: agent verifies, without validating anything
verdict = run_agent(ctx.inject(
    "The expected amount is {price} EUR. Compare it to the total on the payment page "
    "and answer only OK or MISMATCH. Do not validate anything."
), page=page)

# Step 3: a transaction is confirmed at the moment of action, by a human
if verdict.strip() == "OK":
    if input(f"Confirm order for {ctx.get('price')} EUR? [y/N] ").strip().lower() == "y":
        run_agent("Click the order confirmation button.", page=page)

Conditional workflows

Linear sequences quickly hit their limits, because the next step may depend on the previous result. The pattern to remember is this: first ask the agent to read a state by constraining it to respond with a value from a closed set, then branch in Python. For order tracking, a “processing” status leads to package tracking, “delivered” to retrieving the delivery date, “canceled” to retrieving the reason. Format constraints in the prompt are essential: without them, the agent returns a sentence your comparisons will regularly miss. All three branches remain read-only: posting a review or reordering would commit the user and require their confirmation.

def conditional_workflow(page):
    """Workflow with conditional branching."""
    # Step 1: Read status
    status = run_agent(
        "Read the order status. "
        "Return only: processing, delivered, or canceled.",
        page=page
    )

    # Branch based on result
    match status.strip().lower():
        case "processing":
            return run_agent("Click 'Track package' and note the displayed stage.", page=page)
        case "delivered":
            return run_agent("Note the displayed delivery date.", page=page)
        case "canceled":
            return run_agent("Note the displayed cancellation reason.", page=page)
        case _:
            raise ValueError(f"Unexpected status: {status!r}")

Parallelization

When tasks are independent, there’s no need to process them sequentially. Extracting data from three product sheets takes far less time if three agents work simultaneously, each in its own tab. Playwright’s async version and asyncio.gather are sufficient: one coroutine per URL, one new_page per coroutine, and systematic page closure to avoid accumulating tabs in memory. run_agent_async is the loop from Lesson 4 rewritten with the AsyncOpenAI client and Playwright’s async API: the same logic, with await before every call. Keep in mind that parallelism also multiplies your API call throughput, increasing the risk of hitting your rate limits (Lesson 11), and your load on the target site—three tabs go unnoticed, fifty much less so.

import asyncio
from playwright.async_api import async_playwright

async def parallel_extraction(urls: list[str]):
    """Extracts data from multiple pages in parallel."""
    async with async_playwright() as pw:
        browser = await pw.chromium.launch(headless=True)

        async def extract_one(url: str):
            page = await browser.new_page(
                viewport={"width": 1280, "height": 720}
            )
            try:
                await page.goto(url)
                result = await run_agent_async(
                    "Extract the product title and price.", page=page
                )
            finally:
                await page.close()
            return {"url": url, "data": result}

        results = await asyncio.gather(
            *[extract_one(url) for url in urls]
        )

    return results

# Run
urls = [
    "https://shop.example.com/product/1",
    "https://shop.example.com/product/2",
    "https://shop.example.com/product/3",
]
data = asyncio.run(parallel_extraction(urls))

Key takeaways

Break workflows into atomic steps with explicit success criteria
Automatically replay steps that can be retried safely, never those that create data
Use a shared context to pass data between steps
Conditional workflows adapt to page content; branches that commit the user require their confirmation
Parallelize independent tasks to save time