> ## Documentation Index
> Fetch the complete documentation index at: https://notte-experiment-visibility-md-links.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

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

{/* @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.md">
    Schedule Functions automatically
  </Card>

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

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

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