Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.notte.cc/llms.txt

Use this file to discover all available pages before exploring further.

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)

    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)

    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")

    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:
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: {len(status.steps)}")
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

        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:
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: You cannot stop agents once they complete a step - they must finish the current action.

​
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")

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:
graceful_degradation.py
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:
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