Function Invocations

AI agent instructions

STOP. Do not read past this section until you have read and followed /quickstart.md.Goal: produce a reliable SDK automation script. The intended authoring flow is generate then edit, not write browser automation from scratch.Do not hand-write SDK code with element IDs, selectors, or form fields from memory. Element IDs, selectors, and field mappings must come from a live observe() call, CLI notte page observe output, or generated workflow code. Guessed browser targets fail on real pages.Use the Notte skill and CLI to inspect the live site, validate actions/selectors, handle auth/session state, and export workflow code with notte sessions workflow-code before using SDK docs or SDK code.SDK reference pages are for understanding, running, or editing generated workflow code. They are not the starting point for manually authoring the initial browser automation. SDK-first code is guesswork on real-world pages with dynamic selectors, auth state, CAPTCHAs, and anti-bot behavior.

Functions can be invoked in multiple ways - through the Python SDK, HTTP API, cURL commands, or as webhook endpoints.

Via Python SDK

The simplest way to call a Function:

invoke_function.py

from notte_sdk import NotteClient

client = NotteClient()

# Get function by ID
function = client.Function(function_id="func_abc123")

# Run function with parameters
result = function.run(url="https://example.com", search_query="laptop")

print(result.result)  # Access the return value
print(result.status)  # "closed" or "failed"
print(result.session_id)  # Session ID if created

Run Parameters

run_parameters.py

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="workflow_abc123")
result = function.run(
    # Your function parameters (passed as variables)
    url="https://example.com",
    query="search term",
    # Execution options
    stream=True,  # Stream logs in real-time (default: True)
    timeout=300,  # Timeout in seconds (default: 300)
    raise_on_failure=True,  # Raise exception on failure (default: True)
    local=False,  # Run locally vs cloud (default: False)
)

Handling Results

handling_results.py

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="workflow_abc123")
result = function.run(url="https://example.com")

# Check status
if result.status == "closed":
    print("Success!")
    print(result.result)  # Function return value
elif result.status == "failed":
    print("Function failed")
    print(result.result)  # Error message

# Access metadata
print(f"Workflow ID: {result.workflow_id}")
print(f"Run ID: {result.workflow_run_id}")
print(f"Session ID: {result.session_id}")

Via HTTP API

Call Functions as HTTP endpoints:

POST Request

POST https://api.notte.cc/functions/{function_id}/runs/start
Authorization: Bearer YOUR_API_KEY
Content-Type: application/json

{
  "function_id": "function_abc123",
  "variables": {
    "url": "https://example.com",
    "query": "laptop"
  },
  "stream": true
}

Response

{
  "function_id": "function_abc123",
  "function_run_id": "run_xyz789",
  "session_id": "session_456",
  "result": {
    "data": ["result1", "result2"],
    "count": 2
  },
  "status": "closed"
}

Via cURL

Basic cURL Request

curl -X POST https://api.notte.cc/functions/function_abc123/runs/start \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "function_id": "function_abc123",
    "variables": {
      "url": "https://example.com",
      "search_query": "laptop"
    }
  }'

Generate cURL Command

Get the exact cURL command for your Function:

from notte_sdk import NotteClient

client = NotteClient()
function = client.Function(function_id="func_abc123")

# Generate cURL command
curl_command = function.get_curl(
    url="https://example.com",
    search_query="laptop"
)

print(curl_command)

Output:

curl --location 'https://api.notte.cc/functions/function_abc123/runs/start' \
--header 'x-notte-api-key: YOUR_API_KEY' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer YOUR_API_KEY' \
--data '{
    "function_id": "function_abc123",
    "variables": {
        "url": "https://example.com",
        "search_query": "laptop"
    }
}'

Via JavaScript/Node.js

Call from JavaScript applications:

// Using fetch
const response = await fetch(
  'https://api.notte.cc/functions/function_abc123/runs/start',
  {
    method: 'POST',
    headers: {
      'Authorization': 'Bearer YOUR_API_KEY',
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      function_id: 'function_abc123',
      variables: {
        url: 'https://example.com',
        search_query: 'laptop'
      }
    })
  }
);

const result = await response.json();
console.log(result);

With Axios

const axios = require('axios');

const result = await axios.post(
  'https://api.notte.cc/functions/function_abc123/runs/start',
  {
    function_id: 'function_abc123',
    variables: {
      url: 'https://example.com',
      search_query: 'laptop'
    }
  },
  {
    headers: {
      'Authorization': 'Bearer YOUR_API_KEY',
      'Content-Type': 'application/json'
    }
  }
);

console.log(result.data);

Streaming Logs

Watch function execution in real-time:

Via SDK

stream_sdk.py

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="workflow_abc123")
# Stream logs while running
result = function.run(
    url="https://example.com",
    stream=True,  # Logs printed to console
)

Via HTTP (Server-Sent Events)

curl -X POST https://api.notte.cc/functions/function_abc123/runs/start \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "function_id": "function_abc123",
    "variables": {"url": "https://example.com"},
    "stream": true
  }' \
  --no-buffer

Response streams:

data: {"type": "log", "message": "Starting function..."}
data: {"type": "log", "message": "Session created"}
data: {"type": "log", "message": "Navigation complete"}
data: {"type": "result", "message": "{\"function_id\": \"...\", ...}"}

Async Invocation

Create and Start Separately

For long-running functions:

async_create_start.py

from notte_sdk import NotteClient

client = NotteClient()
# Create a run (returns immediately)
run_response = client.functions.create_run("function_abc123", local=False)

run_id = run_response.workflow_run_id
print(f"Run created: {run_id}")

# Start the run
client.functions.run(run_id, function_id="function_abc123", variables={"url": "https://example.com"})

Check Run Status

check_run_status.py

# Check run status
run_status = client.functions.get_run("function_abc123", run_id)

print(f"Status: {run_status.status}")  # "active", "closed", "failed"
print(f"Result: {run_status.result}")

Stop a Running Function

stop_running.py

# Stop a long-running function
client.functions.stop_run("function_abc123", run_id)

Webhook Integration

As Webhook Endpoint

Use Functions as webhook receivers:

webhook_handler.py

def run(event_type: str, data: dict):
    """Handle webhook events."""
    if event_type == "order.created":
        # Process new order
        process_order(data)
    elif event_type == "user.signup":
        # Welcome new user
        send_welcome_email(data)

    return {"status": "processed", "event": event_type}

Configure webhook in external service:

Webhook URL: https://api.notte.cc/functions/function_abc123/runs/start
Method: POST
Headers: Authorization: Bearer YOUR_API_KEY

Testing Webhooks

Test webhook locally:

curl -X POST https://api.notte.cc/functions/function_abc123/runs/start \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "function_id": "function_abc123",
    "variables": {
      "event_type": "order.created",
      "data": {"order_id": "123", "amount": 99.99}
    }
  }'

Error Handling

SDK Error Handling

sdk_error_handling.py

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="workflow_abc123")

try:
    result = function.run(url="https://example.com")

    if result.status == "failed":
        print(f"Function failed: {result.result}")

except Exception as e:
    print(f"Unexpected error: {e}")

Disable Raise on Failure

disable_raise.py

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="workflow_abc123")

# Don't raise exception on failure
result = function.run(url="https://example.com", raise_on_failure=False)

# Check status manually
if result.status == "failed":
    print(f"Function failed: {result.result}")
else:
    print(f"Success: {result.result}")

HTTP Error Handling

import requests

response = requests.post(
    "https://api.notte.cc/functions/function_abc123/runs/start",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    json={
        "function_id": "function_abc123",
        "variables": {"url": "https://example.com"}
    }
)

if response.status_code == 200:
    result = response.json()
    print(result)
else:
    print(f"Error {response.status_code}: {response.text}")

Batch Invocations

Run Multiple Functions

batch_invocation.py

from concurrent.futures import ThreadPoolExecutor

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="func_abc123")

urls = ["https://site1.com", "https://site2.com", "https://site3.com"]

def invoke_function(url):
    return function.run(url=url)

# Run in parallel
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(invoke_function, urls))

for result in results:
    print(result.result)

Sequential Invocations

sequential.py

results = []
for url in urls:
    result = function.run(url=url)
    results.append(result.result)

print(results)

Local vs Cloud Execution

Cloud Execution (Default)

cloud_execution.py

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="workflow_abc123")
# Runs on Notte infrastructure
result = function.run(
    url="https://example.com",
    local=False,  # Default
)

Advantages:

Scalable
No local resources needed
Built-in logging
Session replays available

Local Execution

local_execution.py

from notte_sdk import NotteClient

client = NotteClient()

# Load function with decryption key for local execution
function = client.Function(
    function_id="func_abc123",
    decryption_key="your-key",  # Required for local execution
)

# Runs on your machine
result = function.run(url="https://example.com", local=True)

Advantages:

Debugging
Development/testing
No cloud execution costs

Requirements:

Decryption key (from Console) - passed when creating Function instance
Function code accessible locally

Rate Limits

Account Limits

Functions have rate limits based on your plan:

Free: 10 concurrent runs
Pro: 50 concurrent runs
Enterprise: Custom limits

Handling Rate Limits

import time
from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="workflow_abc123")


def invoke_with_retry(function, **variables):
    max_retries = 3
    for attempt in range(max_retries):
        try:
            return function.run(**variables)
        except Exception as e:
            if "rate limit" in str(e).lower():
                wait_time = 2 ** attempt  # Exponential backoff
                time.sleep(wait_time)
            else:
                raise
    raise RuntimeError("Max retries exceeded")

result = invoke_with_retry(
    function,
    url="https://example.com"
)

Best Practices

1. Use Appropriate Timeout

timeout_examples.py

from notte_sdk import NotteClient

client = NotteClient()
function = client.Function(function_id="function_abc123")

# Short task
result = function.run(url="https://example.com", timeout=60)

# Long task
result = function.run(url="https://example.com", timeout=600)

2. Stream Logs for Debugging

stream_debug.py

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id="function_abc123")

# Development - stream logs
result = function.run(url="https://example.com", stream=True)

# Production - no streaming
result = function.run(url="https://example.com", stream=False)

3. Handle Errors Gracefully

graceful_errors.py

def send_alert(message: str) -> None:
    print(message)


try:

4. Use Variables for Dynamic Data

use_variables.py

# Good - parameterized
result = function.run(url=dynamic_url, query=user_input)

# Bad - hardcoded
result = function.run()  # URL hardcoded in function

Next Steps

Schedules

Schedule Functions automatically

Management

Update and monitor Functions

Creating Functions

Learn to write Functions

API Reference

Full API documentation

​Via Python SDK

​Run Parameters

​Handling Results

​Via HTTP API

​POST Request

​Response

​Via cURL

​Basic cURL Request

​Generate cURL Command

​Via JavaScript/Node.js

​With Axios

​Streaming Logs

​Via SDK

​Via HTTP (Server-Sent Events)

​Async Invocation

​Create and Start Separately

​Check Run Status

​Stop a Running Function

​Webhook Integration

​As Webhook Endpoint

​Testing Webhooks

​Error Handling

​SDK Error Handling

​Disable Raise on Failure

​HTTP Error Handling

​Batch Invocations

​Run Multiple Functions

​Sequential Invocations

​Local vs Cloud Execution

​Cloud Execution (Default)

​Local Execution

​Rate Limits

​Account Limits

​Handling Rate Limits

​Best Practices

​1. Use Appropriate Timeout

​2. Stream Logs for Debugging

​3. Handle Errors Gracefully

​4. Use Variables for Dynamic Data

​Next Steps