Python AI Tools — Author Guide

How to write Script (Python) AI Tools for Netridium. This assumes you know Python and understand Netridium (connectors, AI Tools, running a report). It covers the runtime, the full script API, the output envelope types, the model (prompt() / agent()), limits, and a library of worked examples.

Where this runs. A Python AI Tool is the Script (Python) type in the Edit AI Tool form's Type dropdown. (The type is gated per workspace — if you don't see it, a System Manager must enable Python in Settings.) Paste your script into the code editor, pick the connector(s) it may use, and Run.

1. The execution model in one minute

• Your script runs in a Pyodide sandbox (CPython compiled to WebAssembly) in a throwaway subprocess that holds no secrets and has no network, no filesystem, and no environment access. The only way out is the host bridge: mcp(), prompt(), agent().
• The runner supports top-level await — write rows = await mcp(...).call(...) directly, no asyncio.run().
• Your script produces its result by calling emit({...}) with a typed envelope. The last emit() wins; anything not emitted is discarded. A script that never calls emit() is not an error — it returns a green {"type": "text", "text": "No data returned"} report.
• Everything is read-only. No AI Tool of any type can write to a connector; write tools are hidden and rejected for the whole run (see §5).

What's available

• The Python standard library ships with Pyodide: json, datetime, csv, re, math, statistics, collections, itertools, functools, io, base64, hashlib, decimal, textwrap, string, random, and the rest. import json works fine — the sandbox just doesn't pre-bind a bare json name (it imports it internally as _json), so import json yourself if you want it.
• The host builtins, always present (no import): mcp, emit, list_connectors, prompt, agent, and the exception classes McpError, PromptError, AgentError.

What's not available

• No third-party packages — numpy, pandas, requests, etc. are not loaded. Imports of them fail. Use the standard library plus the connector data.
• No network / sockets / files / env vars — reach external systems only through connectors (mcp()).
• No input parameters (yet). Input parameters defined on an AI Tool are delivered to Prompt (ai_mcp) and LSPL types only. The Script (Python) path is not wired to inputs — your script receives no user-supplied values at runtime. If you need user input, use a Prompt/LSPL type, or hard-code/derive the values in the script.
• print() is not output. Tenant print() goes to the worker logs (for your debugging during development), never to the report. The report is only what you emit(). To "see" a value, emit() it as json/text.

2. Script API reference

mcp(name) → connector proxy

Returns a proxy for a connector linked to this AI Tool. Resolution: by the connector's display name first (the name shown on /connectors), falling back to the connector type only when exactly one connector of that type is linked. If two connectors share a type (e.g. a sandbox and a production Maxio), mcp("maxio") is ambiguous and errors — use the exact name.

conn = mcp("Maxio (sandbox)")     # by display name (always unambiguous)
conn = mcp("salesforce")          # by type — only if exactly one is linked

await conn.call(tool, arguments=None)

Calls one MCP tool and returns its payload. Async — await is required. arguments is a dict (defaults to {}).

The payload is unwrapped for you with a three-tier rule: prefer the tool's structuredContent; else JSON-parse its text content; else, for the Markdown-report style some servers use (Maxio/NetSuite), parse it into record dicts; else hand back the raw string. So you usually get a dict/list, not a blob — but the exact shape is the tool's, so probe an unfamiliar tool first (§7, Example 2).

rows = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 20})

await conn.list_tools()

Returns the connector's tools as a list of {"name", "description", "inputSchema"} dicts — use it to discover exact tool names and argument schemas at runtime. (Counts against the MCP-call cap, §6.)

list_connectors() → list of {"name", "type"}

The connectors available to this run. Handy when you're unsure what's linked.

for c in list_connectors():
    print(c["name"], c["type"])   # goes to logs

emit(envelope) → None

Records the output envelope (see §3 for every type). Last call wins. The envelope must be JSON-serializable — only dict, list, str, int, float, bool, None. Convert anything else first (d.isoformat(), list(s), float(decimal_val)). emit() round-trips through JSON immediately, so a non-serializable value fails there, with your stack trace.

await prompt(instruction, input=None, schema=None, model=None)

Calls the model through the broker (the API key never enters the sandbox). See §4.

await agent(goal, tools=None, max_steps=8, model=None, allow_writes=False)

Runs a small, read-only, tool-using loop. See §4.

Errors — what to catch

Gotcha: the McpError class exists, but .call() does not raise it today — a failed/isError tool call surfaces as a plain exception. Catch Exception (or BaseException) around .call(), not McpError. prompt()/agent() do convert their failures to PromptError/AgentError.

3. emit() envelope reference — every output type

The type field selects how the report is stored and rendered. Exactly one type per emit(). Required fields are marked (*); the worker rejects a bad envelope with a tenant-readable error naming the problem.

Excel → .xlsx (interactive sheet grid)

emit({
    "type": "excel",                          # (*)
    "sheets": [                               # (*) non-empty list
        {
            "name": "Customers",              # (*) tab name; sanitized to <=31 chars
            "data": [                         # (*) list of ROW DICTS (not a dict!)
                {"id": 1, "name": "Alice"},
                {"id": 2, "name": "Bob"},
            ],
        },
        # … more sheets …
    ],
})

• Columns come from the FIRST row's keys. Later rows may omit keys (blank cells); new keys in later rows do not add columns. Normalize your rows if they're ragged.
• Cell values: str/int/float/bool/None render natively; nested dict/list values are JSON-stringified into the cell.
• Empty data → a single (No data) row for that sheet.
• Passing a dict where data expects a list is the #1 error — unwrap the row list first (e.g. data=raw["customers"]).

HTML → .html (sandboxed iframe)

emit({
    "type": "html",                           # (*)
    "html": "<h1>Quarterly recap</h1><p>…</p>",# (*) fragment or full document, string
})

• Rendered in a sandbox="" iframe: no script execution, no form submit, no navigation. Treat it as a static document. Inline SVG works (for a standalone chart prefer image).

Image → .svg / .png / .jpg (inline)

emit({
    "type": "image",                          # (*)
    "format": "svg",                          # (*) "svg" | "png" | "jpeg" (alias "jpg"), case-insensitive
    "data": "<svg xmlns=…> … </svg>",         # (*) string; meaning depends on format
})

• format="svg" → data is the raw SVG markup (a string starting with <svg).
• format="png" / "jpeg" → data is base64 of the raster bytes, with no data:image/…;base64, prefix (a prefix corrupts the image).
• Prefer SVG for charts/diagrams your script composes — crisp at any zoom and smaller than rasters.

Text → .txt (monospace <pre>, wrapped)

emit({
    "type": "text",                           # (*)
    "text": "Line one\nLine two",             # (*) string
})

• Newlines/tabs are preserved. Great for shape-probe runs (Example 2).

PDF → .pdf (wrapped text; Letter, Helvetica, 1" margins, auto-paginated)

# Block form — preferred for structure:
emit({
    "type": "pdf",                            # (*)
    "title": "Q1 Summary",                    # optional cover heading
    "blocks": [                               # (*) one of: blocks | text
        {"type": "heading",   "text": "Section 1"},
        {"type": "paragraph", "text": "A paragraph of prose…"},
        # heading | paragraph are the ONLY block kinds
    ],
})

# Prose form — one big auto-wrapped block (blank lines separate paragraphs):
emit({"type": "pdf", "title": "Q1 Summary", "text": "Long prose…"})

• Wrapped text only — no tables, no images, no inline styling. Tabular data → excel; a chart → image.

JSON → .json (read-only CodeMirror viewer)

emit({
    "type": "json",                           # (*)
    "data": <any JSON-compatible value>,      # (*) canonical field
    # "json": <…>                             # accepted as an alias for "data"
})

• data may be a dict, list, str, int, float, bool, or None.
• Do not json.dumps(...) it yourself — pass the native value; the worker serializes. Pre-encoding stores one giant escaped string.
• The worker normalizes a few shapes before storing (a JSON-encoded string is parsed once; a raw MCP result object has its structuredContent/text extracted; already-structured values are stored as-is) — so connector output rarely lands as an unreadable blob. (The Markdown→records promotion happens earlier, at mcp().call() time — see §2, not here.)

Type inference shortcut: if you omit type but include sheets, the worker treats it as excel. Any other missing/unknown type is an error. Be explicit.

4. The model — prompt() and agent()

The model is a brokered capability: the API key lives only in the worker, which attaches it, enforces the model allowlist and the run's token budget, and audits every call. Reach for it when the value is in the model's judgment (extract, classify, summarize, normalize messy data) — not for work plain Python does deterministically.

await prompt(instruction, input=None, schema=None, model=None)

• instruction is the trusted directive. input is treated strictly as DATA, never as instructions — safe to pass raw, possibly-messy connector output.
• Return value: with no schema, a string (the model's text). With a schema (a JSON-Schema dict), a parsed object validated against it — prefer this for extract/classify/enrich/route, so you get a dict/list you can drop straight into an envelope.
• model (optional): a tier alias "fast", "smart", or "default", or an allowed model id. Tiers map to:

• Raises PromptError on a disallowed model (model_not_allowed), budget exhaustion (budget_exceeded), a missing key (byok_required), or a provider error.

SCHEMA = {
    "type": "object",
    "properties": {"summary": {"type": "string"}},
    "required": ["summary"],
}
out = await prompt("Summarize this in one sentence.", input=doc, schema=SCHEMA, model="fast")
# out == {"summary": "..."}   (a dict, because schema was given)

await agent(goal, tools=None, max_steps=8, model=None, allow_writes=False)

A small in-sandbox loop: each step the model either calls one read-only tool from the connectors you pass, or finishes with a result (which agent() returns). Every reasoning step is a prompt() call (so it shares the run's budget and is audited) and every tool call is brokered (tenant-scoped, read-only).

• tools must be connector proxies, e.g. tools=[mcp("Maxio (sandbox)")].
• max_steps caps tool-call iterations (default 8). Note each step costs ~1 model call against the per-run cap (§6).
• allow_writes is a dormant no-op — the loop is always read-only (only read-looking tools are even offered to the model). It's accepted for source compatibility; leave it False.
• Returns whatever the model puts in its final result — any JSON value (usually a list or dict, but model-shaped, not guaranteed). Coerce/validate it before emitting.
• Raises AgentError on an empty goal, no tools, no read-only tools, or max_steps exhausted without finishing.

Use prompt() (with you driving the mcp() calls) when you know exactly which tools to call; reach for agent() only when the choice of tool/arguments is itself the work.

5. Connectors & the read-only guarantee

Connectors linked to the AI Tool are what your script can reach. Types:

For native connectors the read tool names above are stable; for Maxio/NetSuite (vendor-hosted) always confirm with list_tools(). Write tools (create_*/update_*, sf_create_record, gdrive_create_file, …) exist in the registries but are hidden and rejected for every AI Tool — the platform is read-only. A write call fails with a clear "AI Tools are read-only" error, and agent(allow_writes=True) is ignored.

6. Limits & quotas

Practical implications: paginate large connector pulls (Example 4); keep model use proportionate (a per-row prompt() over thousands of rows will blow the call cap/budget — batch instead); and remember a script that makes no model call needs no model key at all.

7. Examples

Each block is complete — paste it into a Script (Python) AI Tool with the named connector linked. Most use Maxio as the stand-in data source; swap in any connector + tool you have.

Example 1 — The minimal script: pull → Excel

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 20})

# Maxio nests rows as {"customers": [{"customer": {...}}, ...]}; unwrap to flat dicts.
customers = [c["customer"] for c in raw.get("customers", [])]

emit({"type": "excel", "sheets": [{"name": "Customers", "data": customers}]})

Example 2 — Shape probe (run FIRST against an unfamiliar tool)

Emits the raw response as text so you can see the keys before building a real envelope. Cheap (per_page=5) and avoids spreadsheet errors while exploring.

rows = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 5})

emit({
    "type": "text",
    "text": "type: " + type(rows).__name__
            + "\nkeys: " + str(list(rows.keys()) if isinstance(rows, dict) else "(not a dict)")
            + "\n\nraw (truncated):\n" + str(rows)[:2000],
})

Example 3 — Multi-sheet Excel with a derived summary

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 20})
customers = [c["customer"] for c in raw.get("customers", [])]

def count_with(key):
    return sum(1 for c in customers if c.get(key))

summary = [{
    "total_customers": len(customers),
    "with_email": count_with("email"),
    "with_vat_number": count_with("vat_number"),
}]

emit({
    "type": "excel",
    "sheets": [
        {"name": "Summary", "data": summary},
        {"name": "Customers", "data": customers},
    ],
})

Example 4 — Paginate until a short page

PER_PAGE, MAX_PAGES = 50, 20          # MAX_PAGES is a hard cap — each page is a real MCP call

all_customers = []
for page in range(1, MAX_PAGES + 1):
    resp = await mcp("Maxio (sandbox)").call(
        "advancedBillingShowCustomers", {"per_page": PER_PAGE, "page": page}
    )
    rows = [c["customer"] for c in resp.get("customers", [])]
    if not rows:
        break
    all_customers.extend(rows)
    if len(rows) < PER_PAGE:           # a short page means we've reached the end
        break

emit({"type": "excel", "sheets": [{"name": "Customers", "data": all_customers}]})

Example 5 — JSON output (passthrough and reshaped)

import datetime as dt

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 20})
customers = [c["customer"] for c in raw.get("customers", [])]

emit({
    "type": "json",
    "data": {
        "customers": customers,                       # the native value — DON'T json.dumps it
        "meta": {
            "count": len(customers),
            "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
            "source": "Maxio (sandbox)/advancedBillingShowCustomers",
        },
    },
})

Example 6 — A native connector with SOQL (Salesforce)

Native connectors use stable tool names. Salesforce queries with SOQL via sf_soql_query; the result's records is your row list.

res = await mcp("salesforce").call(
    "sf_soql_query",
    {"soql": "SELECT Id, Name, Industry, AnnualRevenue FROM Account ORDER BY Name LIMIT 50"},
)
rows = res.get("records", res if isinstance(res, list) else [])

emit({"type": "excel", "sheets": [{"name": "Accounts", "data": rows}]})

QuickBooks is similar: await mcp("QuickBooks Online").call("list_customers", {}), or run a query with qbo_query. Xero: list_invoices, get_report_profit_and_loss, etc. Unsure of a tool's name or args? Call await mcp(name).list_tools().

Example 7 — Discover what's linked, then pull

emit({
    "type": "json",
    "data": {
        "connectors": list_connectors(),
        # tools for the first connector, if any
        "tools": [
            {"name": t["name"], "description": t.get("description", "")}
            for t in (await mcp(list_connectors()[0]["name"]).list_tools())
        ] if list_connectors() else [],
    },
})

Example 8 — prompt() for structured extraction → Excel

Hand messy connector output to the model and get back clean, typed rows. With a schema, prompt() returns a parsed dict — no prose to clean up.

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 10})

SCHEMA = {
    "type": "object",
    "properties": {
        "rows": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "id": {"type": ["string", "integer"]},
                    "name": {"type": "string"},
                    "email": {"type": "string"},
                    "organization": {"type": "string"},
                },
            },
        }
    },
    "required": ["rows"],
}

result = await prompt(
    "Extract a 'rows' array of customers from the data, each with id, name, email, "
    "and organization. Use empty strings for missing fields.",
    input=raw,
    schema=SCHEMA,
    model="fast",
)

emit({"type": "excel", "sheets": [{"name": "Customers", "data": result["rows"]}]})

Example 9 — prompt() for judgment (classify / enrich rows)

Where the value is the model's judgment, not a lookup. Classify each customer's likely segment, then ship Excel. (One batched call over all rows — not one call per row — to stay well under the call cap and budget.)

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 25})
customers = [c["customer"] for c in raw.get("customers", [])]
slim = [{"id": c.get("id"), "organization": c.get("organization"), "email": c.get("email")}
        for c in customers]

SCHEMA = {
    "type": "object",
    "properties": {
        "rows": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "id": {"type": ["string", "integer"]},
                    "segment": {"type": "string", "enum": ["smb", "mid_market", "enterprise", "unknown"]},
                    "reason": {"type": "string"},
                },
                "required": ["id", "segment"],
            },
        }
    },
    "required": ["rows"],
}

labels = await prompt(
    "Classify each customer's segment (smb, mid_market, enterprise, or unknown) "
    "from the organization/email signals. Return one row per input id.",
    input=slim,
    schema=SCHEMA,
    model="fast",
)

by_id = {str(r["id"]): r for r in labels["rows"]}
enriched = [{**c, "segment": by_id.get(str(c.get("id")), {}).get("segment", "unknown")}
            for c in customers]

emit({"type": "excel", "sheets": [{"name": "Customers", "data": enriched}]})

Example 10 — agent() lets the model drive the tool calls

result = await agent(
    goal="List the first 10 Maxio customers as rows with id, organization, and email.",
    tools=[mcp("Maxio (sandbox)")],   # read-only tools only
    max_steps=6,
    model="fast",
)

# agent() returns whatever the model put in its final result — coerce to a row list.
rows = (result if isinstance(result, list)
        else result.get("rows", []) if isinstance(result, dict)
        else [])

emit({"type": "excel", "sheets": [{"name": "Customers", "data": rows}]})

Example 11 — HTML report (build a table yourself)

import html

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 15})
customers = [c["customer"] for c in raw.get("customers", [])]
cols = ["id", "organization", "email"]

head = "".join(f"<th>{html.escape(c)}</th>" for c in cols)
body = "".join(
    "<tr>" + "".join(f"<td>{html.escape(str(c.get(k, '')))}</td>" for k in cols) + "</tr>"
    for c in customers
)
doc = f"""
<style>
  body {{ font: 14px system-ui, sans-serif; }}
  table {{ border-collapse: collapse; width: 100%; }}
  th, td {{ border: 1px solid #ddd; padding: 6px 10px; text-align: left; }}
  th {{ background: #f5f5f5; }}
</style>
<h1>Customers ({len(customers)})</h1>
<table><thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>
"""
emit({"type": "html", "html": doc})

Example 12 — SVG chart (image, composed in pure Python)

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 50})
customers = [c["customer"] for c in raw.get("customers", [])]

# Count customers per organization (top 8).
from collections import Counter
top = Counter((c.get("organization") or "—") for c in customers).most_common(8)
maxv = max((n for _, n in top), default=1)

bar_h, gap, pad, width = 24, 10, 160, 640
height = pad + len(top) * (bar_h + gap)
bars = []
for i, (label, n) in enumerate(top):
    y = pad + i * (bar_h + gap)
    w = int((width - pad - 60) * n / maxv)
    bars.append(
        f'<text x="8" y="{y + bar_h - 7}" font-size="12">{label[:22]}</text>'
        f'<rect x="{pad}" y="{y}" width="{w}" height="{bar_h}" fill="#4f46e5"/>'
        f'<text x="{pad + w + 6}" y="{y + bar_h - 7}" font-size="12">{n}</text>'
    )
svg = (f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" '
       f'font-family="system-ui, sans-serif">{"".join(bars)}</svg>')

emit({"type": "image", "format": "svg", "data": svg})

Example 13 — PDF summary (blocks)

import statistics as stats

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 50})
customers = [c["customer"] for c in raw.get("customers", [])]
with_email = sum(1 for c in customers if c.get("email"))

emit({
    "type": "pdf",
    "title": "Customer Snapshot",
    "blocks": [
        {"type": "heading",   "text": "Overview"},
        {"type": "paragraph", "text": f"{len(customers)} customers sampled; "
                                      f"{with_email} have an email on file "
                                      f"({round(100 * with_email / max(len(customers), 1))}%)."},
        {"type": "heading",   "text": "Notes"},
        {"type": "paragraph", "text": "Generated by a Python AI Tool. For the full "
                                      "row-level data, use an Excel output instead."},
    ],
})

Example 14 — Robust error handling

# A failed MCP tool call raises a generic exception (NOT McpError) — catch broadly.
try:
    raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 20})
except Exception as e:
    emit({"type": "text", "text": f"Connector call failed: {e}"})
else:
    customers = [c["customer"] for c in raw.get("customers", [])]

    # The model path raises PromptError specifically.
    try:
        out = await prompt("One-line health note about this customer set.",
                           input={"count": len(customers)}, model="fast")
    except PromptError as e:
        out = f"(model unavailable: {e})"

    emit({
        "type": "excel",
        "sheets": [
            {"name": "Note", "data": [{"note": out}]},
            {"name": "Customers", "data": customers},
        ],
    })

Example 15 — Pure-Python value-add (no model, no extra calls)

The cheapest, most deterministic reports are plain Python over connector data.

import statistics as stats

raw = await mcp("Maxio (sandbox)").call("advancedBillingShowCustomers", {"per_page": 100})
customers = [c["customer"] for c in raw.get("customers", [])]

from collections import Counter
by_org = Counter((c.get("organization") or "—") for c in customers)

report = [{"organization": org, "customers": n} for org, n in by_org.most_common()]
emit({
    "type": "excel",
    "sheets": [
        {"name": "By Organization", "data": report},
        {"name": "Customers", "data": customers},
    ],
})

8. Debugging & best practices

• Probe unfamiliar tools with the Example 2 text dump before building an Excel envelope — the row list is almost always nested under a key.
• await everything — mcp(...).call(...), prompt(...), agent(...), list_tools(). A missing await leaves a coroutine object that breaks emit() with "not JSON serializable".
• Inspect a prompt() result by emitting it as JSON: emit({"type": "json", "data": result}). With no schema you'll see a string; with a schema, the parsed object.
• Normalize ragged rows before Excel — columns come from the first row's keys only.
• Don't json.dumps() envelope data — pass native values; the worker serializes.
• Batch model calls. One prompt() over a list beats one call per row — the per-run call cap (50) and token budget (300k) are easy to exhaust otherwise.
• print() for dev only — it lands in worker logs, never the report. The report is exactly your last emit().
• No input parameters reach Python scripts yet — use a Prompt/LSPL AI Tool if you need user-supplied values at run time.

Common errors and what they mean