> ## Documentation Index
> Fetch the complete documentation index at: https://notte-experiment-visibility-md-links.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Agent Lifecycle

> Understand agent execution modes, states, and control flow

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:

{/* @sniptest testers/agents/lifecycle/run_blocking.py */}

```python run_blocking.py theme={null}
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:

{/* @sniptest testers/agents/lifecycle/start_wait.py */}

```python start_wait.py theme={null}
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`:

{/* @sniptest testers/agents/async_execution.py */}

```python async_agent.py theme={null}
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:

```mermaid theme={null}
graph LR
    A[Created] --> B[Running]
    B --> C{Max Steps?}
    C -->|No| D{Task Complete?}
    C -->|Yes| E[Stopped: Max Steps]
    D -->|No| F{Error?}
    D -->|Yes| G[Completed: Success]
    F -->|Yes| H[Failed: Error]
    F -->|No| B
```

### Running

Agent is actively executing:

{/* @sniptest testers/agents/lifecycle/state_running.py */}

```python state_running.py theme={null}
agent = client.Agent(session=session)
agent.start(task="Complete task")
# Agent state: Running
```

### Completed

Agent successfully finished the task:

{/* @sniptest testers/agents/lifecycle/state_completed.py */}

```python state_completed.py theme={null}
from notte_sdk import NotteClient

client = NotteClient()

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

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

### Failed

Agent encountered an error:

{/* @sniptest testers/agents/lifecycle/state_failed.py */}

```python state_failed.py theme={null}
from notte_sdk import NotteClient

client = NotteClient()

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

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

### Stopped (Max Steps)

Agent reached maximum step limit:

{/* @sniptest testers/agents/lifecycle/state_max_steps.py */}

```python state_max_steps.py theme={null}
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")

    if len(result.steps) >= 5:
        # Agent hit max_steps limit
        print("Agent stopped: maximum steps reached")
```

## Monitoring Progress

### Status Checking

Check agent progress at any time:

{/* @sniptest testers/agents/lifecycle/status_checking.py */}

```python status_checking.py theme={null}
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: {len(status.steps)}")
print(f"Success: {status.success}")
```

### Live Log Streaming

Stream agent logs in real-time via WebSocket:

{/* @sniptest testers/agents/lifecycle/live_log_streaming.py */}

```python live_log_streaming.py theme={null}
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:

{/* @sniptest testers/agents/lifecycle/polling_pattern.py */}

```python polling_pattern.py theme={null}
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

        print(f"Progress: {len(status.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:

{/* @sniptest testers/agents/parallel_agents.py */}

```python parallel_agents.py theme={null}
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:

{/* @sniptest testers/agents/lifecycle/manual_stop.py */}

```python manual_stop.py theme={null}
agent = client.Agent(session=session)
agent.start(task="Long task")

# Do something...

# Stop the agent
agent.stop()
```

**Note:** You cannot stop agents once they complete a step - they must finish the current action.

### Timeout Pattern

Implement custom timeouts:

{/* @sniptest testers/agents/lifecycle/timeout_pattern.py */}

```python timeout_pattern.py theme={null}
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:

{/* @sniptest testers/agents/lifecycle/response_structure.py */}

```python response_structure.py theme={null}
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:

{/* @sniptest testers/agents/lifecycle/step_details.py */}

```python step_details.py theme={null}
agent = client.Agent(session=session)
result = agent.run(task="Navigate and extract")

for i, step in enumerate(result.steps):
    print(f"Step {i + 1}:")
    print(f"  Action: {step['action']}")
    print(f"  Success: {step['success']}")
    print(f"  Message: {step['message']}")
```

## Error Handling

### Graceful Degradation

Handle failures gracefully:

{/* @sniptest testers/agents/lifecycle/graceful_degradation.py */}

```python graceful_degradation.py theme={null}
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
```

### Retry Pattern

Retry failed agents:

{/* @sniptest testers/agents/retry_pattern.py */}

```python agent_retry.py theme={null}
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:

{/* @sniptest testers/agents/lifecycle/agent_fallback_example.py */}

```python agent_fallback_example.py theme={null}
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](/features/agents/fallback.md) for details.

## Best Practices

### 1. Use Appropriate Execution Mode

{/* @sniptest testers/agents/lifecycle/bp_execution_mode.py */}

```python bp_execution_mode.py theme={null}
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

{/* @sniptest testers/agents/lifecycle/bp_check_success.py */}

```python bp_check_success.py theme={null}
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

{/* @sniptest testers/agents/lifecycle/bp_step_limits.py */}

```python bp_step_limits.py theme={null}
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

{/* @sniptest testers/agents/lifecycle/bp_cleanup.py */}

```python bp_cleanup.py theme={null}
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

<CardGroup cols={2}>
  <Card title="Configuration" icon="sliders" href="/features/agents/configuration.md">
    Configure agent parameters
  </Card>

  <Card title="Agent Fallback" icon="shield" href="/features/agents/fallback.md">
    Automatic error recovery
  </Card>

  <Card title="Replay" icon="circle-play" href="/features/agents/replay.md">
    Debug with visual replays
  </Card>
</CardGroup>