Skip to main content

​
Overview

This example demonstrates a complete workflow that:
  1. Logs into a desktop application using secure credentials
  2. Searches for a customer by name
  3. Extracts structured customer data
  4. Downloads and organizes invoice files

​
Complete Code

workflow.py
from pathlib import Path
from nen import Agent, Computer, Secure
from pydantic import BaseModel, Field, EmailStr

class Params(BaseModel):
    user_name: str = Field(min_length=1)
    first_name: str = Field(min_length=1)
    last_name: str = Field(min_length=1)
    selected_customers: list[str]


class SecureParams(BaseModel):
    website_password: Secure[str] = Field(min_length=8)


class Result(BaseModel):
    customer_id: str = Field(min_length=8, max_length=16)
    first_name: str
    last_name: str
    phone: str | None = Field(default=None)
    email: EmailStr | None = Field(default=None)


def run(params: Params, secure_params: SecureParams) -> Result:
    agent = Agent()
    computer = Computer()

    # --- Step 1: Navigate and login ---
    agent.execute("Open the customer portal")

    if not agent.verify("Is the user logged in?"):
        agent.execute("Focus on the username field")
        computer.type(params.user_name)
        agent.execute("Focus on the password field")
        computer.type(secure_params.website_password)
        computer.press("Return")

    # --- Step 2: Search for customer ---
    agent.execute(
        f'Click customer search, enter first name "{params.first_name}" and '
        f'last name: "{params.last_name}", then click Search button'
    )

    # --- Step 3: Extract customer data ---
    customer_info = agent.extract(
        "Click the first result from the list and return customer info",
        Result.model_json_schema(),
    )

    # --- Step 4: Download paid invoices ---
    downloads = computer.drive("~/Downloads")
    paid_path = Path("./paid")
    paid_path.mkdir(parents=True, exist_ok=True)

    agent.execute("Select all paid invoices and download them")
    for f in downloads.files():
        if f.name in params.selected_customers:
            (paid_path / f.name).write_bytes(f.read_bytes())

    # --- Step 5: Save unpaid invoices to mount ---
    drive_mount = computer.drive("/mnt/tmp")
    agent.execute(f"Select all unpaid invoices and save to {drive_mount}")
    for f in drive_mount.files():
        with open(f.name, 'wb') as target:
            target.write(f.read_bytes())

    # --- Return structured result ---
    return Result.model_validate(customer_info)

​
Breakdown

​
Input Models

The workflow accepts structured parameters with validation: 
class Params(BaseModel):
    user_name: str = Field(min_length=1)       # Login username
    first_name: str = Field(min_length=1)      # Customer search
    last_name: str = Field(min_length=1)       # Customer search
    selected_customers: list[str]              # Filter for invoices
Secrets are kept separate in SecureParams: 
class SecureParams(BaseModel):
    website_password: Secure[str] = Field(min_length=8)

​
Secure Login

The password is typed via computer.type(), which sends a reference to the orchestrator. The real value never enters the desktop: 
computer.type(secure_params.website_password)  # Secure

​
File Handling

The workflow uses two file strategies:
Files copied to the workflow’s working directory end up in assets.zip:
downloads = computer.drive("~/Downloads")
for f in downloads.files():
    (Path("./paid") / f.name).write_bytes(f.read_bytes())

​
Returning Results

model_validate() runs the full Pydantic validation pipeline on the extracted dict and raises if any Field constraint is violated. This catches bad extractions (e.g. missing fields, wrong types, out-of-range values) before the result reaches your webhook, so prefer it even when the data looks trusted. 
return Result.model_validate(customer_info)

​
API Request

Because this workflow takes a secure_workflow_params, the request must be HMAC-signed — the simple x-api-key flow alone returns a 403. See Secure Parameters for the signing scheme; the trigger body itself looks like this: 
{
  "workflow_id": "customer-lookup",
  "workflow_params": {
    "user_name": "jane@company.com",
    "first_name": "John",
    "last_name": "Smith",
    "selected_customers": ["invoice_001.pdf", "invoice_002.pdf"]
  },
  "secure_workflow_params": {
    "website_password": "s3cure-p@ssword!"
  },
  "webhook_callback": "https://your-domain.com/webhook"
}