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 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,  # 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

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:

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

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

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

# 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"},
    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

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

from notte_sdk import NotteClient

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

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

# Long task
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 });

// Short task
let result = await fn.run({ url: 'https://example.com' }, { timeoutMs: 60_000 });

// 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 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

​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

​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