feat(sdk-py): add update method for crons client (#6742)

# Description
- Update Python SDK to add patch method to cron client
- This method lets users modify cron attributes (except assistant ID and
thread ID)

---------

Co-authored-by: William Fu-Hinthorn <13333726+hinthornw@users.noreply.github.com>

libs/sdk-py/langgraph_sdk/client.py

@@ -3180,6 +3180,82 @@ class CronClient:
         """
         await self.http.delete(f"/runs/crons/{cron_id}", headers=headers, params=params)
 
+    async def update(
+        self,
+        cron_id: str,
+        *,
+        schedule: str | None = None,
+        end_time: datetime | None = None,
+        input: Input | None = None,
+        metadata: Mapping[str, Any] | None = None,
+        config: Config | None = None,
+        context: Context | None = None,
+        webhook: str | None = None,
+        interrupt_before: All | list[str] | None = None,
+        interrupt_after: All | list[str] | None = None,
+        on_run_completed: OnCompletionBehavior | None = None,
+        enabled: bool | None = None,
+        headers: Mapping[str, str] | None = None,
+        params: QueryParamTypes | None = None,
+    ) -> Cron:
+        """Update a cron job by ID.
+
+        Args:
+            cron_id: The cron ID to update.
+            schedule: The cron schedule to execute this job on.
+                Schedules are interpreted in UTC.
+            end_time: The end date to stop running the cron.
+            input: The input to the graph.
+            metadata: Metadata to assign to the cron job runs.
+            config: The configuration for the assistant.
+            context: Static context added to the assistant.
+            webhook: Webhook to call after LangGraph API call is done.
+            interrupt_before: Nodes to interrupt immediately before they get executed.
+            interrupt_after: Nodes to interrupt immediately after they get executed.
+            on_run_completed: What to do with the thread after the run completes.
+                Must be one of 'delete' or 'keep'. 'delete' removes the thread
+                after execution. 'keep' creates a new thread for each execution but does not
+                clean them up.
+            enabled: Enable or disable the cron job.
+            headers: Optional custom headers to include with the request.
+            params: Optional query parameters to include with the request.
+
+        Returns:
+            The updated cron job.
+
+        ???+ example "Example Usage"
+
+            ```python
+            client = get_client(url="http://localhost:2024")
+            updated_cron = await client.crons.update(
+                cron_id="1ef3cefa-4c09-6926-96d0-3dc97fd5e39b",
+                schedule="0 10 * * *",
+                enabled=False,
+            )
+            ```
+
+        """
+        payload = {
+            "schedule": schedule,
+            "end_time": end_time.isoformat() if end_time else None,
+            "input": input,
+            "metadata": metadata,
+            "config": config,
+            "context": context,
+            "webhook": webhook,
+            "interrupt_before": interrupt_before,
+            "interrupt_after": interrupt_after,
+            "on_run_completed": on_run_completed,
+            "enabled": enabled,
+        }
+        payload = {k: v for k, v in payload.items() if v is not None}
+        return await self.http.patch(
+            f"/runs/crons/{cron_id}",
+            json=payload,
+            headers=headers,
+            params=params,
+        )
+
     async def search(
         self,
         *,
@@ -6497,6 +6573,82 @@ class SyncCronClient:
         """
         self.http.delete(f"/runs/crons/{cron_id}", headers=headers, params=params)
 
+    def update(
+        self,
+        cron_id: str,
+        *,
+        schedule: str | None = None,
+        end_time: datetime | None = None,
+        input: Input | None = None,
+        metadata: Mapping[str, Any] | None = None,
+        config: Config | None = None,
+        context: Context | None = None,
+        webhook: str | None = None,
+        interrupt_before: All | list[str] | None = None,
+        interrupt_after: All | list[str] | None = None,
+        on_run_completed: OnCompletionBehavior | None = None,
+        enabled: bool | None = None,
+        headers: Mapping[str, str] | None = None,
+        params: QueryParamTypes | None = None,
+    ) -> Cron:
+        """Update a cron job by ID.
+
+        Args:
+            cron_id: The cron ID to update.
+            schedule: The cron schedule to execute this job on.
+                Schedules are interpreted in UTC.
+            end_time: The end date to stop running the cron.
+            input: The input to the graph.
+            metadata: Metadata to assign to the cron job runs.
+            config: The configuration for the assistant.
+            context: Static context added to the assistant.
+            webhook: Webhook to call after LangGraph API call is done.
+            interrupt_before: Nodes to interrupt immediately before they get executed.
+            interrupt_after: Nodes to interrupt immediately after they get executed.
+            on_run_completed: What to do with the thread after the run completes.
+                Must be one of 'delete' or 'keep'. 'delete' removes the thread
+                after execution. 'keep' creates a new thread for each execution but does not
+                clean them up.
+            enabled: Enable or disable the cron job.
+            headers: Optional custom headers to include with the request.
+            params: Optional query parameters to include with the request.
+
+        Returns:
+            The updated cron job.
+
+        ???+ example "Example Usage"
+
+            ```python
+            client = get_sync_client(url="http://localhost:8123")
+            updated_cron = client.crons.update(
+                cron_id="1ef3cefa-4c09-6926-96d0-3dc97fd5e39b",
+                schedule="0 10 * * *",
+                enabled=False,
+            )
+            ```
+
+        """
+        payload = {
+            "schedule": schedule,
+            "end_time": end_time.isoformat() if end_time else None,
+            "input": input,
+            "metadata": metadata,
+            "config": config,
+            "context": context,
+            "webhook": webhook,
+            "interrupt_before": interrupt_before,
+            "interrupt_after": interrupt_after,
+            "on_run_completed": on_run_completed,
+            "enabled": enabled,
+        }
+        payload = {k: v for k, v in payload.items() if v is not None}
+        return self.http.patch(
+            f"/runs/crons/{cron_id}",
+            json=payload,
+            headers=headers,
+            params=params,
+        )
+
     def search(
         self,
         *,

libs/sdk-py/langgraph_sdk/schema.py

@@ -376,6 +376,33 @@ class Cron(TypedDict):
     """Whether the cron is enabled."""
 
 
+class CronUpdate(TypedDict, total=False):
+    """Payload for updating a cron job. All fields are optional."""
+
+    schedule: str
+    """The cron schedule to execute this job on."""
+    end_time: datetime
+    """The end date to stop running the cron."""
+    input: Input
+    """The input to the graph."""
+    metadata: dict[str, Any]
+    """Metadata to assign to the cron job runs."""
+    config: Config
+    """The configuration for the assistant."""
+    context: Context
+    """Static context added to the assistant."""
+    webhook: str
+    """Webhook to call after LangGraph API call is done."""
+    interrupt_before: All | list[str]
+    """Nodes to interrupt immediately before they get executed."""
+    interrupt_after: All | list[str]
+    """Nodes to interrupt immediately after they get executed."""
+    on_run_completed: OnCompletionBehavior
+    """What to do with the thread after the run completes."""
+    enabled: bool
+    """Enable or disable the cron job."""
+
+
 # Select field aliases for client-side typing of `select` parameters.
 # These mirror the server's allowed field sets.
 

libs/sdk-py/tests/test_crons_client.py

@@ -309,3 +309,179 @@ def test_sync_create_with_enabled_parameter(enabled_value):
         )
 
     assert result == cron
+
+
+def _cron_response() -> dict[str, object]:
+    """Return a mock Cron object response."""
+    return {
+        "cron_id": "cron_123",
+        "assistant_id": "asst_123",
+        "thread_id": "thread_123",
+        "on_run_completed": None,
+        "end_time": "2025-12-31T23:59:59+00:00",
+        "schedule": "0 10 * * *",
+        "created_at": "2024-01-01T00:00:00Z",
+        "updated_at": "2024-01-02T00:00:00Z",
+        "payload": {},
+        "user_id": None,
+        "next_run_date": "2024-01-03T10:00:00Z",
+        "metadata": {},
+        "enabled": True,
+    }
+
+
+@pytest.mark.asyncio
+async def test_async_update():
+    """Test that CronClient.update works with schedule and enabled parameters."""
+    cron = _cron_response()
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "PATCH"
+        assert request.url.path == "/runs/crons/cron_123"
+
+        # Parse the request body
+        body = json.loads(request.content)
+        assert body["schedule"] == "0 10 * * *"
+        assert body["enabled"] is False
+        assert "end_time" not in body  # Should be filtered out by the None check
+
+        return httpx.Response(200, json=cron)
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        result = await cron_client.update(
+            cron_id="cron_123",
+            schedule="0 10 * * *",
+            enabled=False,
+        )
+
+    assert result == cron
+
+
+@pytest.mark.asyncio
+async def test_async_update_with_end_time():
+    """Test that CronClient.update includes end_time in the payload."""
+    cron = _cron_response()
+    end_time = datetime(2025, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "PATCH"
+        assert request.url.path == "/runs/crons/cron_123"
+
+        # Parse the request body
+        body = json.loads(request.content)
+        assert body["schedule"] == "0 10 * * *"
+        assert body["end_time"] == "2025-12-31T23:59:59+00:00"
+        assert body["enabled"] is True
+
+        return httpx.Response(200, json=cron)
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        result = await cron_client.update(
+            cron_id="cron_123",
+            schedule="0 10 * * *",
+            end_time=end_time,
+            enabled=True,
+        )
+
+    assert result == cron
+
+
+def test_sync_update():
+    """Test that SyncCronClient.update works with schedule and enabled parameters."""
+    cron = _cron_response()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "PATCH"
+        assert request.url.path == "/runs/crons/cron_123"
+
+        # Parse the request body
+        body = json.loads(request.content)
+        assert body["schedule"] == "0 10 * * *"
+        assert body["enabled"] is False
+        assert "end_time" not in body  # Should be filtered out by the None check
+
+        return httpx.Response(200, json=cron)
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        result = cron_client.update(
+            cron_id="cron_123",
+            schedule="0 10 * * *",
+            enabled=False,
+        )
+
+    assert result == cron
+
+
+def test_sync_update_with_end_time():
+    """Test that SyncCronClient.update includes end_time in the payload."""
+    cron = _cron_response()
+    end_time = datetime(2025, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "PATCH"
+        assert request.url.path == "/runs/crons/cron_123"
+
+        # Parse the request body
+        body = json.loads(request.content)
+        assert body["schedule"] == "0 10 * * *"
+        assert body["end_time"] == "2025-12-31T23:59:59+00:00"
+        assert body["enabled"] is True
+
+        return httpx.Response(200, json=cron)
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        result = cron_client.update(
+            cron_id="cron_123",
+            schedule="0 10 * * *",
+            end_time=end_time,
+            enabled=True,
+        )
+
+    assert result == cron
+
+
+@pytest.mark.parametrize(
+    "enabled_value",
+    [True, False],
+    ids=["enabled", "disabled"],
+)
+def test_sync_update_with_enabled_parameter(enabled_value):
+    """Test that SyncCronClient.update includes enabled parameter in the payload."""
+    cron = _cron_response()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "PATCH"
+        assert request.url.path == "/runs/crons/cron_456"
+
+        body = json.loads(request.content)
+        assert body["enabled"] == enabled_value
+        assert "schedule" not in body  # Only enabled is set
+
+        return httpx.Response(200, json=cron)
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        result = cron_client.update(
+            cron_id="cron_456",
+            enabled=enabled_value,
+        )
+
+    assert result == cron