Custom Python Functions & Sandbox Execution

The Valstorm Function System allows developers to write, test, and execute asynchronous Python micro-functions securely within the platform. Functions serve as modular building blocks for complex backend workflows, asynchronous batch tasks, data reporting, AI processing, and third-party integrations.

1. Architecture & Sandboxed Execution

Every function runs inside an isolated, safe module proxy (FunctionProxy) managed by the FunctionManager. During execution:

Asynchronous Execution: Functions run as asynchronous coroutines via safe_execute_async.
Platform Context Injection: The runtime automatically injects an authenticated PlatformContext instance as the first positional parameter (platform: PlatformContext).
User & Tenant Scoping: All operations executed through platform automatically inherit the tenant database isolation and permission context of the invoking user (platform.user).
Safe Logging: Developers use platform.log(message, level="info"|"warning"|"error"|"debug") to output centralized execution logs.

Available Runtime Libraries & Tooling

Functions run in a rich Python environment pre-loaded with:

Standard Library: json, re, datetime, io, base64, math, asyncio, typing, pathlib
Data Analysis & Visualization: pandas, seaborn, and matplotlib (Note: Always configure matplotlib.use('Agg') for thread-safe server-side chart rendering)
AI & LLM Services: from ai.llm_functions import simple_prompt
Platform Facade: from valstorm_platform.platform_context import PlatformContext, User

2. Function Entry Point Specification

Every function must define an async def execute(platform: PlatformContext, **kwargs) entry point:

PYTHON

from valstorm_platform.platform_context import PlatformContext

async def execute(platform: PlatformContext, **kwargs):
    """
    Standard function entry point.
    
    Args:
        platform (PlatformContext): Injected unified platform facade.
        **kwargs: Dynamic arguments passed via API, trigger, or Flow.
        
    Returns:
        dict | list | Any: JSON-serializable result payload.
    """
    platform.log("Starting custom function execution", "info")
    
    lead_id = kwargs.get("lead_id")
    if not lead_id:
        platform.log("Missing required parameter: lead_id", "warning")
        return {"status": "error", "message": "lead_id is required"}
    
    # 1. Query lead details using the SQL Query Engine
    leads = await platform.query.sql(f"SELECT id, name, email, phone FROM lead WHERE id = '{lead_id}'")
    if not leads:
        return {"status": "error", "message": "Lead not found"}
        
    lead = leads[0]
    
    # 2. Create follow-up task
    task = await platform.records.create("task", {
        "name": f"Follow up with {lead.get('name', 'Lead')}",
        "status": "Not Started",
        "priority": "High",
        "related_to": {
            "id": lead["id"],
            "schema": "lead"
        }
    })
    
    platform.log(f"Created task {task.get('id')} for lead {lead_id}", "info")
    return {"status": "success", "task_id": task.get("id")}

3. PlatformContext API Reference

The PlatformContext facade provides dedicated namespaces for interacting with all platform services:

`platform.records` (CRUD & Hydration)

create(api_name: str, input_data: dict | list[dict], async_run: bool = False, countdown: int = 0, retries: int = 0, skip_triggers: bool = False): Creates single or bulk records. If async_run=True, offloads execution to Celery workers.
update(api_name: str, input_data: dict | list[dict], async_run: bool = False, countdown: int = 0, retries: int = 0, skip_triggers: bool = False): Updates records by ID.
delete(api_name: str, input_data: dict | list[dict], async_run: bool = False, countdown: int = 0, retries: int = 0, skip_triggers: bool = False): Deletes records by ID.
hydrate_batch(ids: list[str]) -> dict: Batch resolves prefixed record IDs (e.g. ['cont_123', 'task_456']) into display names and object schemas.
calculate_rollup(record_id: str, api_name: str, rollup_field_api_name: str) -> float: Evaluates rollup calculations dynamically.
merge(master_id: str, duplicate_id: str, field_overrides: dict = None): Merges duplicate records.

`platform.query` (Query Engine)

sql(query: str, bypass_cache: bool = False) -> list[dict]: Executes platform SQL queries with support for dynamic date keywords (today, last_n_days:30), ME filtering, and relation queries.
mongo(collection: str, pipeline: list[dict]) -> list[dict]: Executes raw MongoDB aggregation pipelines directly against tenant collections.
graphql(query: str, variables: dict = None) -> dict: Resolves GraphQL queries.
resolve_lookup(api_name: str, record_id: str) -> dict: Fetches target lookup record metadata.

`platform.schemas` (Object & Field Metadata)

get(api_name: str): Retrieves schema definition and field configurations for an object.
list() / list_all(): Lists all standard and custom schemas in the organization.
create(data: dict, save: bool = True) / delete(schema_id: str): Manages schema definitions.
create_field(field_data: dict) / delete_field(object_id: str, field_name: str): Dynamically adds or removes schema fields.

`platform.files` (Virtual File Service - VFS)

get_vfs_file(file_id: str) -> dict: Retrieves file metadata and content.
browse_vault(vault_id: str = 'root', bypass_cache: bool = False) -> dict: Lists vaults and items.
browse_path(string_path: str) -> dict: Resolves folder structures by human-readable path.
get_tree(bypass_cache: bool = False) -> dict: Fetches complete vault hierarchy tree.
move_item(item_id: str, to_vault_id: str, from_vault_id: str = None) -> dict: Moves files/folders between vaults.
delete_vfs_item(item_id: str) -> dict: Deletes an item from the VFS.

`platform.communications` (SMS & Email)

send_sms(to_phone: str, message: str, **kwargs): Sends an SMS message to an E.164 phone number.
send_email(request: dict | EmailRequest, async_run: bool = True): Sends single or templated outbound emails.

`platform.integrations` (External Connectors)

platform.integrations.google:
- create_drive_folder(name: str, parent_id: str = None)
- upload_drive_file(file_name: str, file_content: bytes, mime_type: str, parent_id: str = None)
- export_drive_file(file_id: str, mime_type: str, base64_encode: bool = True)
- move_drive_file(file_id: str, target_folder_id: str)
- modify_gmail_labels(message_id: str, add_label_ids: list[str] = None, remove_label_ids: list[str] = None)
- modify_gmail_thread_labels(thread_id: str, add_label_ids: list[str] = None, remove_label_ids: list[str] = None)
platform.integrations.salesforce:
- query(query: str): Executes SOQL query.
- create(api_name: str, data: dict | list[dict]): Creates records in Salesforce.
- update(data: list[dict]): Updates Salesforce records.
platform.integrations.twilio:
- lookup(phone_number: str): Performs carrier/caller lookup.
- delete_conversation(service_id: str, conversation_sid: str)

`platform.workflows` (Functions & Flows)

run_function(function_name: str, kwargs: dict): Invokes another function synchronously or asynchronously.
run_workflow(workflow_id: str, data: dict): Triggers a Flow Builder automation.
safe_execute(func: Callable, *args, **kwargs): Safely executes arbitrary synchronous/asynchronous callables.

`platform.tasks` (Scheduling)

schedule(name: str, func: str, run_at: datetime, data: dict): Schedules a function to run at a specific future UTC timestamp via scheduled_item.

`platform.utils` (Helpers)

render_template(template_string: str, data: dict) -> str: Renders dynamic variables into a text/HTML template.
phone_formatter(phone: Any) -> dict: Normalizes phone numbers to standard and E.164 formats.
aware_datetime(dt: datetime | str | None) -> datetime: Guarantees UTC-aware datetime objects.
iso_datetime(dt: datetime) -> str: Formats datetimes into ISO-8601 strings.
html_to_md(html: str) / md_to_html(md: str) / clean_html(html: str): Text format conversions.

4. Invoking Functions

Functions can be invoked across four primary execution vectors:

1. Direct REST API Execution

Execute a saved function by its API name or ID:

HTTP

POST /v1/automation/function
Content-Type: application/json
Authorization: Bearer <token>

{
  "function_name": "lead_tracker.py",
  "inputs": {
    "lookback_days": 14,
    "recipient_email": "[email protected]"
  }
}

In-Memory Validation (`/function/validate`)

To test code dynamically from an IDE or developer panel without saving it to the database, use the validation endpoint:

HTTP

POST /v1/automation/function/validate
Content-Type: application/json
Authorization: Bearer <token>

{
  "code": "async def execute(platform, **kwargs):\n    return {'hello': kwargs.get('name', 'World')}",
  "function_name": "test_sandbox.py",
  "inputs": {
    "name": "Valstorm Developer"
  }
}

2. Record Triggers

Call custom functions from inside a RecordTriggerContext routine:

PYTHON

from valstorm_platform.trigger_context import RecordTriggerContext

async def execute(context: RecordTriggerContext) -> None:
    for record_id, record in context.new_map.items():
        if record.get("status") == "Closed Won":
            await context.workflows.run_function(
                function_name="generate_customer_agreement.py",
                kwargs={"deal_id": record_id, "user_id": context.user.id}
            )

3. Visual Flow Builder

In Flow Builder, add a Function Node step:

Select the target function (e.g. process_meeting_notes.py).
Map step inputs using visual flow variables (e.g. {{step_1.transcript}}).
Downstream nodes can consume the return values via {{step_2.tasks}}.

4. Scheduled Items & Recurring Jobs

Schedule one-off or recurring background function tasks programmatically:

PYTHON

from datetime import datetime, timezone, timedelta

# Schedule 24 hours from now
run_time = datetime.now(timezone.utc) + timedelta(days=1)

await platform.tasks.schedule(
    name="Process Overdue Accounts",
    func="realtor_status_fetcher.py",
    run_at=run_time,
    data={"filter_type": "overdue"}
)

5. Practical Implementation Patterns

Pattern A: Data Science, Visual Dashboards & Email Reports

When generating charts with matplotlib / seaborn, always configure the Agg backend:

PYTHON

import io
import base64
import pandas as pd
import matplotlib
matplotlib.use('Agg') # CRITICAL: Server-side thread-safe backend
import matplotlib.pyplot as plt
import seaborn as sns
from valstorm_platform.platform_context import PlatformContext

async def execute(platform: PlatformContext, **kwargs):
    # 1. Fetch data
    deals = await platform.query.sql("SELECT id, name, amount, stage FROM deal WHERE created_date >= last_n_days:30")
    if not deals:
        return {"status": "empty", "message": "No deals found"}
        
    df = pd.DataFrame(deals)
    
    # 2. Render plot
    plt.figure(figsize=(8, 4))
    sns.barplot(x="stage", y="amount", data=df, estimator=sum, errorbar=None)
    plt.title("Pipeline Value by Stage")
    plt.tight_layout()
    
    # 3. Export to base64
    buffer = io.BytesIO()
    plt.savefig(buffer, format="png", dpi=150)
    buffer.seek(0)
    chart_base64 = base64.b64encode(buffer.read()).decode("utf-8")
    plt.close()
    
    return {
        "status": "success",
        "total_value": float(df["amount"].sum()),
        "chart_png_base64": chart_base64
    }

Pattern B: AI-Powered Transcript Analysis & Action Item Generation

Leverage built-in LLM prompting to extract structured insights and create CRM records:

PYTHON

import json
from valstorm_platform.platform_context import PlatformContext
from ai.llm_functions import simple_prompt

async def execute(platform: PlatformContext, **kwargs):
    transcript = kwargs.get("transcript", "")
    contact_id = kwargs.get("contact_id")
    
    if not transcript:
        return {"error": "Missing transcript"}
        
    prompt = f"""
    You are an expert CRM assistant. Analyze this transcript and extract key action items as a JSON array.
    Each item must have 'title' and 'due_in_days' (integer).
    Transcript:
    {transcript}
    """
    
    raw_ai_response = await simple_prompt(prompt=prompt, model="gemini-2.5-flash")
    tasks_to_create = json.loads(raw_ai_response)
    
    created_tasks = []
    for item in tasks_to_create:
        task = await platform.records.create("task", {
            "name": item["title"],
            "status": "Not Started",
            "related_to": {"id": contact_id, "schema": "contact"} if contact_id else None
        })
        created_tasks.append(task)
        
    return {"status": "success", "created_tasks_count": len(created_tasks)}

6. Best Practices & Troubleshooting

Practice	Recommendation
Thread-Safe Plotting	Always call `matplotlib.use('Agg')` immediately after importing `matplotlib`.
Long-Running CUD	Set `async_run=True` on `platform.records.create/update/delete` when mutating >50 records to offload work to background Celery workers.
Query Parameter Escaping	Escape user-supplied strings or use parameterized filters when querying `platform.query.sql()`.
Error Handling	Wrap logic in `try/except` and log detailed tracebacks via `platform.log(str(e), "error")`.
Testing Before Deployment	Use the `POST /v1/automation/function/validate` endpoint to test logic before saving to production functions.

Functions

1. Architecture & Sandboxed Execution

Available Runtime Libraries & Tooling

2. Function Entry Point Specification

3. PlatformContext API Reference

platform.records (CRUD & Hydration)

platform.query (Query Engine)

platform.schemas (Object & Field Metadata)

platform.files (Virtual File Service - VFS)

platform.communications (SMS & Email)

platform.integrations (External Connectors)

platform.workflows (Functions & Flows)

platform.tasks (Scheduling)

platform.utils (Helpers)

4. Invoking Functions

1. Direct REST API Execution

In-Memory Validation (/function/validate)

2. Record Triggers

3. Visual Flow Builder

4. Scheduled Items & Recurring Jobs

5. Practical Implementation Patterns

Pattern A: Data Science, Visual Dashboards & Email Reports

Pattern B: AI-Powered Transcript Analysis & Action Item Generation

6. Best Practices & Troubleshooting

`platform.records` (CRUD & Hydration)

`platform.query` (Query Engine)

`platform.schemas` (Object & Field Metadata)

`platform.files` (Virtual File Service - VFS)

`platform.communications` (SMS & Email)

`platform.integrations` (External Connectors)

`platform.workflows` (Functions & Flows)

`platform.tasks` (Scheduling)

`platform.utils` (Helpers)

In-Memory Validation (`/function/validate`)