Function Management

Update, version, monitor, and manage your deployed Functions

Manage the lifecycle of your Functions including updates, versioning, monitoring executions, and viewing replays.

Viewing Functions

List All Functions

View all deployed Functions:

list_functions.py

from notte_sdk import NotteClient

client = NotteClient()

# List all functions
functions = client.functions.list()

for func in functions.workflows:
    print(f"ID: {func.function_id}")
    print(f"Name: {func.name}")
    print(f"Version: {func.latest_version}")
    print(f"Created: {func.created_at}")
    print("---")

Get Function Details

Retrieve specific Function information:

Python

# Get function by ID
function = client.Function(function_id="func_abc123")

# Access function properties
print(f"Function ID: {function.function_id}")
print(f"Name: {function.response.name}")
print(f"Description: {function.response.description}")
print(f"Latest Version: {function.response.latest_version}")
print(f"Versions: {function.response.versions}")

Updating Functions

Update Function Code

Deploy a new version of your Function:

Python

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

# Update with new code
function.update(
    workflow_path="updated_function.py"
)

print(f"Updated to version: {function.response.latest_version}")

Update Function Metadata

Change Function name or description:

# Update metadata (via console or API)
# Name and description updates are available through the Console

Note: Function updates create new versions automatically. Previous versions remain accessible for rollback.

Versioning

Version Management

Functions automatically version on each update:

Python

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

# Get all versions
print(f"Available versions: {function.response.versions}")

# Get latest version
print(f"Latest: {function.response.latest_version}")

Run Specific Version

Execute a specific Function version:

Python

# Run latest version (default)
result = function.run(
    url="https://example.com"
)

# Run specific version
result = function.run(
    url="https://example.com",
    version="v2"
)

Version History

Track changes across versions:

Python

workflows = client.functions.list()

for workflow in workflows.workflows:
    print(f"Function: {workflow.name}")
    print(f"Versions: {', '.join(workflow.versions)}")
    print(f"Latest: {workflow.latest_version}")

Monitoring Executions

View Run History

List recent Function executions:

Python

# Get recent runs
runs = client.functions.list_runs(
    function_id="function_abc123",
    only_active=False  # Include completed runs
)

for run in runs.runs:
    print(f"Run ID: {run.function_run_id}")
    print(f"Status: {run.status}")
    print(f"Created: {run.created_at}")
    print(f"Updated: {run.updated_at}")
    print("---")

Check Run Status

Monitor specific execution:

Python

# Get run details
run_status = client.functions.get_run(
    function_id="function_abc123",
    run_id="run_xyz789"
)

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

Status values:

• active - Function is currently running
• closed - Function completed successfully
• failed - Function encountered an error

Filter Active Runs

View only running executions:

Python

# Get only active runs
active_runs = client.functions.list_runs(
    function_id="function_abc123",
    only_active=True
)

print(f"Active runs: {len(active_runs.runs)}")

for run in active_runs.runs:
    print(f"Run {run.function_run_id} - {run.status}")

Viewing Replays

Access Function Replay

Watch MP4 replay of Function execution:

Python

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

# Run the function first
result = function.run(url="https://example.com")

# Get replay for this run (uses the session from the last run)
replay = function.replay()

# Save the replay
replay.save("function_replay.mp4")

Replay features:

• Video recording of browser session
• Shows all actions taken
• Useful for debugging failures
• Available after running a function

Replay After Run

Get replay immediately after execution:

Python

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

# Get replay for this run (automatically uses the session from the run)
replay = function.replay()

# Save the video
replay.save("execution_replay.mp4")

Managing Runs

Stop Running Function

Cancel a long-running execution:

Python

# Stop specific run
client.functions.stop_run(
    function_id="function_abc123",
    run_id="run_xyz789"
)

print("Function execution stopped")

Monitoring Long Runs

Track execution progress:

monitor_runs.py

import time
from notte_sdk import NotteClient

client = NotteClient()

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

# Start run
result = function.run(
    url="https://example.com",
    stream=False  # Don't stream logs
)

run_id = result.function_run_id

# Poll status
while True:
    status = client.functions.get_run(
        function_id="func_abc123",
        run_id=run_id
    )

    print(f"Status: {status.status}")

    if status.status in ["closed", "failed"]:
        print(f"Final result: {status.result}")
        break

    time.sleep(5)  # Check every 5 seconds

Forking Functions

Fork Shared Function

Create your own copy of a shared Function:

Python

# Fork a function from marketplace or teammate
original = client.Function(function_id="func_abc123")
forked_function = original.fork()

print(f"Forked function ID: {forked_function.function_id}")
print(f"Original ID: func_abc123")

Use cases:

• Customize shared Functions
• Create templates from marketplace
• Copy teammate’s Functions
• Experiment without affecting original

Modify Forked Function

After forking, update as needed:

Python

# Fork function
original = client.Function(function_id="shared_function_id")
forked = original.fork()

# Update your copy
forked.update(workflow_path="my_modified_version.py")

# Run your version
result = forked.run(url="https://example.com")

Downloading Functions

Download Function Code

Retrieve Function source code:

Python

function = client.Function(
    function_id="func_abc123",
    decryption_key="your-decryption-key"  # Required for downloading
)

# Download function code
code = function.download()

print(code)  # Function source code

# Or download directly to a file
code = function.download(workflow_path="downloaded_function.py")

Requirements:

• Decryption key (available in Console) - passed when creating Function instance
• Only works for Functions you own or have access to

Download for Backup

Back up all your Functions:

Python

import os

functions = client.functions.list()

# Create backup directory
os.makedirs("function_backups", exist_ok=True)

for func in functions.workflows:
    function = client.Function(
        function_id=func.function_id,
        decryption_key="your-key"
    )

    try:
        code = function.download()

        filename = f"function_backups/{func.name}_{func.latest_version}.py"
        with open(filename, "w") as f:
            f.write(code)

        print(f"Backed up: {func.name}")
    except Exception as e:
        print(f"Failed to backup {func.name}: {e}")

Deleting Functions

Delete Function

Remove a Function permanently:

Python

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

# Delete function
function.delete()

print("Function deleted")

Warning:

• Deletion is permanent
• All versions are deleted
• Run history is preserved
• Cannot be undone

Sharing Functions

Make Function Public

Share Functions with others:

Python

# Deploy as shared
function = client.Function(
    workflow_path="my_function.py",
    name="Public Scraper",
    shared=True  # Make publicly accessible
)

Shared Functions:

• ✅ Others can view and fork
• ✅ Original remains under your control
• ❌ Others cannot modify your version
• ✅ Usage tracked to your account

Private Functions

Keep Functions private (default):

Python

function = client.Function(
    workflow_path="my_function.py",
    name="Private Automation",
    shared=False  # Private (default)
)

Best Practices

1. Version Your Functions

Update incrementally with clear changes:

Python

# Before major update, download current version
current_code = function.download(decryption_key="key")

# Save backup
with open(f"backups/function_v{function.response.latest_version}.py", "w") as f:
    f.write(current_code)

# Update function
function.update(workflow_path="new_version.py")

2. Monitor Function Health

Regular health checks:

function_health_check.py

from notte_sdk import NotteClient

client = NotteClient()

def check_function_health(function_id: str):
    """Check recent run success rate."""
    runs = client.functions.list_runs(
        function_id=function_id,
        only_active=False
    )

    total = len(runs.runs)
    failed = sum(1 for r in runs.runs if r.status == "failed")

    success_rate = ((total - failed) / total * 100) if total > 0 else 0

    print(f"Success rate: {success_rate:.1f}%")
    print(f"Total runs: {total}")
    print(f"Failed runs: {failed}")

    return success_rate

# Monitor function
check_function_health("func_abc123")

3. Clean Up Old Runs

Archive or analyze old execution data:

Python

import json
from datetime import datetime, timedelta

def archive_old_runs(function_id: str, days_old: int = 30):
    """Archive runs older than specified days."""
    runs = client.functions.list_runs(
        function_id=function_id,
        only_active=False
    )

    cutoff_date = datetime.now() - timedelta(days=days_old)
    archive = []

    for run in runs.runs:
        run_date = datetime.fromisoformat(run.created_at.replace('Z', '+00:00'))

        if run_date < cutoff_date:
            archive.append({
                "run_id": run.function_run_id,
                "status": run.status,
                "created_at": run.created_at,
                "result": str(run.result)[:100]  # First 100 chars
            })

    # Save archive
    with open(f"archives/{function_id}_{datetime.now().date()}.json", "w") as f:
        json.dump(archive, f, indent=2)

    print(f"Archived {len(archive)} old runs")

archive_old_runs("function_abc123")

4. Document Changes

Keep change log for Function updates:

Python

# changelog.py
"""
Function: Data Scraper
Function ID: func_abc123

Version History:
- v1 (2024-01-15): Initial release
- v2 (2024-01-20): Added error handling
- v3 (2024-01-25): Improved scraping accuracy
- v4 (2024-02-01): Added retry logic
"""

def run(url: str):
    # Your function code
    pass

5. Test Before Production

Validate updates before deploying:

Python

# Test locally first - decryption key needed for local execution
function = client.Function(
    function_id="func_abc123",
    decryption_key="your-key"
)

# Run with local=True for testing
test_result = function.run(
    url="https://test-site.com",
    local=True
)

if test_result.status == "closed":
    print("Test passed, ready to update production")
    function.update(workflow_path="tested_function.py")
else:
    print(f"Test failed: {test_result.result}")

Troubleshooting

Function Fails to Run

Check run details for errors:

Python

run = client.functions.get_run(
    function_id="function_abc123",
    run_id="run_xyz789"
)

if run.status == "failed":
    print(f"Error: {run.result}")

    # Get replay for debugging
    replay = function.replay(function_run_id=run.function_run_id)
    print(f"Watch failure: {replay.url}")

Missing Environment Variables

Verify configuration:

Python

import os

def run():
    required_vars = ["API_KEY", "WEBHOOK_URL"]

    missing = [var for var in required_vars if not os.getenv(var)]

    if missing:
        return {
            "error": f"Missing environment variables: {', '.join(missing)}"
        }

    # Continue with automation

High Failure Rate

Analyze failure patterns:

Python

runs = client.functions.list_runs(function_id="function_abc123")

failures = [r for r in runs.runs if r.status == "failed"]

print(f"Failed runs: {len(failures)}/{len(runs.runs)}")

# Analyze failure reasons
for run in failures[:5]:  # Last 5 failures
    print(f"Run {run.function_run_id}:")
    print(f"  Error: {run.result}")
    print(f"  Time: {run.created_at}")

Next Steps

Creating Functions

Learn how to write Functions

Invocations

Call Functions via API or SDK

Schedules

Schedule Functions with cron

Functions Concept

Back to Functions overview