Build modules

A module is a folder modules/<module>/v<N>/ with three handler functions. Each file defines one function with the same name:

config.py holds the module settings and context.md the context. Add a module with stacksync workflow-module create "<name>".

One module per operationCopy Anchor Link

Build one module per operation and let the user choose the object in a dropdown inside it. This is the rule Stacksync reviewers check first.

modules/
  create_records/    the object is a dropdown: Contact, Company, Deal, custom objects
  update_records/
  upsert_records/
  delete_records/
  get_records/
  list_records/
  send_email/        an action that is not record work gets its own module

• Never build create_contact, create_company, create_deal.
• The object dropdown lists every standard object and the account's custom objects. Fetch them from the API's metadata endpoint when it has one.
• Choosing the object rebuilds the form with that object's fields.
• Never special-case one object with its own code path, and never replace real fields with a free-text JSON body field.
• Put code that several modules use, such as the API client, in utils/ and import it with from utils.api_client import ....

How the form worksCopy Anchor Link

1. The step opens. Stacksync calls /schema and shows the connection picker and the object dropdown.
2. The user picks a connection. Stacksync calls /content for the dropdown options.
3. The user picks an object. The field has on_action: {"load_schema": true}, so Stacksync calls /schema again with the choice in form_data.
4. The form shows that object's fields. When the workflow runs, Stacksync calls /execute with the values and the connection.

schema()Copy Anchor Link

Return the full form each time. It replaces the current one.

from stacksync_cdk import Request, SchemaResponse

BASE_SCHEMA = {
    "metadata": {"workflows_module_schema_version": "1.0.0"},
    "fields": [
        {
            "id": "select_object",
            "type": "string",
            "label": "Select object",
            "validation": {"required": True},
            "ui_options": {"ui_widget": "SelectWidget"},
            "content": {"content_objects": [{"id": "objects"}]},
            "on_action": {"load_schema": True},
        }
    ],
}


def schema(request: Request) -> SchemaResponse:
    selected_object = request.form_data.get("select_object")
    if not selected_object:
        return SchemaResponse.from_schema(BASE_SCHEMA)

    item_fields = fields_for(selected_object, request.credentials)
    return SchemaResponse.from_schema(BASE_SCHEMA).add_record_array(selected_object, item_fields)

• SchemaResponse.from_schema(dict) takes a whole form. SchemaResponse(fields, connections=None, ui_options=None, metadata=None) builds one from a list of fields.
• add_record_array(object_name, item_fields, min_items=1, label=None) adds an array field "<object_name>s" whose rows hold item_fields, so one run can handle several records. For contact, execute() receives contacts.
• Every field needs id and type.
• Don't add a connection picker. The CDK adds it from connection_app_type.
• Never raise because the connection is missing. Return the base form.
• When the fields come from your API, set requires_credentials_for_schema=True in config.py.
• Keep schema() deterministic: the same form_data and connection give the same form.

Field typesCopy Anchor Link

Use "type": "string" for almost every field, including numbers, dates and dropdowns. Users can then type a value or a reference to an earlier step, such as {{ trigger_1.data.amount }}. Convert the strings to real types in execute(). Only booleans, arrays and objects keep their own type.

Rules behind the table:

• Dropdowns use "validation": {"pattern": ".*"}, never enum. An enum rejects a value typed from an earlier step. Add "required": True next to the pattern when needed.
• An option is always {"value": ..., "label": ...}.
• Booleans need "disable_replacement": True to show as a checkbox.
• Arrays and objects need ui_options.ui_order.
• Leave out validation for optional fields.
• For update modules, make every field optional and show booleans as a dropdown with empty, true and false, so an untouched field never overwrites data. For upsert modules, add an optional id field that matches existing records.

Reloading on changeCopy Anchor Link

A field can have both. Example of a dropdown that depends on another:

{"id": "schema_name", "type": "string", "ui_options": {"ui_widget": "SelectWidget"},
 "content": {"content_objects": [{"id": "schemas"}]},
 "on_action": {"load_content": True}, "reload_content_objects": [{"id": "tables"}],
 "validation": {"pattern": ".*"}}

content()Copy Anchor Link

from stacksync_cdk import ContentResponse, Request

OBJECTS = [{"value": "contact", "label": "Contact"}, {"value": "company", "label": "Company"}]


def content(request: Request) -> ContentResponse:
    if not request.credentials:
        return ContentResponse()

    content_objects = {}
    for name in request.content_object_names:
        if name == "objects":
            content_objects["objects"] = OBJECTS + list_custom_objects(request.credentials)
        elif name == "owners":
            content_objects["owners"] = list_owners(request.credentials)
    return ContentResponse(content_objects)

• request.content_object_names lists the dropdown ids to fill in this call. Fill only those.
• Return ContentResponse({id: [{"value": ..., "label": ...}]}). preview is optional. The CDK adds the rest of the answer.
• value is what execute() receives. Keep it a stable id, not a display name.
• For long lists, return one page and mark that more exist: ContentResponse().add("owners", options, has_next_page=True).
• Never raise. Without a connection, or when your API fails, return no options for that dropdown so the form still works. Set a timeout on every call.

execute()Copy Anchor Link

from stacksync_cdk import ExecuteResponse, ManagedError, Request


def execute(request: Request) -> ExecuteResponse:
    selected_object = request.data.get("select_object")
    if not selected_object:
        raise ManagedError.validation_error("select_object is required.")

    records = request.data.get(f"{selected_object}s") or []
    if not records:
        raise ManagedError.validation_error(f"At least one {selected_object} is required.")

    results = [create_one(selected_object, to_api(record), request.credentials) for record in records]
    successful = sum(1 for result in results if result["status"] == "success")
    return ExecuteResponse(
        {"total": len(results), "successful": successful, "failed": len(results) - successful, "results": results},
        metadata={"affected_records": successful},
    )

• request.data holds the step's values, with references to earlier steps already resolved.
• Validate required inputs first, with clear messages.
• Convert strings back to real types (numbers, "true" and "false", comma-separated lists) and drop empty values before calling your API.
• For several records, report each record's result and keep going when one fails.
• ExecuteResponse(data, metadata=None, data_type=None). Later steps read data as {{ MODULE.<step id>.output }}, so keep its shape predictable. Returning a plain value also works: it becomes data.
• Set a timeout on every call. Retry reads, and a 429 after its Retry-After. Never retry a create, update or delete after a server error, because the first call may have succeeded.
• Keep no state between calls: no module-level caches, no global variables that change, no files on disk. The connector runs as several processes.
• Do nothing at import time: no network calls and no reading of the environment.

What Stacksync sendsCopy Anchor Link

Each handler gets a Request:

The bodies Stacksync sends:

// POST /<module>/v<N>/schema
{ "data": { "form_data": { "select_object": "contact" } }, "credentials": { … } }

// POST /<module>/v<N>/content
{ "data": { "form_data": { … }, "content_object_names": [ { "id": "objects" } ] }, "credentials": { … } }

// POST /<module>/v<N>/execute
{ "data": { "select_object": "contact", "contacts": [ { "name": "Ada" } ] },
  "credentials": { … },
  "workspace_id": "…", "workflow_id": "…", "workflow_module_instance_id": "…" }

The answers the CDK builds from your return values:

// schema
{ "data": { "schema": { "metadata": { … }, "fields": [ … ], "connections": { … } } } }

// content
{ "data": { "content_objects": [ { "id": "objects", "content_object_name": "objects",
    "data": [ { "value": "contact", "label": "Contact", "index": 0 } ],
    "pagination": { "data": {}, "has_next_page": false } } ] } }

// execute
{ "data": { … }, "metadata": { "affected_records": 1 } }

Connections and credentialsCopy Anchor Link

connection_app_type in stacksync.yml sets the connection users choose for every module. The CDK adds the connection picker to each form, and Stacksync passes the chosen connection with each call. Users create connections in Stacksync under Connections. Credentials never appear in your code or settings.

No other value is accepted.

Reading credentials:

token = request.credentials.get("api_credentials")       # generic_api_credentials
api_key = request.credentials.api_key                    # api_key or api_key_bearer
access_token = request.credentials.access_token          # OAuth: already refreshed by Stacksync
dsn = request.credentials.connection_string              # database connections
instance_url = request.credentials.get("instance_url")   # any other field by name

• if request.credentials: tells you whether a connection is selected.
• request.credentials.value is the whole credential object. .token_type, .expires_at, .connection_app_type and .connection_id are also available.
• Read exactly the keys of your connection type. A key that is always None means the code reads a name that the type doesn't have: with generic_api_credentials, everything is in api_credentials.
• When a key is missing, raise an error that shows the expected shape with a fake value, so the user knows what to fix:

if not request.credentials.get("api_credentials"):
    raise ManagedError.validation_error(
        "This connection has no API key. Create a generic_api_credentials connection whose value looks like this:\n"
        '{"api_credentials": "your-api-key"}'
    )

• Never log credentials, return them, or put them in an error message.
• Describe the connection and the exact keys you read in the ## Connection section of every context.md.

ErrorsCopy Anchor Link

Raise ManagedError for every failure the user can act on. Its message is shown in the step's result in Stacksync.

The answer is {"error": "<message>"} with the status, plus data and metadata when you pass them. For example, a missing input answers 400 with {"error": "select_object is required."}. Any other exception becomes a 500 with the exception's text, so don't wrap whole handlers in try/except only to hide errors, and never report success when the API failed.

Next stepsCopy Anchor Link