Compare commits

...
Author SHA1 Message Date
Sydney Runkle bbc35a30f8 POC streaming hints 2026-02-23 09:09:13 -05:00
William FHandGitHub b0f14649e0 chore: Update CLI schema (#6858) 2026-02-20 01:11:21 +00:00
ea20432b9b fix: bump js-yaml to 3.14.2 to resolve CVE-2025-64718 (#6879)
## Security Alert Patch

Resolves 1 Dependabot security alert (medium severity).

### Package Updated

| Package | Old Version | New Version | Strategy | CVE Resolved |
|---------|-------------|-------------|----------|--------------|
| `js-yaml` | 3.14.1 | 3.14.2 | Lockfile patch (within-range bump) |
CVE-2025-64718 |

### CVE Details

**CVE-2025-64718** /
[GHSA-mh29-5h37-fv8m](https://github.com/advisories/GHSA-mh29-5h37-fv8m)
— `js-yaml` prototype pollution via YAML merge keys (`<<`). Affects
versions < 3.14.2.

The vulnerable package is a transitive dev dependency pulled in by
`@istanbuljs/load-nyc-config@1.1.0` (a Jest internal). No runtime
impact.

### Fix Strategy

Lockfile-only patch in `libs/cli/js-examples/yarn.lock`. The `^3.13.1`
version range already allows 3.14.2, so no manifest changes were needed.
The `js-yaml@^4.1.1` entry (used by `@eslint/eslintrc`) is untouched.

### Verification

- [x] Lockfile updated — `js-yaml@^3.13.1` now resolves to `3.14.2`
- [x] `js-yaml@^4.1.1` entry unchanged (`4.1.1`)
- [x] `yarn install --frozen-lockfile` passes

🤖 Submitted by langster-patch

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-02-19 19:41:51 +00:00
William FHandGitHub e2efab8061 release(sdk-py): 0.3.8 (#6873)
Includes some updated docstrings.
2026-02-19 11:10:19 -08:00
9babffa054 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 <noreply@anthropic.com>
2026-02-19 11:10:07 -08:00
12 changed files with 1182 additions and 23 deletions
+3 -3
View File
@@ -3533,9 +3533,9 @@ js-tokens@^4.0.0:
integrity sha512-RdJUflcE3cUzKiMqQgsCu06FPu9UdIJO0beYbPhHN4k6apgJtifcoCtT9bcxOpYBtpD2kCM6Sbzg4CausW/PKQ==
js-yaml@^3.13.1:
version "3.14.1"
resolved "https://registry.yarnpkg.com/js-yaml/-/js-yaml-3.14.1.tgz#dae812fdb3825fa306609a8717383c50c36a0537"
integrity sha512-okMH7OXXJ7YrN9Ok3/SXrnu4iX9yOk+25nqX4imS2npuvTYDmo/QEZoqwZkYaIDk3jVvBOTOIEgEhaLOynBS9g==
version "3.14.2"
resolved "https://registry.yarnpkg.com/js-yaml/-/js-yaml-3.14.2.tgz#77485ce1dd7f33c061fd1b16ecea23b55fcb04b0"
integrity sha512-PMSmkqxr106Xa156c2M265Z+FTrPl+oxd/rgOQy2tijQeK5TxQ43psO1ZCwhVOSdnn+RzkzlRz/eY4BgJBYVpg==
dependencies:
argparse "^1.0.7"
esprima "^4.0.0"
+47
View File
@@ -678,6 +678,51 @@ def _update_encryption_path(
)
def _update_checkpointer_path(
config_path: pathlib.Path, config: Config, local_deps: LocalDeps
) -> None:
"""Update checkpointer.path to use Docker container paths."""
checkpointer_conf = config.get("checkpointer")
if not checkpointer_conf or not isinstance(checkpointer_conf, dict):
return
if not (path_str := checkpointer_conf.get("path")):
return
module_str, sep, attr_str = path_str.partition(":")
if not sep or not module_str.startswith("."):
return # Already validated or absolute path
resolved = config_path.parent / module_str
if not resolved.exists():
raise FileNotFoundError(
f"Checkpointer file not found: {resolved} (from {path_str})"
)
if not resolved.is_file():
raise IsADirectoryError(f"Checkpointer path must be a file: {resolved}")
# Check faux packages first (higher priority)
for faux_path, (_, destpath) in local_deps.faux_pkgs.items():
if resolved.is_relative_to(faux_path):
new_path = f"{destpath}/{resolved.relative_to(faux_path)}:{attr_str}"
checkpointer_conf["path"] = new_path
return
# Check real packages
for real_path in local_deps.real_pkgs:
if resolved.is_relative_to(real_path):
new_path = (
f"/deps/{real_path.name}/{resolved.relative_to(real_path)}:{attr_str}"
)
checkpointer_conf["path"] = new_path
return
raise ValueError(
f"Checkpointer file '{resolved}' not covered by dependencies.\n"
"Add its parent directory to the 'dependencies' array in your config.\n"
f"Current dependencies: {config['dependencies']}"
)
def _update_http_app_path(
config_path: pathlib.Path, config: Config, local_deps: LocalDeps
) -> None:
@@ -877,6 +922,8 @@ def python_config_to_docker(
_update_auth_path(config_path, config, local_deps)
# Rewrite encryption path, so it points to the correct location in the Docker container
_update_encryption_path(config_path, config, local_deps)
# Rewrite checkpointer path, so it points to the correct location in the Docker container
_update_checkpointer_path(config_path, config, local_deps)
# Rewrite HTTP app path, so it points to the correct location in the Docker container
_update_http_app_path(config_path, config, local_deps)
+20
View File
@@ -167,6 +167,26 @@ class CheckpointerConfig(TypedDict, total=False):
If omitted, no checkpointer is set up (the object store will still be present, however).
"""
path: str
"""Import path to an async context manager that yields a `BaseCheckpointSaver`
instance.
The referenced object should be an `@asynccontextmanager`-decorated function
so that the server can properly manage the checkpointer's lifecycle (e.g.
opening and closing connections).
Examples:
- "./my_checkpointer.py:create_checkpointer"
- "my_package.checkpointer:create_checkpointer"
When provided, this replaces the default checkpointer.
You can use the `langgraph-checkpoint-conformance` package
(https://pypi.org/project/langgraph-checkpoint-conformance/) to run simple
conformance tests against your custom checkpointer and catch
incompatibilities early.
"""
ttl: ThreadTTLConfig | None
"""Optional. Defines the TTL (time-to-live) behavior configuration.
+4
View File
@@ -542,6 +542,10 @@
"description": "Configuration for the built-in checkpointer, which handles checkpointing of state.\n\nIf omitted, no checkpointer is set up (the object store will still be present, however).",
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Import path to an async context manager that yields a `BaseCheckpointSaver`\ninstance.\n\nThe referenced object should be an `@asynccontextmanager`-decorated function\nso that the server can properly manage the checkpointer's lifecycle (e.g.\nopening and closing connections).\n"
},
"serde": {
"anyOf": [
{
+4
View File
@@ -542,6 +542,10 @@
"description": "Configuration for the built-in checkpointer, which handles checkpointing of state.\n\nIf omitted, no checkpointer is set up (the object store will still be present, however).",
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Import path to an async context manager that yields a `BaseCheckpointSaver`\ninstance.\n\nThe referenced object should be an `@asynccontextmanager`-decorated function\nso that the server can properly manage the checkpointer's lifecycle (e.g.\nopening and closing connections).\n"
},
"serde": {
"anyOf": [
{
+378 -8
View File
@@ -20,10 +20,13 @@ from dataclasses import is_dataclass
from functools import partial
from inspect import isclass
from typing import (
TYPE_CHECKING,
Any,
Generic,
Literal,
cast,
get_type_hints,
overload,
)
from uuid import UUID, uuid5
@@ -48,6 +51,9 @@ from langgraph.store.base import BaseStore
from pydantic import BaseModel, TypeAdapter
from typing_extensions import Self, Unpack, deprecated, is_typeddict
if TYPE_CHECKING:
from langchain_core.messages import AnyMessage
from langgraph._internal._config import (
ensure_config,
merge_configs,
@@ -2404,6 +2410,186 @@ class Pregel(
durability,
)
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[OutputT]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[dict[str, Any]]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[tuple[AnyMessage, dict[str, Any]]]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[Any]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[tuple[tuple[str, ...], OutputT]]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[tuple[tuple[str, ...], tuple[AnyMessage, dict[str, Any]]]]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[tuple[tuple[str, ...], Any]]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[tuple[str, Any]]: ...
@overload
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Iterator[tuple[tuple[str, ...], str, Any]]: ...
def stream(
self,
input: InputT | Command | None,
@@ -2678,6 +2864,186 @@ class Pregel(
run_manager.on_chain_error(e)
raise
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[OutputT]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[dict[str, Any]]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[tuple[AnyMessage, dict[str, Any]]]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[Any]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[tuple[tuple[str, ...], OutputT]]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[tuple[tuple[str, ...], tuple[AnyMessage, dict[str, Any]]]]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[tuple[tuple[str, ...], Any]]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[False] = False,
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[tuple[str, Any]]: ...
@overload
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
print_mode: StreamMode | Sequence[StreamMode] = (),
output_keys: str | Sequence[str] | None = None,
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
durability: Durability | None = None,
subgraphs: Literal[True],
debug: bool | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> AsyncIterator[tuple[tuple[str, ...], str, Any]]: ...
async def astream(
self,
input: InputT | Command | None,
@@ -3068,13 +3434,15 @@ class Pregel(
chunks: list[dict[str, Any] | Any] = []
interrupts: list[Interrupt] = []
for chunk in self.stream(
for chunk in self.stream( # type: ignore[misc]
input,
config,
context=context,
stream_mode=["updates", "values"]
if stream_mode == "values"
else stream_mode,
stream_mode=(
["updates", "values"] # type: ignore[arg-type]
if stream_mode == "values"
else stream_mode
),
print_mode=print_mode,
output_keys=output_keys,
interrupt_before=interrupt_before,
@@ -3158,13 +3526,15 @@ class Pregel(
chunks: list[dict[str, Any] | Any] = []
interrupts: list[Interrupt] = []
async for chunk in self.astream(
async for chunk in self.astream( # type: ignore[misc]
input,
config,
context=context,
stream_mode=["updates", "values"]
if stream_mode == "values"
else stream_mode,
stream_mode=(
["updates", "values"] # type: ignore[arg-type]
if stream_mode == "values"
else stream_mode
),
print_mode=print_mode,
output_keys=output_keys,
interrupt_before=interrupt_before,
+284 -1
View File
@@ -2,13 +2,16 @@ from __future__ import annotations
from abc import abstractmethod
from collections.abc import AsyncIterator, Callable, Iterator, Sequence
from typing import Any, Generic, cast
from typing import TYPE_CHECKING, Any, Generic, Literal, cast, overload
from langchain_core.runnables import Runnable, RunnableConfig
from langchain_core.runnables.graph import Graph as DrawableGraph
from typing_extensions import Self
from langgraph.types import All, Command, StateSnapshot, StateUpdate, StreamMode
if TYPE_CHECKING:
from langchain_core.messages import AnyMessage
from langgraph.typing import ContextT, InputT, OutputT, StateT
__all__ = ("PregelProtocol", "StreamProtocol")
@@ -96,6 +99,146 @@ class PregelProtocol(Runnable[InputT, Any], Generic[StateT, ContextT, InputT, Ou
as_node: str | None = None,
) -> RunnableConfig: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> Iterator[OutputT]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> Iterator[dict[str, Any]]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> Iterator[tuple[AnyMessage, dict[str, Any]]]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> Iterator[Any]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> Iterator[tuple[tuple[str, ...], OutputT]]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> Iterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> Iterator[tuple[tuple[str, ...], tuple[AnyMessage, dict[str, Any]]]]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> Iterator[tuple[tuple[str, ...], Any]]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> Iterator[tuple[str, Any]]: ...
@overload
@abstractmethod
def stream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> Iterator[tuple[tuple[str, ...], str, Any]]: ...
@abstractmethod
def stream(
self,
@@ -109,6 +252,146 @@ class PregelProtocol(Runnable[InputT, Any], Generic[StateT, ContextT, InputT, Ou
subgraphs: bool = False,
) -> Iterator[dict[str, Any] | Any]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> AsyncIterator[OutputT]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> AsyncIterator[dict[str, Any]]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> AsyncIterator[tuple[AnyMessage, dict[str, Any]]]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> AsyncIterator[Any]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> AsyncIterator[tuple[tuple[str, ...], OutputT]]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> AsyncIterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> AsyncIterator[tuple[tuple[str, ...], tuple[AnyMessage, dict[str, Any]]]]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> AsyncIterator[tuple[tuple[str, ...], Any]]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
) -> AsyncIterator[tuple[str, Any]]: ...
@overload
@abstractmethod
def astream(
self,
input: InputT | Command | None,
config: RunnableConfig | None = None,
*,
context: ContextT | None = None,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
) -> AsyncIterator[tuple[tuple[str, ...], str, Any]]: ...
@abstractmethod
def astream(
self,
+305
View File
@@ -4,9 +4,11 @@ import logging
from collections.abc import AsyncIterator, Iterator, Sequence
from dataclasses import asdict
from typing import (
TYPE_CHECKING,
Any,
Literal,
cast,
overload,
)
from uuid import UUID
@@ -41,6 +43,9 @@ from langgraph_sdk.schema import (
)
from typing_extensions import Self
if TYPE_CHECKING:
from langchain_core.messages import AnyMessage
from langgraph._internal._config import merge_configs
from langgraph._internal._constants import (
CONF,
@@ -682,6 +687,156 @@ class RemoteGraph(PregelProtocol):
updated_stream_modes.remove("events")
return (updated_stream_modes, requested_stream_modes, req_single, stream)
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[dict[str, Any]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[dict[str, Any]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[tuple[AnyMessage, dict[str, Any]]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[Any]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[tuple[tuple[str, ...], tuple[AnyMessage, dict[str, Any]]]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[tuple[tuple[str, ...], Any]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[tuple[str, Any]]: ...
@overload
def stream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> Iterator[tuple[tuple[str, ...], str, Any]]: ...
def stream(
self,
input: dict[str, Any] | Any,
@@ -792,6 +947,156 @@ class RemoteGraph(PregelProtocol):
else:
yield chunk
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[dict[str, Any]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[dict[str, Any]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[tuple[AnyMessage, dict[str, Any]]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[Any]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["values"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["updates"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[tuple[tuple[str, ...], dict[str, Any]]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["messages"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[tuple[tuple[str, ...], tuple[AnyMessage, dict[str, Any]]]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: Literal["custom"],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[tuple[tuple[str, ...], Any]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[False] = False,
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[tuple[str, Any]]: ...
@overload
def astream(
self,
input: dict[str, Any] | Any,
config: RunnableConfig | None = None,
*,
stream_mode: list[StreamMode],
interrupt_before: All | Sequence[str] | None = None,
interrupt_after: All | Sequence[str] | None = None,
subgraphs: Literal[True],
headers: dict[str, str] | None = None,
params: QueryParamTypes | None = None,
**kwargs: Any,
) -> AsyncIterator[tuple[tuple[str, ...], str, Any]]: ...
async def astream(
self,
input: dict[str, Any] | Any,
+1 -1
View File
@@ -3,6 +3,6 @@ from langgraph_sdk.client import get_client, get_sync_client
from langgraph_sdk.encryption import Encryption
from langgraph_sdk.encryption.types import EncryptionContext
__version__ = "0.3.7"
__version__ = "0.3.8"
__all__ = ["Auth", "Encryption", "EncryptionContext", "get_client", "get_sync_client"]
+64 -5
View File
@@ -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(
+64 -5
View File
@@ -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(
+8
View File
@@ -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.