Feat/sdk targeted list limits (list-limits part of issue #1107)

README.md

@@ -214,6 +214,10 @@ specific = await robot.get_run("run-id")
 
 # Update metadata or workflow
 await robot.update({"meta": {"name": "New Name"}})
+
+# Update a list, crawl, or search limit without resending the workflow
+await robot.set_list_limit(25)
+
 await robot.refresh()   # reload from server
 
 # Delete

client.py

@@ -2,9 +2,9 @@
 import os
 import httpx
 from datetime import datetime, timezone
-from typing import Optional, Union
+from typing import Optional, Union, List
 from .llm_options import build_llm_payload
-from .types import Config, MaxunError
+from .types import Config, MaxunError, ListLimitUpdate
 
 
 def _document_content_type(file_name: str) -> str:
@@ -93,6 +93,19 @@ async def update_robot(self, robot_id: str, updates: dict):
             raise MaxunError(f"Failed to update robot {robot_id}")
         return data
 
+    async def update_list_limits(self, robot_id: str, limits: List[ListLimitUpdate],):
+        """Update one or more list limits without resending the workflow."""
+        data = await self._handle(
+            self.client.put(
+                f"/robots/{robot_id}",
+                json={"limits": limits},
+            )
+        )
+
+        if not data:
+            raise MaxunError(f"Failed to update list limits for robot {robot_id}")
+
+        return data
     async def delete_robot(self, robot_id: str):
         await self._handle(self.client.delete(f"/robots/{robot_id}"))
 
@@ -269,4 +282,4 @@ async def create_crawl_robot(self, url: str, options: dict):
     async def create_search_robot(self, options: dict):
         return await self._handle(
             self.client.post("/search", json=options)
-        )
+        )

robot.py

@@ -1,5 +1,6 @@
 from typing import Optional, List
 from .client import Client
+from .types import MaxunError
 
 
 class Robot:
@@ -62,6 +63,46 @@ async def update(self, updates: dict) -> None:
         updated = await self.client.update_robot(self.id, updates)
         self.robot_data = updated
 
+    async def set_list_limit(self, limit: int) -> None:
+        """Set the limit for the first matching scrapeList, crawl, or search action."""
+        supported_actions = {"scrapeList", "crawl", "search"}
+        workflow = (
+            (self.robot_data.get("recording") or {})
+            .get("workflow")
+            or []
+        )
+
+        for pair_index, pair in enumerate(workflow):
+            for action_index, action in enumerate(
+                pair.get("what") or []
+            ):
+                if action.get("action") not in supported_actions:
+                    continue
+
+                for arg_index, arg in enumerate(
+                    action.get("args") or []
+                ):
+                    if isinstance(arg, dict) and "limit" in arg:
+                        self.robot_data = (
+                            await self.client.update_list_limits(
+                                self.id,
+                                [
+                                    {
+                                        "pairIndex": pair_index,
+                                        "actionIndex": action_index,
+                                        "argIndex": arg_index,
+                                        "limit": limit,
+                                    }
+                                ],
+                            )
+                        )
+                        return
+
+        raise MaxunError(
+            "This robot has no scrapeList, crawl, or search "
+            "action with a limit to update."
+        )
+
     async def duplicate(self, target_url: str):
         new_robot_data = await self.client.duplicate_robot(self.id, target_url)
         return Robot(self.client, new_robot_data)

types.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Optional, List, Dict, Any, Literal
+from typing import Optional, List, Dict, Any, Literal, TypedDict
 
 # ======================
 # Core Types
@@ -117,6 +117,11 @@ class SearchOptions:
 RunResult = Dict[str, Any]
 ApiResponse = Dict[str, Any]
 
+class ListLimitUpdate(TypedDict):
+    pairIndex: int
+    actionIndex: int
+    argIndex: int
+    limit: int
 
 class MaxunError(Exception):
     def __init__(self, message: str, status_code: Optional[int] = None, details: Any = None):