Function Invocations

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.

Functions can be invoked through the Python or TypeScript SDK, HTTP API, cURL commands, or as webhook endpoints.

Via Python SDK

The simplest way to call a Function:

from notte_sdk import NotteClient

client = NotteClient()

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

# 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

import { NotteClient } from 'notte-sdk';

const client = new NotteClient();

// Get function by ID
const fn = client.NotteFunction({ function_id: functionId });

// Run function with parameters
const result = await fn.run({ url: 'https://example.com', search_query: 'laptop' });

console.log(result.result); // Access the return value
console.log(result.status); // "closed" or "failed"
console.log(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,  # Seconds without data from the server before giving up (default: 300)
    raise_on_failure=True,  # Raise exception on failure (default: True)
    local=False,  # Run locally vs cloud (default: False)
)

Handling Results

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"Function ID: {result.function_id}")
print(f"Run ID: {result.function_run_id}")
print(f"Session ID: {result.session_id}")

const result = await fn.run({ url: 'https://example.com' });

// Check status
if (result.status === 'closed') {
  console.log('Success!');
  console.log(result.result); // Function return value
} else if (result.status === 'failed') {
  console.log('Function failed');
  console.log(result.result); // Error message
}

// Access metadata
console.log(`Function ID: ${result.function_id}`);
console.log(`Run ID: ${result.function_run_id}`);
console.log(`Session ID: ${result.session_id}`);

Via HTTP API

Call Functions as HTTP endpoints. Send your API key both as Authorization: Bearer and in the x-notte-api-key header. Set "stream": false to get a single JSON response instead of a log stream:

POST Request

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

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

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 "x-notte-api-key: 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',
      'x-notte-api-key': 'YOUR_API_KEY',
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      function_id: 'function_abc123',
      variables: {
        url: 'https://example.com',
        search_query: 'laptop'
      },
      stream: false
    })
  }
);

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'
    },
    stream: false
  },
  {
    headers: {
      'Authorization': 'Bearer YOUR_API_KEY',
      'x-notte-api-key': 'YOUR_API_KEY',
      'Content-Type': 'application/json'
    }
  }
);

console.log(result.data);

Streaming Logs

Watch function execution in real-time:

Via SDK

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

// Stream logs while waiting for the final result.
const result = await fn.run({ url: 'https://example.com' }, { stream: true });

Via HTTP (Server-Sent Events)

curl -X POST https://api.notte.cc/functions/function_abc123/runs/start \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "x-notte-api-key: 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": "session_start", "message": "session_456"}
data: {"type": "log", "message": "Navigation complete"}
data: {"type": "result", "message": "{\"function_id\": \"...\", ...}"}

Async Invocation

Create and Start Separately

Use Python’s function.create_run() or TypeScript’s fn.createRun() only when you need a run ID before execution begins. It creates the run record only. Pass that ID to function.run(function_run_id=...) or fn.run(variables, { functionRunId }) to execute it. Omitting the ID automatically creates a new run. For Python local execution, pass local=True to both calls. Set NOTTE_FUNCTION_ID to your deployed function’s ID before running these examples:

export NOTTE_FUNCTION_ID=your_function_id

import os

from notte_sdk import NotteClient

client = NotteClient()
function = client.Function(os.environ["NOTTE_FUNCTION_ID"])
# Optional: create a run record when you need its ID before execution.
created = function.create_run()

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

# Execute that run and wait for its result.
result = function.run(function_run_id=run_id, url="https://example.com")
print(result.result)
run = function.get_run(run_id)
print(run.status)

import { NotteClient } from 'notte-sdk';

const client = new NotteClient();
const fn = client.NotteFunction({ function_id: process.env.NOTTE_FUNCTION_ID! });

// Optional: create a run record when you need its ID before execution.
const created = await fn.createRun();
console.log(`Run created: ${created.function_run_id}`);

// Execute that run and wait for its result. Omit functionRunId to create a new run.
const result = await fn.run(
  { url: 'https://example.com' },
  { functionRunId: created.function_run_id },
);
console.log(result.result);
const run = await fn.getRun(created.function_run_id);
console.log(run.status);

Run Without Blocking the Event Loop

function.run() is synchronous and waits for completion. stream=True (the default) prints live logs; stream=False receives the final JSON response without live logs. Disabling streaming does not make standard-runtime execution non-blocking. Use asyncio.to_thread() to keep your application’s event loop available:

async_run.py

import asyncio

from notte_sdk import NotteClient


async def main():
    function = NotteClient().Function("function_abc123")
    created = await asyncio.to_thread(function.create_run)
    pending = asyncio.create_task(
        asyncio.to_thread(
            function.run,
            function_run_id=created.function_run_id,
            url="https://example.com",
        )
    )

    # Other async work can run while execution is in progress.
    print(f"Run ID: {created.function_run_id}")
    result = await pending
    print(result.result)


asyncio.run(main())

Keep the application running until the task finishes. This runs the synchronous SDK call in a local thread; it does not create a detached server-side job.

Check Run Status

function.get_run(run_id) fetches the current status and result without waiting for execution to finish. From async code, use await asyncio.to_thread(function.get_run, run_id). In TypeScript, use await fn.getRun(runId); fn.retrieve(runId) remains a compatible alias.

run_status = function.get_run(run_id)

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

const runStatus = await fn.getRun(runId);

console.log(`Status: ${runStatus.status}`); // "active", "closed", "failed"
console.log(`Result: ${runStatus.result}`);

Stop a Running Function

stop_running.py

function.stop_run(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
  x-notte-api-key: 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 "x-notte-api-key: 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

# Inspect a failed run without raising an exception.
result = function.run(url="https://example.com", fail=True, raise_on_failure=False)
if result.status == "failed":
    print(f"Function failed: {result.result}")
else:
    print(f"Success: {result.result}")

// Inspect a failed run without throwing an exception.
const result = await fn.run(
  { url: 'https://example.com', fail: true },
  { raiseOnFailure: false },
);
if (result.status === 'failed') {
  console.log(`Function failed: ${result.result}`);
} else {
  console.log(`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", "x-notte-api-key": "YOUR_API_KEY"},
    json={"function_id": "function_abc123", "variables": {"url": "https://example.com"}, "stream": False},
)

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

Batch Invocations

Run Multiple Functions

from concurrent.futures import ThreadPoolExecutor

from notte_sdk import NotteClient

client = NotteClient()

function = client.Function(function_id=function_id)

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)

import { NotteClient } from 'notte-sdk';

const client = new NotteClient();

const fn = client.NotteFunction({ function_id: functionId });

const urls = ['https://site1.com', 'https://site2.com', 'https://site3.com'];

async function invokeFunction(url: string) {
  return fn.run({ url });
}

// Run in parallel
const results = await Promise.all(urls.map(invokeFunction));

for (const result of results) {
  console.log(result.result);
}

Sequential Invocations

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

print(results)

const results: unknown[] = [];
for (const url of urls) {
  const result = await fn.run({ url });
  results.push(result.result);
}

console.log(results);

Local vs Cloud Execution

Cloud Execution (Default)

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

// Run on Notte infrastructure (the default).
const result = await fn.run({ url: 'https://example.com' });

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 the function (Notte-managed functions also need decryption_key="...")
function = client.Function(function_id="func_abc123")

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

Advantages:

Debugging
Development/testing
No cloud execution costs

Requirements:

Access to the function’s code (you own it or it is shared with you)
Notte-managed functions also need their decryption key, passed as decryption_key when creating the Function instance

Rate Limits

Rate-Limited Runs

Running a published function that you do not own is rate limited per caller. When you exceed the limit, the API answers with HTTP 429.

Handling Rate Limits

Retry with backoff when a run is rejected with 429:

import time

import requests
from notte_sdk import NotteClient

client = NotteClient()

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


def invoke_with_retry(function, **variables):
    max_retries = 3
    for attempt in range(max_retries):
        try:
            return function.run(**variables)
        except requests.HTTPError as e:
            if e.response is not None and e.response.status_code == 429:
                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

In Python, timeout is how many seconds the SDK waits for data from the server before giving up (default 300); a long run that keeps streaming logs is not cut off. In TypeScript, timeoutMs is a deadline for the whole call (default 5 minutes) and throws NotteTimeoutError when it elapses. Both only control how long your code waits for the result.

from notte_sdk import NotteClient

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

# Stop waiting if the server sends nothing for 60 seconds (default: 300)
result = function.run(url="https://example.com", timeout=60)

# Tolerate longer quiet periods, e.g. long steps that log nothing
result = function.run(url="https://example.com", timeout=600)

import { NotteClient } from 'notte-sdk';

const client = new NotteClient();
const fn = client.NotteFunction({ function_id: functionId });

// Stop waiting after 60 seconds in total (throws NotteTimeoutError)
let result = await fn.run({ url: 'https://example.com' }, { timeoutMs: 60_000 });

// Allow up to 10 minutes for a long task
result = await fn.run({ url: 'https://example.com' }, { timeoutMs: 600_000 });

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 log output (still waits for the result)
result = function.run(url="https://example.com", stream=False)

3. Handle Errors Gracefully

graceful_errors.py

try:
    result = function.run(url="https://example.com")
    process_result(result.result)
except Exception as e:
    log_error(e)
    send_alert(str(e))

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

​Run Without Blocking the Event Loop

​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

​Rate-Limited Runs

​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

Schedules

Management

Creating Functions

API Reference

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

Run Without Blocking the Event Loop

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

Rate-Limited Runs

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