Function Invocations

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

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

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

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:

from notte_sdk import NotteClient

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

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

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

Check Run Status

from notte_sdk import NotteClient

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

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

Stop a Running Function

from notte_sdk import NotteClient

client = NotteClient()

# Stop a long-running function
client.functions.stop_run(
    function_id="function_abc123",
    run_id=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

from notte_sdk import NotteClient

client = NotteClient()

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

from notte_sdk.endpoints.functions import FailedToRunCloudFunctionError

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

except FailedToRunCloudFunctionError as e:
    print(f"Function failed: {e.message}")
    print(f"Run ID: {e.function_run_id}")
    print(f"Result: {e.response.result}")

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

Disable Raise on Failure

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

from notte_sdk import NotteClient

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

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

print(results)

Local vs Cloud Execution

Cloud Execution (Default)

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

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

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

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

4. Use Variables for Dynamic Data

# 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

Getting Started

Sessions

Agents

Functions

Agent Tools

Scraping

Guides

Integrations

​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