fix(web-app-n8n): page through API keys and workflows when provisioning

n8n 2 returns ten API keys and a hundred workflows per page by default. A
stale managed key behind the first page survived, and a managed workflow
behind it was created a second time. Both scripts now follow every page,
narrowed by label or name, and keep the exact match in the script.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

roles/web-app-n8n/files/python/provision/client.py

@@ -39,6 +39,7 @@ PROVIDERS = json.loads(os.environ.get("N8N_MCP_PROVIDERS", "[]"))
 
 REST = "/rest"
 PUBLIC = "/api/v1"
+PAGE = 250
 CREDENTIAL_TYPE = "httpBearerAuth"
 BEARER_AUTH = "bearerAuth"
 CLIENT_PREFIX = "infinito:mcp-client:"
@@ -137,6 +138,48 @@ def record(body, what):
     return created
 
 
+def api_keys_labelled(label):
+    """Return every API key whose label n8n matches against ``label``, all pages.
+
+    Args:
+        label: text n8n looks for inside each label.
+    """
+    found = []
+    while True:
+        query = urllib.parse.urlencode(
+            {"label": label, "take": PAGE, "skip": len(found)}
+        )
+        status, body = call(f"{REST}/api-keys?{query}")
+        if status != 200:
+            sys.exit(f"FAILED listing api keys: {status} {body}")
+        page = records(body, "api keys")
+        found.extend(page)
+        if len(page) < PAGE:
+            return found
+
+
+def workflows_named(key, name):
+    """Return every workflow n8n matches against ``name``, following ``nextCursor``.
+
+    Args:
+        key: the public-API key.
+        name: workflow name n8n filters on.
+    """
+    found = []
+    query = {"name": name, "limit": PAGE}
+    while True:
+        status, body = call(
+            f"{PUBLIC}/workflows?{urllib.parse.urlencode(query)}", api_key=key
+        )
+        if status != 200:
+            sys.exit(f"FAILED listing workflows: {status} {body}")
+        found.extend(records(body, "workflows"))
+        cursor = body.get("nextCursor") if isinstance(body, dict) else None
+        if not cursor:
+            return found
+        query = {"name": name, "cursor": cursor}
+
+
 def login():
     status, body = call(
         f"{REST}/login",
@@ -153,17 +196,11 @@ def api_key():
     n8n returns the usable secret only once, as ``rawApiKey``, and a listed key
     is redacted, so the key cannot be carried between runs. It is minted here
     and revoked before the run ends; an entry left behind under the managed name
-    belongs to a run that died and is deleted rather than guessed at. n8n pages
-    the listing at ten keys and matches ``label`` as a substring, so the query
-    narrows to the managed name and the exact match happens here.
+    belongs to a run that died and is deleted rather than guessed at. n8n matches
+    ``label`` as a substring, so the exact match happens here.
     """
-    query = urllib.parse.urlencode({"label": KEY_NAME})
-    status, body = call(f"{REST}/api-keys?{query}")
-    if status != 200:
-        sys.exit(f"FAILED listing api keys: {status} {body}")
-
     matches = [
-        key for key in records(body, "api keys") if key.get("label") == KEY_NAME
+        key for key in api_keys_labelled(KEY_NAME) if key.get("label") == KEY_NAME
     ]
     if len(matches) > 1:
         sys.exit(f"FAILED: {len(matches)} api keys named {KEY_NAME}")
@@ -364,11 +401,10 @@ def managed_workflow(key):
     Args:
         key: the public-API key.
     """
-    status, body = call(f"{PUBLIC}/workflows", api_key=key)
-    if status != 200:
-        sys.exit(f"FAILED listing workflows: {status} {body}")
     matches = [
-        flow for flow in records(body, "workflows") if flow.get("name") == WORKFLOW_NAME
+        flow
+        for flow in workflows_named(key, WORKFLOW_NAME)
+        if flow.get("name") == WORKFLOW_NAME
     ]
     if len(matches) > 1:
         sys.exit(f"FAILED: {len(matches)} workflows named {WORKFLOW_NAME}")

roles/web-app-n8n/files/python/provision/mcp.py

@@ -33,6 +33,7 @@ CREDENTIAL_NAME = os.environ.get("N8N_MCP_CREDENTIAL", "")
 
 REST = "/rest"
 PUBLIC = "/api/v1"
+PAGE = 250
 TRIGGER_TYPE = "@n8n/n8n-nodes-langchain.mcpTrigger"
 TRIGGER_VERSION = 2
 TRIGGER_NAME = "Infinito MCP Trigger"
@@ -142,6 +143,48 @@ def record(body, what):
     return created
 
 
+def api_keys_labelled(label):
+    """Return every API key whose label n8n matches against ``label``, all pages.
+
+    Args:
+        label: text n8n looks for inside each label.
+    """
+    found = []
+    while True:
+        query = urllib.parse.urlencode(
+            {"label": label, "take": PAGE, "skip": len(found)}
+        )
+        status, body = call(f"{REST}/api-keys?{query}")
+        if status != 200:
+            sys.exit(f"FAILED listing api keys: {status} {body}")
+        page = records(body, "api keys")
+        found.extend(page)
+        if len(page) < PAGE:
+            return found
+
+
+def workflows_named(key, name):
+    """Return every workflow n8n matches against ``name``, following ``nextCursor``.
+
+    Args:
+        key: the public-API key.
+        name: workflow name n8n filters on.
+    """
+    found = []
+    query = {"name": name, "limit": PAGE}
+    while True:
+        status, body = call(
+            f"{PUBLIC}/workflows?{urllib.parse.urlencode(query)}", api_key=key
+        )
+        if status != 200:
+            sys.exit(f"FAILED listing workflows: {status} {body}")
+        found.extend(records(body, "workflows"))
+        cursor = body.get("nextCursor") if isinstance(body, dict) else None
+        if not cursor:
+            return found
+        query = {"name": name, "cursor": cursor}
+
+
 def login():
     status, body = call(
         f"{REST}/login",
@@ -158,17 +201,11 @@ def api_key():
     n8n returns the usable secret only once, as ``rawApiKey``, and a listed key
     is redacted, so the key cannot be carried between runs. It is minted here
     and revoked before the run ends; an entry left behind under the managed name
-    belongs to a run that died and is deleted rather than guessed at. n8n pages
-    the listing at ten keys and matches ``label`` as a substring, so the query
-    narrows to the managed name and the exact match happens here.
+    belongs to a run that died and is deleted rather than guessed at. n8n matches
+    ``label`` as a substring, so the exact match happens here.
     """
-    query = urllib.parse.urlencode({"label": KEY_NAME})
-    status, body = call(f"{REST}/api-keys?{query}")
-    if status != 200:
-        sys.exit(f"FAILED listing api keys: {status} {body}")
-
     matches = [
-        key for key in records(body, "api keys") if key.get("label") == KEY_NAME
+        key for key in api_keys_labelled(KEY_NAME) if key.get("label") == KEY_NAME
     ]
     if len(matches) > 1:
         sys.exit(f"FAILED: {len(matches)} api keys named {KEY_NAME}")
@@ -319,12 +356,10 @@ def upsert_workflow(key, credential):
         key: the public-API key.
         credential: ``{"id", "name"}`` reference to the managed bearer credential.
     """
-    status, body = call(f"{PUBLIC}/workflows", api_key=key)
-    if status != 200:
-        sys.exit(f"FAILED listing workflows: {status} {body}")
-
     matches = [
-        flow for flow in records(body, "workflows") if flow.get("name") == WORKFLOW_NAME
+        flow
+        for flow in workflows_named(key, WORKFLOW_NAME)
+        if flow.get("name") == WORKFLOW_NAME
     ]
     if len(matches) > 1:
         sys.exit(f"FAILED: {len(matches)} workflows named {WORKFLOW_NAME}")

tests/unit/python/roles/web-app-n8n/files/provision/test_client.py

@@ -14,6 +14,20 @@ from . import PROJECT_ROOT
 
 SCRIPT_PATH = PROJECT_ROOT / "roles/web-app-n8n/files/python/provision/client.py"
 PAGE_SIZE = 10
+WORKFLOW_PAGE_SIZE = 100
+MAX_PAGE_SIZE = 250
+
+
+def paged(items, offset, limit):
+    """Return one page the way n8n slices a listing.
+
+    Args:
+        items: every record the listing holds.
+        offset: index of the first record on the page.
+        limit: requested page size, capped like n8n caps it.
+    """
+    return items[offset : offset + min(limit, MAX_PAGE_SIZE)]
+
 
 PROVIDER = {
     "id": "web-app-gitea",
@@ -81,17 +95,33 @@ class FakeApi:
             self.credentials = [c for c in self.credentials if c["id"] != dead]
             return 200, {"success": True}
         if path.startswith("/rest/api-keys?") and method == "GET":
-            label = parse_qs(urlsplit(path).query)["label"][0].lower()
+            query = parse_qs(urlsplit(path).query)
+            label = query["label"][0].lower()
             items = [dict(key) for key in self.keys if label in key["label"].lower()]
-            return 200, {"data": {"items": items[:PAGE_SIZE]}}
+            skip = int(query.get("skip", ["0"])[0])
+            take = int(query.get("take", [str(PAGE_SIZE)])[0])
+            return 200, {"data": {"items": paged(items, skip, take)}}
         if path == "/rest/api-keys" and method == "POST":
             self.keys = [{"id": "k1", "label": payload["label"]}]
             return 200, {"data": {"id": "k1", "rawApiKey": "n8n-raw-key"}}
         if path.startswith("/rest/api-keys/") and method == "DELETE":
             self.deleted_keys.append(path.rsplit("/", 1)[-1])
             return 200, {"success": True}
-        if path == "/api/v1/workflows" and method == "GET":
-            return 200, {"data": [dict(flow) for flow in self.workflows]}
+        if path.split("?")[0] == "/api/v1/workflows" and method == "GET":
+            query = parse_qs(urlsplit(path).query)
+            name = query.get("name", [""])[0]
+            hits = [dict(flow) for flow in self.workflows if name in flow["name"]]
+            cursor = query.get("cursor", [f"0:{WORKFLOW_PAGE_SIZE}"])[0].split(":")
+            offset, limit = (int(cursor[0]), int(cursor[1]))
+            if "limit" in query:
+                limit = int(query["limit"][0])
+            page = paged(hits, offset, limit)
+            following = offset + len(page)
+            more = following < len(hits)
+            return 200, {
+                "data": page,
+                "nextCursor": f"{following}:{limit}" if more else None,
+            }
         if path == "/api/v1/workflows" and method == "POST":
             self.created_workflows.append(payload)
             return 200, dict(payload, id="w1", active=False)
@@ -191,6 +221,34 @@ class TestProvisionClient(unittest.TestCase):
             module.main()
         self.assertEqual(["old", "k1"], api.deleted_keys)
 
+    def test_a_stale_managed_key_behind_a_full_filtered_page_is_still_replaced(
+        self,
+    ) -> None:
+        module = load_script()
+        near = [
+            {"id": f"n{n}", "label": f"infinito:mcp-copy-{n}"}
+            for n in range(MAX_PAGE_SIZE)
+        ]
+        api = FakeApi(keys=[*near, {"id": "old", "label": "infinito:mcp"}])
+        with patch.object(module, "call", api):
+            module.main()
+        self.assertEqual(["old", "k1"], api.deleted_keys)
+
+    def test_a_managed_workflow_on_a_later_page_is_updated_not_duplicated(
+        self,
+    ) -> None:
+        module = load_script()
+        near = [
+            {"id": f"n{n}", "name": f"infinito:mcp-agent copy {n}"}
+            for n in range(MAX_PAGE_SIZE)
+        ]
+        managed = {"id": "w9", "name": "infinito:mcp-agent"}
+        api = FakeApi(workflows=[*near, managed], credentials=[dict(REFERENCE)])
+        with patch.object(module, "call", api):
+            module.main()
+        self.assertEqual([], api.created_workflows)
+        self.assertEqual(1, len(api.updated_workflows))
+
     def test_an_unknown_listing_shape_names_its_keys_but_never_a_value(self) -> None:
         module = load_script()
         api = FakeApi()

tests/unit/python/roles/web-app-n8n/files/provision/test_mcp.py

@@ -14,6 +14,20 @@ from . import PROJECT_ROOT
 
 SCRIPT_PATH = PROJECT_ROOT / "roles/web-app-n8n/files/python/provision/mcp.py"
 PAGE_SIZE = 10
+WORKFLOW_PAGE_SIZE = 100
+MAX_PAGE_SIZE = 250
+
+
+def paged(items, offset, limit):
+    """Return one page the way n8n slices a listing.
+
+    Args:
+        items: every record the listing holds.
+        offset: index of the first record on the page.
+        limit: requested page size, capped like n8n caps it.
+    """
+    return items[offset : offset + min(limit, MAX_PAGE_SIZE)]
+
 
 ENV = {
     "N8N_BASE": "http://n8n:5678",
@@ -76,9 +90,12 @@ class FakeApi:
             self.credentials = [c for c in self.credentials if c["id"] != dead]
             return 200, {"success": True}
         if path.startswith("/rest/api-keys?") and method == "GET":
-            label = parse_qs(urlsplit(path).query)["label"][0].lower()
+            query = parse_qs(urlsplit(path).query)
+            label = query["label"][0].lower()
             items = [dict(key) for key in self.keys if label in key["label"].lower()]
-            return 200, {"data": {"items": items[:PAGE_SIZE]}}
+            skip = int(query.get("skip", ["0"])[0])
+            take = int(query.get("take", [str(PAGE_SIZE)])[0])
+            return 200, {"data": {"items": paged(items, skip, take)}}
         if path == "/rest/api-keys" and method == "POST":
             self.keys = [{"id": "k1", "label": payload["label"]}]
             self.scopes = set(payload["scopes"])
@@ -86,8 +103,21 @@ class FakeApi:
         if path.startswith("/rest/api-keys/") and method == "DELETE":
             self.deleted_keys.append(path.rsplit("/", 1)[-1])
             return 200, {"success": True}
-        if path == "/api/v1/workflows" and method == "GET":
-            return 200, {"data": [dict(flow) for flow in self.workflows]}
+        if path.split("?")[0] == "/api/v1/workflows" and method == "GET":
+            query = parse_qs(urlsplit(path).query)
+            name = query.get("name", [""])[0]
+            hits = [dict(flow) for flow in self.workflows if name in flow["name"]]
+            cursor = query.get("cursor", [f"0:{WORKFLOW_PAGE_SIZE}"])[0].split(":")
+            offset, limit = (int(cursor[0]), int(cursor[1]))
+            if "limit" in query:
+                limit = int(query["limit"][0])
+            page = paged(hits, offset, limit)
+            following = offset + len(page)
+            more = following < len(hits)
+            return 200, {
+                "data": page,
+                "nextCursor": f"{following}:{limit}" if more else None,
+            }
         if path == "/api/v1/workflows" and method == "POST":
             self.created_workflows.append(payload)
             return 200, dict(payload, id="w1", active=False)
@@ -227,6 +257,34 @@ class TestProvisionMcp(unittest.TestCase):
             module.main()
         self.assertEqual(["old", "k1"], api.deleted_keys)
 
+    def test_a_stale_managed_key_behind_a_full_filtered_page_is_still_replaced(
+        self,
+    ) -> None:
+        module = load_script()
+        near = [
+            {"id": f"n{n}", "label": f"infinito:mcp-copy-{n}"}
+            for n in range(MAX_PAGE_SIZE)
+        ]
+        api = FakeApi(keys=[*near, {"id": "old", "label": "infinito:mcp"}])
+        with patch.object(module, "call", api):
+            module.main()
+        self.assertEqual(["old", "k1"], api.deleted_keys)
+
+    def test_a_managed_workflow_on_a_later_page_is_updated_not_duplicated(
+        self,
+    ) -> None:
+        module = load_script()
+        near = [
+            {"id": f"n{n}", "name": f"infinito:mcp-server copy {n}"}
+            for n in range(MAX_PAGE_SIZE)
+        ]
+        managed = {"id": "w9", "name": "infinito:mcp-server", "active": True}
+        api = FakeApi(workflows=[*near, managed])
+        with patch.object(module, "call", api):
+            module.main()
+        self.assertEqual([], api.created_workflows)
+        self.assertEqual(1, len(api.updated_workflows))
+
     def test_a_key_that_only_contains_the_managed_name_is_left_alone(self) -> None:
         module = load_script()
         api = FakeApi(keys=[{"id": "near", "label": "infinito:mcp-legacy"}])