Agent Lifecycle

Understand agent execution modes, states, and control flow

AI agent instructions

Read the Notte skill and follow its setup instructions when setting up Notte or building browser automation. This is the shared source for CLI setup, authentication, and the browser-to-code workflow.

Agents support multiple execution patterns - from simple blocking calls to advanced asynchronous monitoring. Choose the right pattern for your use case.

Execution Modes

Run (Blocking)

The simplest way to execute an agent - start and wait for completion:

run_blocking.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session, max_steps=10)

    result = agent.run(task="Go to example.com and find the contact email")

    if result.success:
        print(result.answer)
    else:
        print(f"Failed: {result.answer}")

When to use:

• Simple scripts
• Synchronous workflows
• You don’t need to do other work while the agent runs

---

Start + Wait (Non-blocking)

Start the agent and wait for completion separately:

start_wait.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session)

    # Start agent (returns immediately)
    agent.start(task="Complete this task")

    # Do other work here...

    # Wait for completion
    result = agent.wait()

When to use:

• You need to start multiple agents in parallel
• You want to do other work while the agent runs
• You need more control over execution

---

Async Execution

Run agents asynchronously with async/await:

async_agent.py

import asyncio

from notte_sdk import NotteClient

client = NotteClient()


async def run_agent_task():
    with client.Session() as session:
        agent = client.Agent(session=session)
        agent.start(task="Extract data from the page")
        result = await agent.async_watch_logs_and_wait()
        return result


# Run async
result = asyncio.run(run_agent_task())
print(result.answer)

When to use:

• Building async applications
• Running multiple agents concurrently
• Integrating with async frameworks (FastAPI, aiohttp)

Agent States

Agents transition through these states during execution:

Running

Agent is actively executing:

state_running.py

agent = client.Agent(session=session)
agent.start(task="Complete task")
# Agent state: Running

Completed

Agent successfully finished the task:

state_completed.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session)
    agent.start(task="Find the contact email on example.com")

    result = agent.wait()
    if result.success:
        # Agent state: Completed
        print(result.answer)

Failed

Agent encountered an error:

state_failed.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session)
    agent.start(task="Find the contact email on example.com")

    result = agent.wait()
    if not result.success:
        # Agent state: Failed
        print(f"Error: {result.answer}")

Stopped (Max Steps)

Agent reached maximum step limit:

state_max_steps.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session, max_steps=5)
    result = agent.run(task="Very complex task")

    # Count agent steps (one "agent_step_start" element per step)
    nb_steps = sum(1 for s in result.steps if s["type"] == "agent_step_start")
    if not result.success and nb_steps >= 5:
        # Agent hit max_steps limit
        print("Agent stopped: maximum steps reached")

Monitoring Progress

Status Checking

Check agent progress at any time:

status_checking.py

agent = client.Agent(session=session)
agent.start(task="Long running task")

# Check status
status = agent.status()

print(f"Agent ID: {status.agent_id}")
print(f"Current state: {status.status}")
print(f"Steps completed: {sum(1 for s in status.steps if s['type'] == 'agent_step_stop')}")
print(f"Success: {status.success}")

Live Log Streaming

Stream agent logs in real-time via WebSocket:

live_log_streaming.py

def monitor_agent():
    with client.Session() as session:
        agent = client.Agent(session=session)
        agent.start(task="Complete task")

        # Stream logs and get final status
        status = agent.watch_logs(log=True)
        if status is None:
            status = agent.status()
        return status

Polling Pattern

Check status periodically:

polling_pattern.py

import time

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session)
    agent.start(task="Long task")

    while True:
        status = agent.status()

        if status.status == "closed":
            break

        nb_steps = sum(1 for s in status.steps if s["type"] == "agent_step_stop")
        print(f"Progress: {nb_steps} steps completed")
        time.sleep(5)  # Check every 5 seconds

    print(f"Final result: {status.answer}")

Parallel Execution

Multiple Independent Agents

Run multiple agents simultaneously:

parallel_agents.py

import asyncio

from notte_sdk import NotteClient

client = NotteClient()


async def run_multiple_agents():
    async def run_one(task_description: str):
        with client.Session() as session:
            agent = client.Agent(session=session)
            agent.start(task=task_description)
            return await agent.async_watch_logs_and_wait()

    # Run all agents in parallel (each with its own session)
    results = await asyncio.gather(*[run_one(t) for t in ["Task 1", "Task 2", "Task 3"]])
    return results


results = asyncio.run(run_multiple_agents())
for i, result in enumerate(results):
    print(f"Agent {i + 1}: {result.answer}")

Stopping Agents

Manual Stop

Stop a running agent:

manual_stop.py

agent = client.Agent(session=session)
agent.start(task="Long task")

# Do something...

# Stop the agent
agent.stop()

Note: stop() does not interrupt an action that is already executing - the agent stops once the current action finishes.

Timeout Pattern

Implement custom timeouts:

timeout_pattern.py

import asyncio

from notte_sdk import NotteClient


async def run_with_timeout(agent, timeout_seconds=60):
    try:
        async with asyncio.timeout(timeout_seconds):
            agent.start(task="Complete task")
            result = await agent.async_watch_logs_and_wait()
        return result
    except TimeoutError as e:
        # agent.stop() is already called internally by async_watch_logs_and_wait on cancellation
        raise TimeoutError(f"Agent exceeded {timeout_seconds}s timeout") from e


client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session)
    # Run with 60 second timeout
    result = asyncio.run(run_with_timeout(agent, timeout_seconds=60))

Response Structure

Agent responses contain execution details:

response_structure.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session)
    result = agent.run(task="Extract data")

    # Access result properties
    print(result.success)  # bool: Did agent succeed?
    print(result.answer)  # str: Agent's response
    print(result.steps)  # list: All steps taken
    print(result.agent_id)  # str: Unique agent ID
    print(result.session_id)  # str: Session used

Step Details

Inspect individual steps:

step_details.py

agent = client.Agent(session=session)
result = agent.run(task="Navigate and extract")

# Each entry is a trajectory element: {"type": ..., "value": {...}}
# Executed actions have type "execution_result"
actions = [s["value"] for s in result.steps if s["type"] == "execution_result"]
for i, step in enumerate(actions):
    print(f"Action {i + 1}: {step['action']['type']}")
    print(f"  Success: {step['success']}")
    print(f"  Message: {step['message']}")

Error Handling

Graceful Degradation

Handle failures gracefully:

graceful_degradation.py

from loguru import logger
from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    agent = client.Agent(session=session)
    result = agent.run(task="Complete task")

    if result.success:
        # Process successful result
        print(result.answer)
    else:
        # Handle failure
        logger.error(f"Agent failed: {result.answer}")

        # Fallback strategy
        print("Executing fallback approach...")

Retry Pattern

Retry failed agents:

agent_retry.py

import time

from notte_sdk import NotteClient

client = NotteClient()

MAX_RETRIES = 3

for attempt in range(MAX_RETRIES):
    with client.Session() as session:
        agent = client.Agent(session=session)
        result = agent.run(task="Complete task")

        if result.success:
            print(f"Success: {result.answer}")
            break

        print(f"Attempt {attempt + 1} failed, retrying...")
        time.sleep(2**attempt)  # Exponential backoff
else:
    raise RuntimeError("Agent failed after all retries")

Agent Fallback

Use AgentFallback for automatic error recovery:

agent_fallback_example.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    with client.AgentFallback(session, task="Add item to cart") as agent_fb:
        # Try deterministic actions first
        session.execute(type="click", selector="#add-to-cart")
        session.execute(type="click", selector="#checkout")
        # Agent automatically handles any failures

if agent_fb.success:
    print("Task completed (possibly with agent help)")

See Agent Fallback for details.

Best Practices

1. Use Appropriate Execution Mode

bp_execution_mode.py

import asyncio

from notte_sdk import NotteClient

client = NotteClient()


async def main():
    with client.Session() as session:
        agent = client.Agent(session=session)

        # Simple tasks: Use run()
        result = agent.run(task="Quick task")

        # Multiple agents: Use async (each with its own session)
        async def run_one(task: str):
            with client.Session() as s:
                a = client.Agent(session=s)
                a.start(task=task)
                return await a.async_watch_logs_and_wait()

        results = await asyncio.gather(run_one("Task 1"), run_one("Task 2"))

        # Long tasks with monitoring: Use start() + polling
        agent.start(task="Long task")
        done = False
        while not done:
            status = agent.status()
            # Update UI, log progress, etc.
            done = True  # placeholder


asyncio.run(main())

2. Always Check Success

bp_check_success.py

agent = client.Agent(session=session)

result = agent.run(task="Critical task")

if not result.success:
    # Don't proceed if agent failed
    raise RuntimeError(f"Critical task failed: {result.answer}")

# Safe to proceed with result
print(result.answer)

3. Set Appropriate Step Limits

bp_step_limits.py

from notte_sdk import NotteClient

client = NotteClient()

with client.Session() as session:
    # Match max_steps to expected complexity
    agent = client.Agent(
        session=session,
        max_steps=10,  # For tasks requiring 5-8 steps
    )

4. Clean Up Resources

bp_cleanup.py

from notte_sdk import NotteClient

client = NotteClient()

# Use context managers for automatic cleanup
with client.Session() as session:
    agent = client.Agent(session=session)
    result = agent.run(task="Task")
# Session automatically closed

Next Steps

Configuration

Configure agent parameters

Agent Fallback

Automatic error recovery

Replay

Debug with visual replays