Skip to main content

​
Overview

Every Nen workflow is a Python file with a run() function. 
def run(params: Params) -> Result:
    ...

​
Lifecycle

The workflow engine validates inputs, calls your run() function, and validates the result. If the run completes successfully, the workflow engine marks the run as completed.

​
Failing a Run

Raise any exception from run() to mark the run as failed. The exception message is sent to the error webhook and surfaced to the caller. 
def run(params: Params) -> Result:
    if not agent.verify("Is the dashboard loaded?", timeout=30):
        raise RuntimeError("Dashboard did not load within 30 seconds")
Use this for unrecoverable conditions — missing files, unreachable services, or failed preconditions. See the Error Handling guide for retry patterns and exception types.

​
Signature

run() supports two signatures: 
def run(params: Params) -> Result:
    ...
ParameterTypeDescription
paramsPydantic BaseModelValidated from JSON input
secure_paramsPydantic BaseModel with Secure[str] fieldsOptional. Secret references injected by the orchestrator
ReturnPydantic BaseModelValidated and serialized to JSON

​
Version Header

Every workflow file should declare the SDK version it targets: 
__nen_version__ = "1.0.0"

​
Examples

​
Simple workflow

workflow.py
from pydantic import BaseModel

class Params(BaseModel):
    name: str

class Result(BaseModel):
    greeting: str

def run(params: Params) -> Result:
    return Result(greeting=f"Hello, {params.name}!")

​
With field constraints

workflow.py
from pydantic import BaseModel, Field

class Params(BaseModel):
    url: str = Field(default="https://example.com", min_length=1)
    max_retries: int = Field(default=3, ge=1, le=10)

​
With secure params

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

class Params(BaseModel):
    url: str = Field(min_length=1)

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

class Result(BaseModel):
    success: bool

def run(params: Params, secure_params: SecureParams) -> Result:
    agent = Agent()
    computer = Computer()
    agent.execute(f"Open the browser to {params.url}")
    agent.execute("Click the password field")
    computer.type(secure_params.password)
    computer.press("Return")
    return Result(success=agent.verify("Is the user logged in?"))