From 9babffa054dfa239f5c4b07dd838ad73a52a1e8a Mon Sep 17 00:00:00 2001 From: William FH <13333726+hinthornw@users.noreply.github.com> Date: Thu, 19 Feb 2026 11:10:07 -0800 Subject: [PATCH] feat(sdk-py): add stream_mode, stream_subgraphs, stream_resumable, durability to crons (#6876) ## Summary - Adds `stream_mode`, `stream_subgraphs`, `stream_resumable`, and `durability` parameters to cron `create`, `create_for_thread`, and `update` methods in both async and sync clients - Adds corresponding fields to the `CronUpdate` TypedDict in `schema.py` - Adds `checkpoint_during` deprecation warnings to cron create methods (consistent with the runs client pattern) These fields were added to the OpenAPI spec in langgraph-api but were not yet reflected in the Python SDK. ## Test plan - [x] `make format` passes - [x] `make lint` passes - [x] `make test` passes (69/69, including sync/async API parity test) --------- Co-authored-by: Claude Opus 4.6 --- libs/sdk-py/langgraph_sdk/_async/cron.py | 69 ++++++++++++++++++++++-- libs/sdk-py/langgraph_sdk/_sync/cron.py | 69 ++++++++++++++++++++++-- libs/sdk-py/langgraph_sdk/schema.py | 8 +++ 3 files changed, 136 insertions(+), 10 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/_async/cron.py b/libs/sdk-py/langgraph_sdk/_async/cron.py index 808980366..5929ad333 100644 --- a/libs/sdk-py/langgraph_sdk/_async/cron.py +++ b/libs/sdk-py/langgraph_sdk/_async/cron.py @@ -2,7 +2,8 @@ from __future__ import annotations -from collections.abc import Mapping +import warnings +from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any @@ -14,11 +15,13 @@ from langgraph_sdk.schema import ( Cron, CronSelectField, CronSortBy, + Durability, Input, OnCompletionBehavior, QueryParamTypes, Run, SortOrder, + StreamMode, ) @@ -60,13 +63,17 @@ class CronClient: metadata: Mapping[str, Any] | None = None, config: Config | None = None, context: Context | None = None, - checkpoint_during: bool | None = None, + checkpoint_during: bool | None = None, # deprecated interrupt_before: All | list[str] | None = None, interrupt_after: All | list[str] | None = None, webhook: str | None = None, multitask_strategy: str | None = None, end_time: datetime | None = None, enabled: bool | None = None, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + stream_subgraphs: bool | None = None, + stream_resumable: bool | None = None, + durability: Durability | None = None, headers: Mapping[str, str] | None = None, params: QueryParamTypes | None = None, ) -> Run: @@ -83,7 +90,7 @@ class CronClient: config: The configuration for the assistant. context: Static context to add to the assistant. !!! version-added "Added in version 0.6.0" - checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + checkpoint_during: (deprecated) Whether to checkpoint during the run (or only at the end/interruption). interrupt_before: Nodes to interrupt immediately before they get executed. interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. @@ -93,6 +100,13 @@ class CronClient: Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled or not. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether to persist the stream chunks in order to resume the stream later. + durability: Durability level for the run. Must be one of 'sync', 'async', or 'exit'. + "async" means checkpoints are persisted async while next graph step executes, replaces checkpoint_during=True + "sync" means checkpoints are persisted sync after graph step executes, replaces checkpoint_during=False + "exit" means checkpoints are only persisted when the run exits, does not save intermediate steps headers: Optional custom headers to include with the request. params: Optional query parameters to include with the request. @@ -118,6 +132,13 @@ class CronClient: ) ``` """ + if checkpoint_during is not None: + warnings.warn( + "`checkpoint_during` is deprecated and will be removed in a future version. Use `durability` instead.", + DeprecationWarning, + stacklevel=2, + ) + payload = { "schedule": schedule, "input": input, @@ -131,6 +152,10 @@ class CronClient: "webhook": webhook, "end_time": end_time.isoformat() if end_time else None, "enabled": enabled, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "durability": durability, } if multitask_strategy: payload["multitask_strategy"] = multitask_strategy @@ -151,7 +176,7 @@ class CronClient: metadata: Mapping[str, Any] | None = None, config: Config | None = None, context: Context | None = None, - checkpoint_during: bool | None = None, + checkpoint_during: bool | None = None, # deprecated interrupt_before: All | list[str] | None = None, interrupt_after: All | list[str] | None = None, webhook: str | None = None, @@ -159,6 +184,10 @@ class CronClient: multitask_strategy: str | None = None, end_time: datetime | None = None, enabled: bool | None = None, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + stream_subgraphs: bool | None = None, + stream_resumable: bool | None = None, + durability: Durability | None = None, headers: Mapping[str, str] | None = None, params: QueryParamTypes | None = None, ) -> Run: @@ -174,7 +203,7 @@ class CronClient: config: The configuration for the assistant. context: Static context to add to the assistant. !!! version-added "Added in version 0.6.0" - checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + checkpoint_during: (deprecated) Whether to checkpoint during the run (or only at the end/interruption). interrupt_before: Nodes to interrupt immediately before they get executed. interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. webhook: Webhook to call after LangGraph API call is done. @@ -186,6 +215,13 @@ class CronClient: Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled or not. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether to persist the stream chunks in order to resume the stream later. + durability: Durability level for the run. Must be one of 'sync', 'async', or 'exit'. + "async" means checkpoints are persisted async while next graph step executes, replaces checkpoint_during=True + "sync" means checkpoints are persisted sync after graph step executes, replaces checkpoint_during=False + "exit" means checkpoints are only persisted when the run exits, does not save intermediate steps headers: Optional custom headers to include with the request. params: Optional query parameters to include with the request. @@ -211,6 +247,13 @@ class CronClient: ``` """ + if checkpoint_during is not None: + warnings.warn( + "`checkpoint_during` is deprecated and will be removed in a future version. Use `durability` instead.", + DeprecationWarning, + stacklevel=2, + ) + payload = { "schedule": schedule, "input": input, @@ -225,6 +268,10 @@ class CronClient: "on_run_completed": on_run_completed, "end_time": end_time.isoformat() if end_time else None, "enabled": enabled, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "durability": durability, } if multitask_strategy: payload["multitask_strategy"] = multitask_strategy @@ -277,6 +324,10 @@ class CronClient: interrupt_after: All | list[str] | None = None, on_run_completed: OnCompletionBehavior | None = None, enabled: bool | None = None, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + stream_subgraphs: bool | None = None, + stream_resumable: bool | None = None, + durability: Durability | None = None, headers: Mapping[str, str] | None = None, params: QueryParamTypes | None = None, ) -> Cron: @@ -299,6 +350,10 @@ class CronClient: after execution. 'keep' creates a new thread for each execution but does not clean them up. enabled: Enable or disable the cron job. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether to persist the stream chunks in order to resume the stream later. + durability: Durability level for the run. Must be one of 'sync', 'async', or 'exit'. headers: Optional custom headers to include with the request. params: Optional query parameters to include with the request. @@ -329,6 +384,10 @@ class CronClient: "interrupt_after": interrupt_after, "on_run_completed": on_run_completed, "enabled": enabled, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "durability": durability, } payload = {k: v for k, v in payload.items() if v is not None} return await self.http.patch( diff --git a/libs/sdk-py/langgraph_sdk/_sync/cron.py b/libs/sdk-py/langgraph_sdk/_sync/cron.py index b335a15ef..49ca770a1 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/cron.py +++ b/libs/sdk-py/langgraph_sdk/_sync/cron.py @@ -2,7 +2,8 @@ from __future__ import annotations -from collections.abc import Mapping +import warnings +from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any @@ -14,11 +15,13 @@ from langgraph_sdk.schema import ( Cron, CronSelectField, CronSortBy, + Durability, Input, OnCompletionBehavior, QueryParamTypes, Run, SortOrder, + StreamMode, ) @@ -54,13 +57,17 @@ class SyncCronClient: metadata: Mapping[str, Any] | None = None, config: Config | None = None, context: Context | None = None, - checkpoint_during: bool | None = None, + checkpoint_during: bool | None = None, # deprecated interrupt_before: All | list[str] | None = None, interrupt_after: All | list[str] | None = None, webhook: str | None = None, multitask_strategy: str | None = None, end_time: datetime | None = None, enabled: bool | None = None, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + stream_subgraphs: bool | None = None, + stream_resumable: bool | None = None, + durability: Durability | None = None, headers: Mapping[str, str] | None = None, params: QueryParamTypes | None = None, ) -> Run: @@ -77,7 +84,7 @@ class SyncCronClient: config: The configuration for the assistant. context: Static context to add to the assistant. !!! version-added "Added in version 0.6.0" - checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + checkpoint_during: (deprecated) Whether to checkpoint during the run (or only at the end/interruption). interrupt_before: Nodes to interrupt immediately before they get executed. interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. webhook: Webhook to call after LangGraph API call is done. @@ -85,6 +92,13 @@ class SyncCronClient: Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled. By default, it is considered enabled. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether to persist the stream chunks in order to resume the stream later. + durability: Durability level for the run. Must be one of 'sync', 'async', or 'exit'. + "async" means checkpoints are persisted async while next graph step executes, replaces checkpoint_during=True + "sync" means checkpoints are persisted sync after graph step executes, replaces checkpoint_during=False + "exit" means checkpoints are only persisted when the run exits, does not save intermediate steps headers: Optional custom headers to include with the request. Returns: @@ -109,6 +123,13 @@ class SyncCronClient: ) ``` """ + if checkpoint_during is not None: + warnings.warn( + "`checkpoint_during` is deprecated and will be removed in a future version. Use `durability` instead.", + DeprecationWarning, + stacklevel=2, + ) + payload = { "schedule": schedule, "input": input, @@ -123,6 +144,10 @@ class SyncCronClient: "multitask_strategy": multitask_strategy, "end_time": end_time.isoformat() if end_time else None, "enabled": enabled, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "durability": durability, } payload = {k: v for k, v in payload.items() if v is not None} return self.http.post( @@ -141,7 +166,7 @@ class SyncCronClient: metadata: Mapping[str, Any] | None = None, config: Config | None = None, context: Context | None = None, - checkpoint_during: bool | None = None, + checkpoint_during: bool | None = None, # deprecated interrupt_before: All | list[str] | None = None, interrupt_after: All | list[str] | None = None, webhook: str | None = None, @@ -149,6 +174,10 @@ class SyncCronClient: multitask_strategy: str | None = None, end_time: datetime | None = None, enabled: bool | None = None, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + stream_subgraphs: bool | None = None, + stream_resumable: bool | None = None, + durability: Durability | None = None, headers: Mapping[str, str] | None = None, params: QueryParamTypes | None = None, ) -> Run: @@ -164,7 +193,7 @@ class SyncCronClient: config: The configuration for the assistant. context: Static context to add to the assistant. !!! version-added "Added in version 0.6.0" - checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + checkpoint_during: (deprecated) Whether to checkpoint during the run (or only at the end/interruption). interrupt_before: Nodes to interrupt immediately before they get executed. interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. webhook: Webhook to call after LangGraph API call is done. @@ -176,6 +205,13 @@ class SyncCronClient: Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely. enabled: Whether the cron job is enabled. By default, it is considered enabled. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether to persist the stream chunks in order to resume the stream later. + durability: Durability level for the run. Must be one of 'sync', 'async', or 'exit'. + "async" means checkpoints are persisted async while next graph step executes, replaces checkpoint_during=True + "sync" means checkpoints are persisted sync after graph step executes, replaces checkpoint_during=False + "exit" means checkpoints are only persisted when the run exits, does not save intermediate steps headers: Optional custom headers to include with the request. Returns: @@ -201,6 +237,13 @@ class SyncCronClient: ``` """ + if checkpoint_during is not None: + warnings.warn( + "`checkpoint_during` is deprecated and will be removed in a future version. Use `durability` instead.", + DeprecationWarning, + stacklevel=2, + ) + payload = { "schedule": schedule, "input": input, @@ -216,6 +259,10 @@ class SyncCronClient: "multitask_strategy": multitask_strategy, "end_time": end_time.isoformat() if end_time else None, "enabled": enabled, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "durability": durability, } payload = {k: v for k, v in payload.items() if v is not None} return self.http.post( @@ -266,6 +313,10 @@ class SyncCronClient: interrupt_after: All | list[str] | None = None, on_run_completed: OnCompletionBehavior | None = None, enabled: bool | None = None, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + stream_subgraphs: bool | None = None, + stream_resumable: bool | None = None, + durability: Durability | None = None, headers: Mapping[str, str] | None = None, params: QueryParamTypes | None = None, ) -> Cron: @@ -288,6 +339,10 @@ class SyncCronClient: after execution. 'keep' creates a new thread for each execution but does not clean them up. enabled: Enable or disable the cron job. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether to persist the stream chunks in order to resume the stream later. + durability: Durability level for the run. Must be one of 'sync', 'async', or 'exit'. headers: Optional custom headers to include with the request. params: Optional query parameters to include with the request. @@ -318,6 +373,10 @@ class SyncCronClient: "interrupt_after": interrupt_after, "on_run_completed": on_run_completed, "enabled": enabled, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "durability": durability, } payload = {k: v for k, v in payload.items() if v is not None} return self.http.patch( diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index 486ae8ce0..7567f4077 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -424,6 +424,14 @@ class CronUpdate(TypedDict, total=False): """What to do with the thread after the run completes.""" enabled: bool """Enable or disable the cron job.""" + stream_mode: StreamMode | list[StreamMode] + """The stream mode(s) to use.""" + stream_subgraphs: bool + """Whether to stream output from subgraphs.""" + stream_resumable: bool + """Whether to persist the stream chunks in order to resume the stream later.""" + durability: Durability + """Durability level for the run. Must be one of 'sync', 'async', or 'exit'.""" # Select field aliases for client-side typing of `select` parameters.