Function Invocations

Call Functions via API, SDK, cURL, or webhooks

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

HTTP

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

JSON

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

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:

Python

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:

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:

JavaScript

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

JavaScript

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

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:

Output

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

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

Python

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

Python

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