diff --git a/Dockerfile b/Dockerfile
index aea899650..2095ad194 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM python:3.8-slim
+FROM python:3.9-slim
 
 # Set the working directory to /app
 WORKDIR /app
diff --git a/langgraph/channels/base.py b/langgraph/channels/base.py
index d317cc93d..b30bee9b2 100644
--- a/langgraph/channels/base.py
+++ b/langgraph/channels/base.py
@@ -72,7 +72,7 @@ class BaseChannel(Generic[Value, Update, C], ABC):
         Raises EmptyChannelError if the channel is empty (never updated yet)."""
 
     @abstractmethod
-    def checkpoint(self) -> C | None:
+    def checkpoint(self) -> Optional[C]:
         """Return a string representation of the channel's current state.
 
         Raises EmptyChannelError if the channel is empty (never updated yet),
diff --git a/langgraph/channels/topic.py b/langgraph/channels/topic.py
index 4e4149fc7..1a7cec7f3 100644
--- a/langgraph/channels/topic.py
+++ b/langgraph/channels/topic.py
@@ -6,7 +6,7 @@ from typing_extensions import Self
 from langgraph.channels.base import BaseChannel, Value
 
 
-def flatten(values: Sequence[Value | list[Value]]) -> Iterator[Value]:
+def flatten(values: Sequence[Union[Value, list[Value]]]) -> Iterator[Value]:
     for value in values:
         if isinstance(value, list):
             yield from value
@@ -16,7 +16,9 @@ def flatten(values: Sequence[Value | list[Value]]) -> Iterator[Value]:
 
 class Topic(
     Generic[Value],
-    BaseChannel[Sequence[Value], Value | list[Value], tuple[set[Value], list[Value]]],
+    BaseChannel[
+        Sequence[Value], Union[Value, list[Value]], tuple[set[Value], list[Value]]
+    ],
 ):
     """A configurable PubSub Topic.
 
@@ -60,7 +62,7 @@ class Topic(
         finally:
             pass
 
-    def update(self, values: Sequence[Value | list[Value]]) -> None:
+    def update(self, values: Sequence[Union[Value, list[Value]]]) -> None:
         if not self.accumulate:
             self.values = list[Value]()
         if flat_values := flatten(values):
diff --git a/langgraph/checkpoint/base.py b/langgraph/checkpoint/base.py
index 562aba529..35525d395 100644
--- a/langgraph/checkpoint/base.py
+++ b/langgraph/checkpoint/base.py
@@ -2,7 +2,7 @@ import asyncio
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from datetime import datetime, timezone
-from typing import Any, TypedDict
+from typing import Any, Optional, TypedDict
 
 from langchain_core.load.serializable import Serializable
 from langchain_core.pydantic_v1 import Field
@@ -43,14 +43,14 @@ class BaseCheckpointSaver(Serializable, ABC):
         return []
 
     @abstractmethod
-    def get(self, config: RunnableConfig) -> Checkpoint | None:
+    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
         ...
 
     @abstractmethod
     def put(self, config: RunnableConfig, checkpoint: Checkpoint) -> None:
         ...
 
-    async def aget(self, config: RunnableConfig) -> Checkpoint | None:
+    async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
         return await asyncio.get_running_loop().run_in_executor(None, self.get, config)
 
     async def aput(self, config: RunnableConfig, checkpoint: Checkpoint) -> None:
diff --git a/langgraph/checkpoint/memory.py b/langgraph/checkpoint/memory.py
index 41a145397..7812c6f9a 100644
--- a/langgraph/checkpoint/memory.py
+++ b/langgraph/checkpoint/memory.py
@@ -1,3 +1,5 @@
+from typing import Optional
+
 from langchain_core.pydantic_v1 import Field
 from langchain_core.runnables import RunnableConfig
 from langchain_core.runnables.utils import ConfigurableFieldSpec
@@ -21,7 +23,7 @@ class MemorySaver(BaseCheckpointSaver):
             ),
         ]
 
-    def get(self, config: RunnableConfig) -> Checkpoint | None:
+    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
         return self.storage.get(config["configurable"]["thread_id"], None)
 
     def put(self, config: RunnableConfig, checkpoint: Checkpoint) -> None:
diff --git a/langgraph/pregel/__init__.py b/langgraph/pregel/__init__.py
index 33a0e49e3..cfc74a3d3 100644
--- a/langgraph/pregel/__init__.py
+++ b/langgraph/pregel/__init__.py
@@ -86,7 +86,7 @@ class Channel:
         cls,
         channels: str,
         key: Optional[str] = None,
-        when: Callable[[Any], bool] | None = None,
+        when: Optional[Callable[[Any], bool]] = None,
     ) -> ChannelInvoke:
         ...
 
@@ -96,16 +96,16 @@ class Channel:
         cls,
         channels: Sequence[str],
         key: None = None,
-        when: Callable[[Any], bool] | None = None,
+        when: Optional[Callable[[Any], bool]] = None,
     ) -> ChannelInvoke:
         ...
 
     @classmethod
     def subscribe_to(
         cls,
-        channels: str | Sequence[str],
+        channels: Union[str, Sequence[str]],
         key: Optional[str] = None,
-        when: Callable[[Any], bool] | None = None,
+        when: Optional[Callable[[Any], bool]] = None,
     ) -> ChannelInvoke:
         """Runs process.invoke() each time channels are updated,
         with a dict of the channel values as input."""
@@ -115,7 +115,7 @@ class Channel:
             )
         return ChannelInvoke(
             channels=cast(
-                Mapping[None, str] | Mapping[str, str],
+                Union[Mapping[None, str], Mapping[str, str]],
                 {key: channels}
                 if isinstance(channels, str)
                 else {chan: chan for chan in channels},
@@ -144,14 +144,16 @@ class Channel:
         )
 
 
-class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
-    nodes: Mapping[str, ChannelInvoke | ChannelBatch]
+class Pregel(
+    RunnableSerializable[Union[dict[str, Any], Any], Union[dict[str, Any], Any]]
+):
+    nodes: Mapping[str, Union[ChannelInvoke, ChannelBatch]]
 
     channels: Mapping[str, BaseChannel] = Field(default_factory=dict)
 
-    output: str | Sequence[str] = "output"
+    output: Union[str, Sequence[str]] = "output"
 
-    input: str | Sequence[str] = "input"
+    input: Union[str, Sequence[str]] = "input"
 
     step_timeout: Optional[float] = None
 
@@ -213,12 +215,12 @@ class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
 
     def _transform(
         self,
-        input: Iterator[dict[str, Any] | Any],
+        input: Iterator[Union[dict[str, Any], Any]],
         run_manager: CallbackManagerForChainRun,
         config: RunnableConfig,
         *,
-        output: str | Sequence[str] | None = None,
-    ) -> Iterator[tuple[dict[str, Any] | Any, CheckpointView]]:
+        output: Optional[Union[str, Sequence[str]]] = None,
+    ) -> Iterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
         if config["recursion_limit"] < 1:
             raise ValueError("recursion_limit must be at least 1")
         # assign defaults
@@ -321,12 +323,12 @@ class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
 
     async def _atransform(
         self,
-        input: AsyncIterator[dict[str, Any] | Any],
+        input: AsyncIterator[Union[dict[str, Any], Any]],
         run_manager: AsyncCallbackManagerForChainRun,
         config: RunnableConfig,
         *,
-        output: str | Sequence[str] | None = None,
-    ) -> AsyncIterator[tuple[dict[str, Any] | Any, CheckpointView]]:
+        output: Optional[Union[str, Sequence[str]]] = None,
+    ) -> AsyncIterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
         if config["recursion_limit"] < 1:
             raise ValueError("recursion_limit must be at least 1")
         # if running from astream_log() run each proc with streaming
@@ -441,13 +443,13 @@ class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
 
     def invoke(
         self,
-        input: dict[str, Any] | Any,
-        config: RunnableConfig | None = None,
+        input: Union[dict[str, Any], Any],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
+        output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
-    ) -> dict[str, Any] | Any:
-        latest: dict[str, Any] | Any = None
+    ) -> Union[dict[str, Any], Any]:
+        latest: Union[dict[str, Any], Any] = None
         for chunk in self.stream(
             input,
             config,
@@ -459,50 +461,50 @@ class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
 
     def stream(
         self,
-        input: dict[str, Any] | Any,
-        config: RunnableConfig | None = None,
+        input: Union[dict[str, Any], Any],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
+        output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
-    ) -> Iterator[dict[str, Any] | Any]:
+    ) -> Iterator[Union[dict[str, Any], Any]]:
         return self.transform(iter([input]), config, output=output, **kwargs)
 
     def transform(
         self,
-        input: Iterator[dict[str, Any] | Any],
-        config: RunnableConfig | None = None,
+        input: Iterator[Union[dict[str, Any], Any]],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
-        **kwargs: Any | None,
-    ) -> Iterator[dict[str, Any] | Any]:
+        output: Optional[Union[str, Sequence[str]]] = None,
+        **kwargs: Any,
+    ) -> Iterator[Union[dict[str, Any], Any]]:
         for out, _ in self._transform_stream_with_config(
             input, self._transform, config, output=output, **kwargs
         ):
             if out is not None:
-                yield cast(dict[str, Any] | Any, out)
+                yield cast(Union[dict[str, Any], Any], out)
 
     def step(
         self,
-        input: dict[str, Any] | Any,
-        config: RunnableConfig | None = None,
+        input: Union[dict[str, Any], Any],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
+        output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
-    ) -> Iterator[tuple[dict[str, Any] | Any, CheckpointView]]:
+    ) -> Iterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
         for tup in self._transform_stream_with_config(
             iter([input]), self._transform, config, output=output, **kwargs
         ):
-            yield cast(tuple[dict[str, Any] | Any, CheckpointView], tup)
+            yield cast(tuple[Union[dict[str, Any], Any], CheckpointView], tup)
 
     async def ainvoke(
         self,
-        input: dict[str, Any] | Any,
-        config: RunnableConfig | None = None,
+        input: Union[dict[str, Any], Any],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
+        output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
-    ) -> dict[str, Any] | Any:
-        latest: dict[str, Any] | Any = None
+    ) -> Union[dict[str, Any], Any]:
+        latest: Union[dict[str, Any], Any] = None
         async for chunk in self.astream(
             input,
             config,
@@ -514,13 +516,13 @@ class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
 
     async def astream(
         self,
-        input: dict[str, Any] | Any,
-        config: RunnableConfig | None = None,
+        input: Union[dict[str, Any], Any],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
+        output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
-    ) -> AsyncIterator[dict[str, Any] | Any]:
-        async def input_stream() -> AsyncIterator[dict[str, Any] | Any]:
+    ) -> AsyncIterator[Union[dict[str, Any], Any]]:
+        async def input_stream() -> AsyncIterator[Union[dict[str, Any], Any]]:
             yield input
 
         async for chunk in self.atransform(
@@ -530,12 +532,12 @@ class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
 
     async def atransform(
         self,
-        input: AsyncIterator[dict[str, Any] | Any],
-        config: RunnableConfig | None = None,
+        input: AsyncIterator[Union[dict[str, Any], Any]],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
-        **kwargs: Any | None,
-    ) -> AsyncIterator[dict[str, Any] | Any]:
+        output: Optional[Union[str, Sequence[str]]] = None,
+        **kwargs: Any,
+    ) -> AsyncIterator[Union[dict[str, Any], Any]]:
         async for out, _ in self._atransform_stream_with_config(
             input, self._atransform, config, output=output, **kwargs
         ):
@@ -544,24 +546,24 @@ class Pregel(RunnableSerializable[dict[str, Any] | Any, dict[str, Any] | Any]):
 
     async def astep(
         self,
-        input: dict[str, Any] | Any,
-        config: RunnableConfig | None = None,
+        input: Union[dict[str, Any], Any],
+        config: Optional[RunnableConfig] = None,
         *,
-        output: str | Sequence[str] | None = None,
+        output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
-    ) -> AsyncIterator[tuple[dict[str, Any] | Any, CheckpointView]]:
-        async def input_stream() -> AsyncIterator[dict[str, Any] | Any]:
+    ) -> AsyncIterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
+        async def input_stream() -> AsyncIterator[Union[dict[str, Any], Any]]:
             yield input
 
         async for tup in self._atransform_stream_with_config(
             input_stream(), self._atransform, config, output=output, **kwargs
         ):
-            yield cast(tuple[dict[str, Any] | Any, CheckpointView], tup)
+            yield cast(tuple[Union[dict[str, Any], Any], CheckpointView], tup)
 
 
 def _interrupt_or_proceed(
-    done: set[concurrent.futures.Future[Any]] | set[asyncio.Task[Any]],
-    inflight: set[concurrent.futures.Future[Any]] | set[asyncio.Task[Any]],
+    done: Union[set[concurrent.futures.Future[Any]], set[asyncio.Task[Any]]],
+    inflight: Union[set[concurrent.futures.Future[Any]], set[asyncio.Task[Any]]],
     step: int,
 ) -> None:
     while done:
@@ -645,7 +647,7 @@ def _apply_writes_from_view(
 
 def _prepare_next_tasks(
     checkpoint: Checkpoint,
-    processes: Mapping[str, ChannelInvoke | ChannelBatch],
+    processes: Mapping[str, Union[ChannelInvoke, ChannelBatch]],
     channels: Mapping[str, BaseChannel],
 ) -> list[tuple[Runnable, Any, str]]:
     tasks: list[tuple[Runnable, Any, str]] = []
diff --git a/langgraph/pregel/io.py b/langgraph/pregel/io.py
index ff30b5d87..b844b1e55 100644
--- a/langgraph/pregel/io.py
+++ b/langgraph/pregel/io.py
@@ -1,11 +1,12 @@
-from typing import Any, Iterator, Mapping, Sequence
+from typing import Any, Iterator, Mapping, Optional, Sequence, Union
 
 from langgraph.channels.base import BaseChannel
 from langgraph.pregel.log import logger
 
 
 def map_input(
-    input_channels: str | Sequence[str], chunk: dict[str, Any] | Any | None
+    input_channels: Union[str, Sequence[str]],
+    chunk: Optional[Union[dict[str, Any], Any]],
 ) -> Iterator[tuple[str, Any]]:
     """Map input chunk to a sequence of pending writes in the form (channel, value)."""
     if chunk is None:
@@ -23,7 +24,7 @@ def map_input(
 
 
 def map_output(
-    output_channels: str | Sequence[str],
+    output_channels: Union[str, Sequence[str]],
     pending_writes: Sequence[tuple[str, Any]],
     channels: Mapping[str, BaseChannel],
 ) -> dict[str, Any] | Any | None:
diff --git a/langgraph/pregel/read.py b/langgraph/pregel/read.py
index 923219ccb..33fadf46c 100644
--- a/langgraph/pregel/read.py
+++ b/langgraph/pregel/read.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Callable, List, Mapping, Optional, Sequence
+from typing import Any, Callable, Mapping, Optional, Sequence, Union
 
 from langchain_core.pydantic_v1 import Field
 from langchain_core.runnables import (
@@ -67,9 +67,9 @@ default_bound: RunnablePassthrough = RunnablePassthrough()
 
 
 class ChannelInvoke(RunnableBindingBase):
-    channels: Mapping[None, str] | Mapping[str, str]
+    channels: Union[Mapping[None, str], Mapping[str, str]]
 
-    triggers: List[str] = Field(default_factory=list)
+    triggers: list[str] = Field(default_factory=list)
 
     when: Optional[Callable[[Any], bool]] = None
 
@@ -119,9 +119,11 @@ class ChannelInvoke(RunnableBindingBase):
 
     def __or__(
         self,
-        other: Runnable[Any, Other]
-        | Callable[[Any], Other]
-        | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other]],
+        other: Union[
+            Runnable[Any, Other],
+            Callable[[Any], Other],
+            Mapping[str, Runnable[Any, Other] | Callable[[Any], Other]],
+        ],
     ) -> ChannelInvoke:
         if self.bound is default_bound:
             return ChannelInvoke(
@@ -145,9 +147,11 @@ class ChannelInvoke(RunnableBindingBase):
 
     def __ror__(
         self,
-        other: Runnable[Other, Any]
-        | Callable[[Any], Other]
-        | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any]],
+        other: Union[
+            Runnable[Other, Any],
+            Callable[[Any], Other],
+            Mapping[str, Union[Runnable[Other, Any], Callable[[Other], Any]]],
+        ],
     ) -> RunnableSerializable:
         raise NotImplementedError()
 
@@ -178,9 +182,11 @@ class ChannelBatch(RunnableEach):
 
     def __or__(  # type: ignore[override]
         self,
-        other: Runnable[Any, Other]
-        | Callable[[Any], Other]
-        | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other]],
+        other: Union[
+            Runnable[Any, Other],
+            Callable[[Any], Other],
+            Mapping[str, Runnable[Any, Other] | Callable[[Any], Other]],
+        ],
     ) -> ChannelBatch:
         if self.bound is default_bound:
             return ChannelBatch(
@@ -194,8 +200,10 @@ class ChannelBatch(RunnableEach):
 
     def __ror__(
         self,
-        other: Runnable[Other, Any]
-        | Callable[[Any], Other]
-        | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any]],
+        other: Union[
+            Runnable[Other, Any],
+            Callable[[Any], Other],
+            Mapping[str, Runnable[Other, Any] | Callable[[Other], Any]],
+        ],
     ) -> RunnableSerializable:
         raise NotImplementedError()
diff --git a/langgraph/pregel/validate.py b/langgraph/pregel/validate.py
index 813c3089f..096061665 100644
--- a/langgraph/pregel/validate.py
+++ b/langgraph/pregel/validate.py
@@ -1,4 +1,4 @@
-from typing import Any, Mapping, Sequence
+from typing import Any, Mapping, Sequence, Union
 
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.last_value import LastValue
@@ -7,10 +7,10 @@ from langgraph.pregel.reserved import ReservedChannels
 
 
 def validate_graph(
-    nodes: Mapping[str, ChannelInvoke | ChannelBatch],
+    nodes: Mapping[str, Union[ChannelInvoke, ChannelBatch]],
     channels: dict[str, BaseChannel],
-    input: str | Sequence[str],
-    output: str | Sequence[str],
+    input: Union[str, Sequence[str]],
+    output: Union[str, Sequence[str]],
 ) -> None:
     subscribed_channels = set[str]()
     for node in nodes.values():
diff --git a/langgraph/pregel/write.py b/langgraph/pregel/write.py
index e4be13e2e..f292303b5 100644
--- a/langgraph/pregel/write.py
+++ b/langgraph/pregel/write.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Callable, Sequence
+from typing import Any, Callable, Optional, Sequence
 
 from langchain_core.runnables import (
     Runnable,
@@ -15,7 +15,7 @@ TYPE_SEND = Callable[[Sequence[tuple[str, Any]]], None]
 
 
 class ChannelWrite(RunnablePassthrough):
-    channels: Sequence[tuple[str, Runnable | None]]
+    channels: Sequence[tuple[str, Optional[Runnable]]]
     """
     Mapping of write channels to Runnables that return the value to be written,
     or None to skip writing.
@@ -27,7 +27,7 @@ class ChannelWrite(RunnablePassthrough):
     def __init__(
         self,
         *,
-        channels: Sequence[tuple[str, Runnable | None]],
+        channels: Sequence[tuple[str, Optional[Runnable]]],
     ):
         super().__init__(func=self._write, afunc=self._awrite, channels=channels)
         self.name = f"ChannelWrite<{','.join(chan for chan, _ in self.channels)}>"
diff --git a/pyproject.toml b/pyproject.toml
index d13cd2cac..3ff1de7e0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -8,7 +8,7 @@ readme = "README.md"
 repository = "https://www.github.com/langchain-ai/langgraph"
 
 [tool.poetry.dependencies]
-python = ">=3.8.1,<4.0"
+python = ">=3.9.0,<4.0"
 langchain-core = "^0.1.8"