Script examples you can adapt

def execute(input, context):
    reference = input["trigger_value"].strip().upper()
    return {"values": {"CustomerReference": reference}}

def execute(input, context):
    amount = input["amount"]
    if isinstance(amount, bool) or not isinstance(amount, (int, float)):
        raise ValueError("amount must be a number")
    if amount < 0:
        raise ValueError("amount must not be negative")
    return {"requires_review": amount >= 10000}

def execute(input, context):
    event_type = input["event_type"]
    entity_id = input["entity_id"]
    return {
        "acknowledged": True,
        "message": f"Received {event_type} for entity {entity_id}"
    }

import re
from pypdf import PdfReader, PdfWriter

def execute(input, context):
    api = context["api"]
    reader = PdfReader(BytesIO(api.get_document_content(input["document_id"])))
    parts = []
    for page in reader.pages:
        found = re.search(r"\b(\d{6})-(\d{4})\b", page.extract_text() or "")
        if found:
            parts.append({"id": found.group(1) + found.group(2), "writer": PdfWriter()})
        if parts:
            parts[-1]["writer"].add_page(page)

    created = []
    for part in parts:
        buffer = BytesIO()
        part["writer"].write(buffer)
        data = buffer.getvalue()
        upload = api.initialize_upload(f"{part['id']}.pdf", "application/pdf", len(data))
        for offset in range(0, len(data), upload["chunkSize"]):
            api.upload_chunk(upload["uploadId"], data[offset:offset + upload["chunkSize"]], offset)
        doc = api.create_document(upload["uploadId"], input["document_type_id"], keywords={
            "standalone": [{"keywordTypeId": input["keyword_type_id"], "value": part["id"]}],
        })
        created.append(doc["documentId"])
    return {"created": created}

from pyhanko.keys import load_certs_from_pemder_data
from pyhanko.pdf_utils.reader import PdfFileReader
from pyhanko.sign.validation import validate_pdf_signature
from pyhanko.sign.validation.status import SignatureCoverageLevel
from pyhanko_certvalidator import ValidationContext

def execute(input, context):
    data = context["api"].get_document_content(input["document_id"])
    signatures = PdfFileReader(BytesIO(data)).embedded_signatures
    if not signatures:
        return {"accepted": False, "reason": "not signed"}
    roots = list(load_certs_from_pemder_data(context["secrets"]["Issuer_CA_Certificate"].encode()))
    status = validate_pdf_signature(signatures[0], ValidationContext(trust_roots=roots))
    email = signatures[0].signer_cert.subject.native.get("email_address") or ""
    accepted = (status.bottom_line
                and status.coverage == SignatureCoverageLevel.ENTIRE_FILE
                and email.endswith("@issuer.example"))
    return {"accepted": accepted, "signer": email}

def execute(input, context):
    """input: {"old_value": str, "new_value": str}"""
    api = context["api"]
    set_id = next(s["id"] for s in api.get_autofill_keyword_sets() if s["name"] == "EmployeeData")
    primary = api.get_autofill_keyword_set(set_id)["primaryKeywordTypeId"]

    found = api.search_autofill_entries(set_id, keyword_type_id=primary, value=input["old_value"])
    for entry in found["items"]:
        # Create the new entry first, so a failed create leaves the original in place.
        api.create_autofill_entry(set_id, {**entry["values"], primary: input["new_value"]})
        api.delete_autofill_entry(set_id, entry["id"])
    return {"moved": len(found["items"])}

{ document_id: $document.id, test_run: $instance.choice == "test" }

LETTER_TYPE_ID = 212   # the Letter document type
BATCH_NO_ID = 301      # BatchNo: the batch a letter belongs to, also on the batch document
SIGNED_DATE_ID = 302   # SignedDate: filled in when the recipient signs


def keyword_value(api, document_id, keyword_id):
    for keyword in api.get_document_keywords(document_id)["keywords"]["standalone"]:
        if keyword["keywordId"] == keyword_id and keyword["values"]:
            return keyword["values"][0]["value"]
    return None


def batch_number(api, document_id):
    batch = keyword_value(api, document_id, BATCH_NO_ID)
    if batch is None:
        raise ValueError(f"Document {document_id} has no BatchNo")
    return batch


def letters(api, batch, signed):
    return api.search_documents(
        document_type_id=LETTER_TYPE_ID,
        keywords=[
            {"id": BATCH_NO_ID, "operator": "Equal", "value": batch},
            {"id": SIGNED_DATE_ID, "operator": "IsNotNull" if signed else "IsNull"},
        ],
        max_results=10000,
    )["results"]


def execute(input, context):
    """input: {"document_id": int, "test_run": bool}"""
    api = context["api"]
    batch = batch_number(api, input["document_id"])
    unsigned = [row["documentInfo"]["documentId"] for row in letters(api, batch, signed=False)]
    # By document ID: a letter signed between the two searches is in both.
    kept = {row["documentInfo"]["documentId"] for row in letters(api, batch, signed=True)}

    if input.get("test_run"):
        summary = f"Test run for batch {batch}. Unsigned letters that would be deleted: {len(unsigned)}."
    else:
        deleted = 0
        for document_id in unsigned:
            # A recipient can sign while the run is going: read the letter again just before deleting it.
            if document_id in kept or keyword_value(api, document_id, SIGNED_DATE_ID) is not None:
                kept.add(document_id)
                continue
            api.delete_document(document_id)
            deleted += 1
        summary = f"Batch {batch} withdrawn. Unsigned letters deleted: {deleted}."
    summary += f" Signed letters kept: {len(kept)}."
    return {"summary": summary, "level": "warning" if kept else "success"}