I run seven n8n workflows on a small VPS. Together they have 82 nodes, and most of those are HTTP Request nodes with the same retry settings and the same three credentials. Built by hand, that is the same options set again and again, and a change to a shared setting means opening every node.
So I do not build them in the editor. One Python file, build.py, generates all seven workflow JSON files. I push them to n8n through its public REST API. I use the n8n UI to create credentials and to read executions.
This post shows the parts of that script that matter, the API calls, and the gotchas.
A workflow is just four keys
An n8n workflow, as the API accepts it, is a JSON object with a name, a list of nodes, a connections map and a settings object. My script builds exactly that:
def workflow(name, nodes, conns, error_wf=None, timezone=None):
for node in nodes:
if node["type"] != "n8n-nodes-base.code":
node["parameters"] = braces(node["parameters"])
settings = {"executionOrder": "v1", "saveDataErrorExecution": "all", "saveDataSuccessExecution": "all"}
if error_wf:
settings["errorWorkflow"] = error_wf
if timezone:
settings["timezone"] = timezone
return {"name": name, "nodes": nodes, "connections": conns, "settings": settings}Ignore braces() for a moment. I come back to it, because it is the most important line in the file.
One small helper per node type
Each node is a dict with an id, a name, a type, a typeVersion, a canvas position and parameters. Each kind of node gets a helper.
A Code node and a schedule trigger are short:
def nid():
return str(uuid.uuid4())
def code(name, pos, js, **node):
return {"id": nid(), "name": name, "type": "n8n-nodes-base.code", "typeVersion": 2,
"position": pos, "parameters": {"jsCode": js.strip()}, **node}
def schedule(pos, *rules):
return {"id": nid(), "name": "Schedule", "type": "n8n-nodes-base.scheduleTrigger",
"typeVersion": 1.2, "position": pos, "parameters": {"rule": {"interval": list(rules)}}}The HTTP Request helper does most of the work. This is a trimmed version. The real one also handles headers, binary bodies, multipart bodies and file responses:
def http(name, pos, method, url, cred=None, json_body=None, query=None, timeout=None, **node):
p = {"method": method, "url": url, "options": {}}
if cred:
p["authentication"] = "predefinedCredentialType"
p["nodeCredentialType"] = CREDS[cred][0]
if query:
p["sendQuery"] = True
p["queryParameters"] = {"parameters": [{"name": k, "value": v} for k, v in query.items()]}
if json_body:
p.update(sendBody=True, contentType="json", specifyBody="json", jsonBody=json_body)
if timeout:
p["options"]["timeout"] = timeout
out = {"id": nid(), "name": name, "type": "n8n-nodes-base.httpRequest", "typeVersion": 4.2,
"position": pos, "parameters": p}
if cred:
out["credentials"] = {CREDS[cred][0]: CREDS[cred][1]}
out.update(node)
return outThe trailing **node is what makes this pay off. Node-level settings such as retryOnFail, maxTries, waitBetweenTries and onError sit next to parameters, not inside it, so I pass them as keyword arguments:
def drive_list(name, pos, q, fields, page_size, cred="drive"):
return http(name, pos, "GET", DRIVE_API, cred, query={
"q": q, "orderBy": "createdTime", "pageSize": str(page_size), "fields": f"files({fields})",
}, retryOnFail=True, maxTries=3, waitBetweenTries=5000)Every Google Drive listing in every workflow now has the same retry policy.
Credentials are referenced, never stored
The script holds no secrets. Credentials are created once in the n8n UI. A node only refers to one by type, ID and name:
CREDS = {
"drive": ("googleDriveOAuth2Api", {"id": "<credential id>", "name": "Google Drive (ClipStack)"}),
"youtube": ("youTubeOAuth2Api", {"id": "<credential id>", "name": "YouTube (ClipStack)"}),
"gmail": ("gmailOAuth2", {"id": "<credential id>", "name": "Gmail (ClipStack alerts)"}),
}The generated JSON files are committed to git. They contain those credential IDs and names, and no tokens.
Connections in one line
The connections map is verbose by hand. A node name maps to main, which is a list of outputs, and each output is a list of targets. This helper takes plain tuples:
def connect(*links):
"""links: (from, to) or (from, output_index, to)."""
conns = {}
for link in links:
src, out, dst = (link[0], 0, link[1]) if len(link) == 2 else link
outputs = conns.setdefault(src, {"main": []})["main"]
while len(outputs) <= out:
outputs.append([])
outputs[out].append({"node": dst, "type": "main", "index": 0})
return connsThe three-item form picks an output index. For an IF node, output 0 is the true branch and 1 is the false branch. For a node with onError="continueErrorOutput", output 1 is the error branch.
Connections use node names, not IDs. That is why I can generate a fresh UUID for every node on every build without breaking anything.
The gotcha: n8n ends an expression at the first }}
This is the gotcha that shaped the script the most. An n8n expression is ={{ ... }}. n8n ends the expression at the first }} it sees. If the JavaScript inside builds a nested object, the object’s own closing braces form a }} and cut the expression short.
This is a normal thing to write for a JSON body:
json_body="={{ JSON.stringify({appProperties: {acStatus: 'done'}}) }}"The }} after 'done' closes the expression early. The fix is a space between the two braces, } }. I did not want to remember that in every node, so the script does it for every parameter of every node:
def braces(value):
"""n8n ends an expression at the first '}}', so nested JS objects need '} }'."""
if isinstance(value, dict):
return {k: braces(v) for k, v in value.items()}
if isinstance(value, list):
return [braces(v) for v in value]
if not (isinstance(value, str) and value.startswith("=") and "{{" in value):
return value
out, rest = "=", value[1:]
while "{{" in rest:
head, rest = rest.split("{{", 1)
depth, i = 0, 0
while i < len(rest): # the expression closes at the '}}' that balances its braces
if rest[i] == "{":
depth += 1
elif rest[i] == "}":
if depth == 0 and rest[i:i + 2] == "}}":
break
depth -= 1
i += 1
body = rest[:i]
while "}}" in body or "{{" in body:
body = body.replace("}}", "} }").replace("{{", "{ {")
out += head + "{{" + body + "}}"
rest = rest[i + 2:]
return out + restIt walks every string that starts with =, finds each expression by counting brace depth, and spaces out any doubled braces inside the body. Code nodes are skipped in workflow(), because jsCode is plain JavaScript and not an expression.
There is a second brace problem that is pure Python. Inside an f-string, a literal {{ has to be written as {{{{. So a URL with an expression looks like this in the source:
http("Claim", [660, 0], "PATCH", f"={DRIVE_API}/{{{{ $json.id }}}}", "drive", ...)It is ugly, but it is mechanical.
Canvas position decides run order
I set "executionOrder": "v1" on every workflow. With that setting, when a node fans out to two branches, the branch that sits higher on the canvas runs first. In a generated workflow, that makes the position argument part of the logic.
One of my workflows uploads a batch of files on one branch and marks the batch as done on another. The “mark done” node must run last. The script places it lower on the canvas for that reason, and says so:
# Lower on the canvas, so with v1 order it runs after every clip above is uploaded.
http("Mark done", [1320, 260], "PATCH", ...)Settings at the top, baked into Code nodes
Tunable values are constants at the top of the file. Code nodes cannot read Python variables, so I put a placeholder in the JavaScript and replace it at build time:
status: {privacyStatus: '%PRIVACY%', selfDeclaredMadeForKids: false},
""".replace("%PRIVACY%", PRIVACY)),For values that are not simple words I use json.dumps() in the replace call, so the JavaScript gets a properly quoted literal.
Build and deploy
The script takes a small JSON file of folder IDs, an output directory and the ID of the error workflow:
usage: python -I build.py <folders.json> <out_dir> [errorWorkflowId]It writes one JSON file per workflow. Deploying uses the n8n public API. Create an API key in n8n under Settings → n8n API, and send it in the X-N8N-API-KEY header. The base URL is https://n8n.example.com/api/v1. I use three calls:
POST /workflowscreates a workflow and returns its ID.POST /workflows/{id}/activateturns it on.PUT /workflows/{id}replaces an existing workflow.
I keep the returned IDs in a plain ids.txt, one slug id pair per line, so the next deploy knows where to PUT.
The API details are in the n8n public API docs.
Deploy gotchas
The PUT body takes only name, nodes, connections and settings. That is why workflow() returns exactly those four keys and nothing else.
A PUT is not the last step. After every update I deactivate the workflow and activate it again, so that the new version goes live.
The error workflow has to exist first. The other six workflows point at it through settings.errorWorkflow, and that value is its ID. So I create the error workflow, note its ID, and pass it as the third argument when I build the rest.
Scheduled workflows are awkward to test. To test one, I deployed a throwaway copy with a webhook trigger in place of the schedule, so I could run it on demand and check the result.
Was it worth it
For one or two workflows, no. The editor is faster. For seven workflows that share credentials, retry rules and upload logic, yes. A change to the retry policy is one line. A review is a git diff. And a new workflow that reuses an existing chain of nodes is mostly copy and paste in Python, with the brace handling done for me.