Automations

Satia's Python execution engine runs business logic, external API integrations, conversational channel bridges, and custom agent tools inside an isolated cloud sandbox that automatically enforces the active Workspace's security boundary (EffectiveBoundary).

1. Architecture & the 3 Script Modes in Satia

1.1 The Data Hub Rule (2-Node Flows)

Satia is the company's System of Record (Data Hub). Every automation must have Satia's business tables as one of its two endpoints:

• Inbound (External System ──► Satia DB): An external webhook or scheduled cron job queries an external API and creates/updates records in a Satia business table.
• Outbound (Satia DB ──► External System): A record change in a Satia table (event) or a periodic sweep (cron) pushes data to an external system or notifies a customer.
• Internal (Satia DB ──► Satia DB): Scheduled processing over the business's own tables (closings, audits, reminders).

Golden Rule of Short Flows: Always design 2-node flows (Source ──► Destination), never monolithic chains across multiple external platforms without passing through Satia. If the user asks to "connect Siigo with Slack", build two separate automations coupled through a Satia business table (Siigo ──► Satia and Satia ──► Slack). This guarantees full traceability, isolated retries, and complete auditability.

1.2 The 3 Python Execution Modes

1.3 Sandbox Execution Environment Rules

• Automatic Entrypoints: NEVER write if __name__ == "__main__": at the end of a script. Satia's engine executes the file and automatically invokes run(event_data), main(event_data), on_inbound(event_data), or on_outbound(event_data) if defined (and prevents double execution if you already called them at the bottom).
• Pre-installed Allowed Libraries: satia_sdk, requests, pandas, numpy, datetime, json, math, pytz, re, uuid, hashlib, hmac, base64, urllib.parse, google-api-python-client.
• Isolation & Security:
  • The sandbox blocks reading host filesystem files or internal host environment variables and operates exclusively through satia_sdk.
  • If a script accidentally prints the value of a secret configured in automation_secrets using print(), the runner automatically redacts it in the logs as [REDACTED_SECRET].
  • Truncated placeholders are strictly forbidden: Never copy truncated text from conversation history ([AGED_ARG...] or <truncated ... chars>) into the code parameter.

2. Mandatory Development Workflow (Sandbox TDD) & Secret Management

1. Design and build a realistic mock_context:
   • Prepare a JSON dictionary that accurately simulates the exact event_data structure the script will receive in production.
2. Always run automation__test_script(code=..., mock_context=...) before saving:
   • Strict rule: Never call automation__save_script or insert/update agent_tools.script_code without first running automation__test_script and verifying success: true in the results.
   • If the script is a Custom Channel Bridge (on_inbound / on_outbound), test both directions:
     • Inbound: mock_context={"_direction": "inbound", "payload": {...}}
     • Outbound: mock_context={"_direction": "outbound", "message_to_send": "Hola", "contact": {"id": "...", "platform_identifiers": {...}}}
3. Secret Management & Automations Panel (/automations):
   • NEVER ask the user to paste API keys, tokens, or passwords in the chat.
   • When saving with automation__save_script, declare required credentials in required_secrets (or create them with automations__request_secret_key):
     Copy

     [
       {
         "key": "SIIGO_ACCESS_KEY",
         "description": "API access key generated in Siigo Cloud > Alliances > API Credentials."
       }
     ]
     

   • automation__save_script automatically creates the empty secret slots in automation_secrets.
   • Where the user manages automations and secrets (/automations): Always tell the user that the Automations (/automations) page in the web platform is where they can:
     • Securely enter and save the secrets / API keys required by each automation.
     • View general information for each automation, inspect its detailed execution history (script_executions) and logs, and activate or deactivate (is_active) any automation with a single click.
   • To check which secrets already exist in the Workspace without exposing their values, run:
     Copy

     SELECT id, key, description, value FROM automation_secrets
     

     (The value column never returns the plaintext secret; the engine returns [CONFIGURED] if already filled by the user or [EMPTY - PENDING USER INPUT] if still empty, and you must never filter or alias value in SQL).

3. Complete satia_sdk Reference

Every script has the event_data dictionary and satia_sdk modules pre-injected globally:

import satia_sdk
from satia_sdk import db, http, ai, agents, channels, templates, tasks, browser
# Note: `satia_sdk.database` also exists as an alias for `satia_sdk.db`.

A. Database (satia_sdk.db)

Operates directly on business tables by their slug (e.g., 'pedidos', 'clientes', 'productos', 'citas') and on authorized system tables (knowledge_docs, knowledge_folders, schedules, schedule_rules, event_items, agent_tasks, etc.).

• Automatic flattening on reads: All business table columns are returned directly at the root of each returned dictionary (row["estado"], row["total"], row["starts_at"]).
• Automatic grouping on writes: In db.insert and db.update, pass column names directly as keyword arguments (**kwargs).

1. db.query(sql, params=None) -> list[dict]

Executes a standard SQL SELECT query against business table slugs or system tables.

• Supports positional parameters (%s passing params=[val1, val2]) or named parameters (:nombre passing params={"nombre": val}), automatically escaping single quotes:

pedidos = db.query(
    """
    SELECT p.id, p.name, p.total, p.estado, p.starts_at, c.name AS cliente_nombre
    FROM pedidos p
    LEFT JOIN clientes c ON c.id = (p.cliente_id->>0)::uuid
    WHERE p.estado = :estado AND p.starts_at >= :desde
    ORDER BY p.starts_at DESC
    LIMIT 20
    """,
    params={"estado": "pagado", "desde": "2026-10-01T00:00:00Z"}
)

2. db.get(table_slug, id) -> dict | None

Fetches a single record by its table slug and UUID (or None if not found):

cliente = db.get("clientes", "UUID-DEL-CLIENTE")
if cliente:
    print(cliente.get("name"), cliente.get("phone"))

3. db.filter(table_slug, **kwargs) -> list[dict]

Filters records of a table by exact equality (=) on one or more columns:

ventas_pagadas = db.filter("pedidos", estado="pagado", metodo_pago="transferencia")

4. db.find_entity(table_slug, filters, case_insensitive=True) -> list[dict]

Searches records using a filter dictionary. When case_insensitive=True (default), compares text columns using ILIKE:

contactos = db.find_entity("clientes", {"email": "juan@empresa.com", "ciudad": "Bogotá"})

5. db.insert(table_slug, **kwargs) -> Response

Inserts a new record by passing its columns as **kwargs.

• Returns the RPC response object whose .data attribute contains the list with the created record dictionary (res.data[0]["id"]).
• If an event or offering table has a schedule column, db.insert and db.update automatically synchronize its value with the indexed starts_at / ends_at columns.

res = db.insert(
    "pagos",
    name="Pago Factura #1042",
    starts_at="2026-10-20T14:30:00Z",
    monto=150000,
    metodo_pago="tarjeta",
    pedido_id=["UUID-DEL-PEDIDO"]  # 'relation' columns always take a list of UUIDs
)
nuevo_pago_id = res.data[0]["id"] if getattr(res, "data", None) else None

6. db.update(table_slug, id, **kwargs) -> Response

Surgically updates only the columns passed in **kwargs (performing an automatic merge without losing other fields on the record).

• To clear/remove a column's value: Pass campo=None (e.g., db.update("clientes", cliente_id, nota_temporal=None)).
• Optional full replacement: Pass replace_attributes=True only if you want to completely overwrite all dynamic attributes of the record.
• Native anti-loop protection: If all values passed to db.update are already identical to the current values stored in the database, the SDK silently skips the write (UPDATE_SKIPPED) to avoid firing redundant update events.

db.update("pedidos", "UUID-DEL-PEDIDO", estado="facturado", factura_id="FE-9012")

7. db.delete(table_slug, id, reason=None) -> Response

Deletes a record by its UUID (performing safe or cascading deletion as appropriate):

db.delete("citas", "UUID-DE-LA-CITA", reason="Cancelada desde sistema externo")

8. db.execute(table_name, action, data, match=None) -> Response

Direct DML method (action in 'insert', 'update', 'delete'), useful when working with dynamically constructed dictionaries or system tables (knowledge_docs, schedule_rules, event_items):

# Add a new log entry ('entry') to a logbook or medical record ('book') in Satia Docs:
db.execute("knowledge_docs", "insert", {
    "parent_id": "UUID-DEL-LIBRO-PADRE",
    "doc_type": "entry",
    "title": "Control Post-Operatorio",
    "content": "Paciente presenta evolución favorable..."
})

9. Defensive db Helpers (db.get_attr & db.normalize)

• db.get_attr(obj, key, default=None): In Satia, relation columns are always stored as a list of UUIDs (["uuid-1"]), even for 1-to-1 relationships. Passing a list directly to db.get("clientes", event_data.get("cliente_id")) will fail. Always use db.get_attr:
  Copy

  # Automatically extracts the first UUID if 'cliente_id' comes as a list ['uuid-1']:
  cliente_id = db.get_attr(event_data, "cliente_id")
  if cliente_id:
      cliente = db.get("clientes", cliente_id)
  

• db.normalize(text): Converts any value to a lowercase, whitespace-trimmed string ("" if None) for safe comparisons:
  Copy

  if db.normalize(event_data.get("metodo_pago")) == "transferencia":
      ...
  

10. Handling Special Columns (media, line_items, schedule) in Scripts

• media Columns:
  • For columns with is_multiple: false, the value is a MediaItem dict: {"url": "https://...", "name": "recibo.jpg", "type": "image"|"video"|"audio"|"document", "caption": "..."}.
  • For columns with is_multiple: true, it is a list of dicts [MediaItem, ...].
  • To send the file to an external API (Outbound), extract .get("url"). To store an external file in Satia (Inbound), pass a valid dictionary containing at least "url" and "type".
• line_items Columns (event_items Table):
  • Cart/order line items live in the event_items table (event_id, offering_id, attribute_key, name, unit_price, quantity, total_price, attributes), never as inline JSON inside the parent event row.
• Dates & Timezones (starts_at / SATIA_TIMEZONE):
  • The database stores all timestamps in UTC (starts_at, ends_at, created_at). When building time filters in SQL, use datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") (always with the Z suffix).
  • To display a human-readable date or time to the end user, always convert from UTC to the business's timezone using the pre-injected SATIA_TIMEZONE environment variable with pytz (never use hardcoded timedelta(hours=...)):
    Copy

    import os, pytz
    from datetime import datetime
    
    tz = pytz.timezone(os.environ.get("SATIA_TIMEZONE", "UTC"))
    dt_utc = datetime.fromisoformat(row["starts_at"].replace("Z", "+00:00"))
    dt_local = dt_utc.astimezone(tz)
    hora_legible = dt_local.strftime("%Y-%m-%d %I:%M %p")
    

B. External HTTP Connections (satia_sdk.http)

Provides the 5 standard HTTP methods:

• http.get(url, headers=None, params=None, connector_key=None, connection_id=None, contact_id=None)
• http.post(url, json=None, data=None, headers=None, params=None, connector_key=None, connection_id=None, contact_id=None)
• http.put(url, json=None, data=None, headers=None, params=None, connector_key=None, connection_id=None, contact_id=None)
• http.patch(url, json=None, data=None, headers=None, params=None, connector_key=None, connection_id=None, contact_id=None)
• http.delete(url, headers=None, params=None, connector_key=None, connection_id=None, contact_id=None)

All methods return a standard requests.Response object (.status_code, .json(), .text, .raise_for_status()).

Mode 1: APIs with Static Credentials (automation_secrets -> os.environ)

For services using static API Keys, Basic Auth, or Bearer Tokens (e.g., Stripe, Siigo, SendGrid, Alegra, Wompi):

import os
from satia_sdk import http

api_key = os.environ.get("STRIPE_SECRET_KEY")
if not api_key:
    raise ValueError("Missing STRIPE_SECRET_KEY configuration on the Automations page (/automations).")

res = http.post(
    "https://api.stripe.com/v1/payment_intents",
    headers={"Authorization": f"Bearer {api_key}"},
    data={"amount": 50000, "currency": "usd"}
)
res.raise_for_status()
data = res.json()

Mode 2: Dynamic OAuth 2.0 Connectors (connector_key / oauth_connections)

Satia automatically manages OAuth 2.0 tokens (refresh_token renewal and injection via Pipedream proxy or direct Bearer header) for active connectors in integration_connectors.

1. Mandatory pre-check: Before coding against a connector, verify its existence, authentication type, and official documentation in the database:
   Copy

   SELECT key, name, auth_type, documentation FROM integration_connectors WHERE is_active = true
   

   And check whether the business or team member already has an active connected account:
   Copy

   SELECT id, connector_key, contact_id, account_email, expires_at FROM oauth_connections
   

2. Authorization Magic Link (If not connected):
   • If the connector has auth_type = 'oauth2' and does not appear in oauth_connections, there is no "Integrations" screen in Settings. Provide the user with the direct Markdown link: [Connect <Connector Name>](/api/oauth/<connector_key>/authorize)
3. Rules when calling http.* with OAuth:
   • NEVER look up the OAuth token in os.environ or manually build the Authorization header.
   • Pass only connector_key="<key>".
   • Per-Employee/Resource Connections (contact_id or connection_id): If the business has calendars or OAuth accounts linked to individual staff members (e.g., a specific doctor's calendar), pass contact_id="UUID-DEL-EMPLEADO" (or connection_id="UUID-DE-LA-CONEXION"). If you omit contact_id, the SDK uses the company-wide connection (contact_id IS NULL).

from satia_sdk import http

res = http.post(
    "https://www.googleapis.com/calendar/v3/calendars/primary/events",
    json={
        "summary": "Consulta Odontológica",
        "start": {"dateTime": "2026-10-21T15:00:00Z"},
        "end": {"dateTime": "2026-10-21T15:30:00Z"}
    },
    connector_key="google_calendar",
    contact_id="UUID-DEL-DOCTOR"  # Optional: uses that specialist's OAuth account
)

C. Artificial Intelligence (satia_sdk.ai)

Runs native AI inference inside the script (summaries, structured extraction, text classification) while automatically tracking token usage in script_executions:

from satia_sdk import ai

clasificacion = ai.generate(
    prompt=f"Classify the sentiment of this review as POSITIVO, NEUTRO, or NEGATIVO:\n{event_data.get('comentario')}"
)

D. Agent Orchestration (satia_sdk.agents)

Wakes up a transactional agent to start or continue a conversation with a specific contact.

• WhatsApp Restriction: agents.trigger_conversation sends free-form text generated by the agent, so on WhatsApp channels it only works if the contact has messaged within the last 24 hours. On Telegram, Web, or open-window channels, it works at any time.

from satia_sdk import agents

agents.trigger_conversation(
    contact_id="UUID-DEL-CONTACTO",
    instruction="Confirm to the customer that their payment was approved and ask whether they prefer in-store pickup or home delivery.",
    agent_id="UUID-DEL-AGENTE",  # Optional if the script already has an assigned agent_id
    context={"pedido_id": event_data.get("id"), "total": event_data.get("total")}
)

E. WhatsApp Templates (satia_sdk.templates)

To contact users over WhatsApp outside the 24-hour service window (first outreach, appointment reminders, payment collection, bulk campaigns), sending Meta-approved templates is mandatory.

• channel_id must always be the WhatsApp channel id from the channels table (SELECT id FROM channels WHERE channel_type = 'whatsapp'), never a conversations id.

from satia_sdk import templates

# 1. List approved templates on the channel
aprobadas = templates.list(channel_id="UUID-CANAL-WHATSAPP", status="APPROVED")

# 2. Send a template to a contact (supports positional variables in a list or named variables in a dict, plus media headers)
templates.send(
    contact_id="UUID-CONTACTO",
    template_name="recordatorio_cita",
    language="es",
    variables=["María", "Mañana a las 10:00 AM"],  # Or dict: {"nombre": "María", "hora": "10:00 AM"}
    channel_id="UUID-CANAL-WHATSAPP",
    media_url="https://dominio.com/orden.pdf",     # Required if the template has an IMAGE, VIDEO, or DOCUMENT header
    header_type="document",                        # Optional: 'image', 'video', 'document' (auto-detected if omitted)
    header_filename="Orden.pdf"                    # Optional for document headers
)

# 3. Optimized bulk sending by batch (Outbox Batch with automatic rate-limit control)
resultado = templates.send_bulk(
    contact_ids=["uuid-1", "uuid-2", "uuid-3"],
    template_name="promocion_mensual",
    language="es",
    variables=["20%"],
    channel_id="UUID-CANAL-WHATSAPP"
)
# resultado -> {"sent": 3, "failed": 0, "errors": [], "total": 3}

# 4. Create a new template in Meta (remains in PENDING status until approved by Meta)
templates.create(
    name="confirmacion_despacho",
    category="UTILITY",
    body_text="Hola {{1}}, tu pedido {{2}} ha sido despachado.",
    language="es",
    examples=["Carlos", "PED-104"],
    channel_id="UUID-CANAL-WHATSAPP"
)

F. Scheduled Tasks (satia_sdk.tasks)

Schedules a deferred task (agent_tasks) for an agent to execute at an exact future date/time:

from satia_sdk import tasks

tasks.schedule(
    instruction="Check whether the customer completed payment for the order and follow up",
    scheduled_at="2026-10-21T10:00:00-05:00",  # Future ISO 8601 timestamp
    contact_id="UUID-CONTACTO",                # Optional
    agent_id="UUID-AGENTE",                    # Optional (defaults to the current agent)
    context={"pedido_id": event_data.get("id")}
)

G. Headless Browser (satia_sdk.browser)

Controls an AI-assisted Headless Chrome browser to interact with government portals, judicial systems, or legacy ERPs that lack a public API.

• Golden rules:
  1. Always define schema to receive a clean, structured JSON dictionary.
  2. Restrict max_steps (e.g., 15 to 25) and allowed_domains to prevent the browser agent from drifting to external search engines.
  3. Wrap the call in try/except and pass credentials from os.environ inside context.

import os
from satia_sdk import browser

res = browser.navigate(
    instruction="Log in, search for the invoice matching the given NIT, and extract the CUFE and DIAN status.",
    url="https://nube.siigo.com/login",
    max_steps=15,
    timeout=300,
    allowed_domains=["nube.siigo.com", "siigo.com"],
    context={
        "user": os.environ.get("SIIGO_USER"),
        "pass": os.environ.get("SIIGO_PASS"),
        "nit": event_data.get("nit")
    },
    schema={
        "cufe": {"type": "string", "description": "Unique electronic invoice code (CUFE)"},
        "estado_dian": {"type": "string"}
    }
)

H. Conversational Channel Bridges (satia_sdk.channels)

Connects any external CRM or messaging platform (e.g., Kommo CRM, Chatwoot, Respond.io, ManyChat, iMessage) as a bidirectional conversational channel (channels.channel_type = 'webhook').

• Strict Rule (1 Agent = 1 Webhook Channel = 1 Bridge Script): Each agent has its own dedicated channel and its own script with a unique webhook_slug.
• Do not use agents.trigger_conversation for incoming chat messages: Always use channels.receive_message(...) so the message enters Satia's official pipeline with conversation memory, debouncing, multimodal support, and automatic propagation of inbound_payload and turn_metadata.
• Engineering guide and complete templates: Consult discovery__get_manual(topic='webhook_channels') (or topic='agents_and_channels') before writing or modifying a channel bridge.

from satia_sdk import channels, http

def on_inbound(event_data: dict):
    payload = event_data.get("payload", {})
    # 1. If the message was sent by a human operator in the external CRM, record the echo and activate handover:
    if payload.get("is_human_echo"):
        return channels.record_echo(
            external_id=payload.get("sender_id"),
            message_content=payload.get("text", ""),
            activate_handover=True
        )
    # 2. Ingest the customer message into the Satia agent pipeline:
    return channels.receive_message(
        external_id=payload.get("sender_id"),
        message_content=payload.get("text", ""),
        contact_name=payload.get("sender_name"),
        phone=payload.get("phone"),
        media_url=payload.get("media_url"),
        media_mime_type=payload.get("media_mime_type"),
        platform_identifiers={"crm_talk_id": payload.get("talk_id")},
        turn_metadata={"pipeline_id": payload.get("pipeline_id")}
    )

def on_outbound(event_data: dict):
    # Automatically invoked when the agent replies
    contact = event_data.get("contact") or {}
    talk_id = (contact.get("platform_identifiers") or {}).get("crm_talk_id")
    http.post(
        "https://api.crm-externo.com/v1/messages",
        json={
            "talk_id": talk_id,
            "text": event_data.get("message_to_send"),
            "media_url": event_data.get("media_url")
        },
        connector_key="crm_externo"
    )

4. Trigger Types, trigger_config, & Exact event_data Anatomy

When saving with automation__save_script(name, code, trigger_type, trigger_config, ...), configure the trigger according to this table:

Anatomy of event_data in a Custom Tool (agent_tools with type = 'custom')

When an agent invokes a custom tool (agent_tools.script_code), the engine executes def run(event_data): and injects:

def run(event_data: dict) -> dict:
    tool_args = event_data.get("tool_args", {})          # Arguments passed by the LLM matching input_schema
    contact_id = event_data.get("contact_id")            # UUID of the contact in the active conversation
    conversation_id = event_data.get("conversation_id")  # UUID of the active conversation
    channel_id = event_data.get("channel_id")            # UUID of the channel
    agent_id = event_data.get("agent_id")                # UUID of the agent that invoked the tool
    inbound_payload = event_data.get("inbound_payload")  # Original turn payload (on webhook channels)
    turn_metadata = event_data.get("turn_metadata")      # Metadata for the current turn
    ...
    return {"ok": True, "resultado": "..."}

5. Critical Idempotency & Infinite-Loop Prevention Rules

1. In event triggers (Infinite-Loop & Price Accumulation Protection):

   • If your script triggers on event_type = "update" or "any" for a table (e.g., pedidos) and calls db.update("pedidos", ...) inside the script, that write will generate a new UPDATE event in PostgreSQL.
   • Rule 1 — Mutually Exclusive State Transitions: The initial state that triggers the logic and the final state written by the script must be distinct and must never be combined in the same if condition:
     Copy

     estado_actual = event_data.get("estado")
     # ✅ Only acts when status is 'pagado' and no invoice has been issued yet:
     if estado_actual == "pagado" and not event_data.get("factura_emitida_id"):
         factura_id = emitir_factura(event_data)
         db.update("pedidos", event_data["id"], estado="facturado", factura_emitida_id=factura_id)
     else:
         print(f"Skipping execution: estado={estado_actual}")
     

   • Rule 2 — Immutable Base Prices vs. Calculated Totals: Never use an accumulated total column (total) as a fallback for the base price (subtotal) when adding extra charges; always read the immutable base column so retries never duplicate charges (or better yet, delegate column calculations to the Data Modeler via reactive formulas).
   • Rule 3 — Guard before proactive messages: Before calling agents.trigger_conversation or templates.send, verify on the record whether the notification has already been sent.

2. In cron triggers (Diagnostic Logging + Filter → Process → Mark):

   • Every cron script must log the queried UTC window at the start and the count of processed records at the end, and immediately mark each processed record so the next run does not pick it up again:
     Copy

     from datetime import datetime, timedelta
     now = datetime.utcnow()
     start_str = (now + timedelta(minutes=25)).strftime("%Y-%m-%dT%H:%M:%SZ")
     end_str = (now + timedelta(minutes=35)).strftime("%Y-%m-%dT%H:%M:%SZ")
     print(f"[CRON] UTC Now: {now.isoformat()} | Window: {start_str} -> {end_str}")
     
     citas = db.query(
         "SELECT id, cliente_id, starts_at FROM citas "
         "WHERE estado = 'programada' AND starts_at >= :ini AND starts_at <= :fin",
         params={"ini": start_str, "fin": end_str}
     )
     print(f"[CRON] Appointments found: {len(citas)}")
     for cita in citas:
         cliente_id = db.get_attr(cita, "cliente_id")
         if cliente_id:
             agents.trigger_conversation(contact_id=cliente_id, instruction="Remind the customer about their appointment in 30 minutes.")
             db.update("citas", cita["id"], estado="recordatorio_enviado")
     

3. In webhook triggers (External ID Deduplication & [UNIQUE] Columns):

   • External providers (Stripe, MercadoPago, Typeform, Shopify) retry webhooks on network delays. Before calling db.insert, always check with db.filter whether the external identifier has already been processed:
     Copy

     ext_id = event_data.get("id")
     if ext_id and db.filter("pagos", referencia_externa=str(ext_id)):
         print(f"Webhook {ext_id} already processed. Ignoring.")
     else:
         db.insert("pagos", name=f"Pago {ext_id}", referencia_externa=str(ext_id), monto=event_data.get("amount"))
     

6. Inspecting Past Executions & Surgical Code Editing

1. Auditing Execution History (script_executions & messages.tool_calls):

   • To diagnose why an automation failed or what it printed, query script_executions filtering by script_id:
     Copy

     SELECT id, status, error_message, execution_time_ms, entities_count, started_at
     FROM script_executions
     WHERE script_id = '<UUID-DEL-SCRIPT>'
     ORDER BY started_at DESC
     LIMIT 5
     

   • If the logs of an execution are large, inspect them with database__inspect_text(table_name="script_executions", record_id="<EXEC_UUID>", column_name="logs", search_pattern="ERROR").
   • To inspect past runs of ephemeral scripts (python_executor__execute_script) in chat history, search messages.tool_calls (when role = 'model', content is NULL and the code lives inside tool_calls) using database__inspect_text(table_name="messages", record_id="...", column_name="tool_calls").

2. Surgical Editing of Existing Scripts (database__inspect_text + database__patch_text):

   • When you need to make a targeted adjustment to an existing script (automation_scripts.code or agent_tools.script_code), do not rewrite the entire file from scratch:
     1. Inspect the relevant lines with database__inspect_text(table_name="automation_scripts", record_id="<ID>", column_name="code", start_line=1, end_line=120).
     2. Test the updated code with automation__test_script.
     3. Apply the surgical change with database__patch_text(table_name="automation_scripts", record_id="<ID>", column_name="code", replacements=[{"target_text": "...", "replacement_text": "...", "start_line": ..., "end_line": ...}]) (or use automation__save_script(script_id="<ID>", ...) if you are also updating trigger_config or required_secrets).