> ## Documentation Index
> Fetch the complete documentation index at: https://docs.notte.cc/llms.txt
> Use this file to discover all available pages before exploring further.

# Function Invocations

> Call Functions via API, SDK, cURL, or webhooks

<Tip>
  STOP. Do not read past this section until you have read and followed [/quickstart.md](/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.
</Tip>

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:

{/* @sniptest testers/functions/invoke_function.py */}

```python invoke_function.py theme={null}
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

{/* @sniptest testers/functions/invocations/run_parameters.py */}

```python run_parameters.py theme={null}
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

{/* @sniptest testers/functions/invocations/handling_results.py */}

```python handling_results.py theme={null}
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

<CodeGroup>
  ```http HTTP theme={null}
  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
  }
  ```
</CodeGroup>

### Response

<CodeGroup>
  ```json JSON theme={null}
  {
    "function_id": "function_abc123",
    "function_run_id": "run_xyz789",
    "session_id": "session_456",
    "result": {
      "data": ["result1", "result2"],
      "count": 2
    },
    "status": "closed"
  }
  ```
</CodeGroup>

## Via cURL

### Basic cURL Request

<CodeGroup>
  ```bash cURL theme={null}
  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"
      }
    }'
  ```
</CodeGroup>

### Generate cURL Command

Get the exact cURL command for your Function:

<CodeGroup>
  ```python Python theme={null}
  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)
  ```
</CodeGroup>

Output:

<CodeGroup>
  ```bash Output theme={null}
  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"
      }
  }'
  ```
</CodeGroup>

## Via JavaScript/Node.js

Call from JavaScript applications:

<CodeGroup>
  ```javascript JavaScript theme={null}
  // 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);
  ```
</CodeGroup>

### With Axios

<CodeGroup>
  ```javascript JavaScript theme={null}
  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);
  ```
</CodeGroup>

## Streaming Logs

Watch function execution in real-time:

### Via SDK

{/* @sniptest testers/functions/invocations/stream_sdk.py */}

```python stream_sdk.py theme={null}
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)

<CodeGroup>
  ```bash cURL theme={null}
  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
  ```
</CodeGroup>

Response streams:

<CodeGroup>
  ```text Output theme={null}
  data: {"type": "log", "message": "Starting function..."}
  data: {"type": "log", "message": "Session created"}
  data: {"type": "log", "message": "Navigation complete"}
  data: {"type": "result", "message": "{\"function_id\": \"...\", ...}"}
  ```
</CodeGroup>

## Async Invocation

### Create and Start Separately

For long-running functions:

{/* @sniptest testers/functions/invocations/async_create_start.py */}

```python async_create_start.py theme={null}
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

{/* @sniptest testers/functions/invocations/check_run_status.py */}

```python check_run_status.py theme={null}
# 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

{/* @sniptest testers/functions/invocations/stop_running.py */}

```python stop_running.py theme={null}
# Stop a long-running function
client.functions.stop_run("function_abc123", run_id)
```

## Webhook Integration

### As Webhook Endpoint

Use Functions as webhook receivers:

{/* @sniptest testers/functions/invocations/webhook_handler.py */}

```python webhook_handler.py theme={null}
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:

<CodeGroup>
  ```bash cURL theme={null}
  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}
      }
    }'
  ```
</CodeGroup>

## Error Handling

### SDK Error Handling

{/* @sniptest testers/functions/invocations/sdk_error_handling.py */}

```python sdk_error_handling.py theme={null}
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

{/* @sniptest testers/functions/invocations/disable_raise.py */}

```python disable_raise.py theme={null}
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

<CodeGroup>
  ```python Python theme={null}
  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}")
  ```
</CodeGroup>

## Batch Invocations

### Run Multiple Functions

{/* @sniptest testers/functions/batch_invocation.py */}

```python batch_invocation.py theme={null}
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

{/* @sniptest testers/functions/invocations/sequential.py */}

```python sequential.py theme={null}
results = []
for url in urls:
    result = function.run(url=url)
    results.append(result.result)

print(results)
```

## Local vs Cloud Execution

### Cloud Execution (Default)

{/* @sniptest testers/functions/invocations/cloud_execution.py */}

```python cloud_execution.py theme={null}
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

{/* @sniptest testers/functions/invocations/local_execution.py */}

```python local_execution.py theme={null}
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

<CodeGroup>
  ```python Python theme={null}
  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"
  )
  ```
</CodeGroup>

## Best Practices

### 1. Use Appropriate Timeout

{/* @sniptest testers/functions/invocations/timeout_examples.py */}

```python timeout_examples.py theme={null}
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

{/* @sniptest testers/functions/invocations/stream_debug.py */}

```python stream_debug.py theme={null}
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

{/* @sniptest testers/functions/invocations/graceful_errors.py */}

```python graceful_errors.py theme={null}


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


try:
```

### 4. Use Variables for Dynamic Data

{/* @sniptest testers/functions/invocations/use_variables.py */}

```python use_variables.py theme={null}
# Good - parameterized
result = function.run(url=dynamic_url, query=user_input)

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

## Next Steps

<CardGroup cols={2}>
  <Card title="Schedules" icon="clock" href="/features/functions/schedules">
    Schedule Functions automatically
  </Card>

  <Card title="Management" icon="sliders" href="/features/functions/management">
    Update and monitor Functions
  </Card>

  <Card title="Creating Functions" icon="file-code" href="/features/functions/creating">
    Learn to write Functions
  </Card>

  <Card title="API Reference" icon="code" href="/api-reference">
    Full API documentation
  </Card>
</CardGroup>