x

This PR adds validation to keys used in filtering logic in the SQLite store implementation.

docs/docs/cloud/reference/langgraph_server_changelog.md

@@ -4,6 +4,22 @@
 
 ---
 
+## v0.2.103 (2025-07-25)
+- Corrected the metadata endpoint to ensure accurate data retrieval.
+
+## v0.2.102 (2025-07-24)
+- Captured interrupt events in the wait method to preserve legacy behavior and stream updates by default.
+- Added support for SDK structlog in the JavaScript environment, enhancing logging capabilities.
+
+## v0.2.101 (2025-07-24)
+- Used the correct metadata endpoint for self-hosted environments, resolving an access issue.
+
+## v0.2.99 (2025-07-22)
+- Improved license validation by adding an in-memory cache and handling Redis connection errors more effectively.
+- Automatically remove agents from memory that are removed from `langgraph.json` to prevent persistence issues.
+- Ensured the UI namespace for generated UI is a valid JavaScript property name to prevent errors.
+- Raised a 422 error for improved request validation feedback.
+
 ## v0.2.98 (2025-07-19)
 - Added langgraph node context for improved log filtering and trace visibility.
 

docs/docs/concepts/human_in_the_loop.md

@@ -23,7 +23,7 @@ To review, edit, and approve tool calls in an agent or workflow, [use LangGraph'
 
 ## Key capabilities
 
-* **Persistent execution state**: Interrupts use LangGraph's [persistence](../../concepts/persistence.md) layer, which saves the graph state, to indefinitely pause graph execution until you resume. This is possible because LangGraph checkpoints the graph state after each step, which allows the system to persist execution context and later resume the workflow, continuing from where it left off. This supports asynchronous human review or input without time constraints.
+* **Persistent execution state**: Interrupts use LangGraph's [persistence](./persistence.md) layer, which saves the graph state, to indefinitely pause graph execution until you resume. This is possible because LangGraph checkpoints the graph state after each step, which allows the system to persist execution context and later resume the workflow, continuing from where it left off. This supports asynchronous human review or input without time constraints.
 
     There are two ways to pause a graph:
 

docs/docs/reference/runtime.md

@@ -0,0 +1,18 @@
+# Runtime
+
+::: langgraph.runtime.Runtime
+    options:
+      show_root_heading: true
+      show_root_full_path: false
+      members:
+        - context
+        - store
+        - stream_writer
+        - previous
+
+::: langgraph.runtime
+    options:
+      members:
+        - get_runtime
+
+

docs/mkdocs.yml

@@ -250,6 +250,7 @@ nav:
       - Storage: reference/store.md
       - Caching: reference/cache.md
       - Types: reference/types.md
+      - Runtime: reference/runtime.md
       - Config: reference/config.md
       - Errors: reference/errors.md
       - Constants: reference/constants.md

libs/checkpoint-sqlite/langgraph/store/sqlite/base.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import concurrent.futures
 import datetime
 import logging
+import re
 import sqlite3
 import threading
 from collections import defaultdict
@@ -107,6 +108,23 @@ def _decode_ns_text(namespace: str) -> tuple[str, ...]:
     return tuple(namespace.split("."))
 
 
+def _validate_filter_key(key: str) -> None:
+    """Validate that a filter key is safe for use in SQL queries.
+
+    Args:
+        key: The filter key to validate
+
+    Raises:
+        ValueError: If the key contains invalid characters that could enable SQL injection
+    """
+    # Allow alphanumeric characters, underscores, dots, and hyphens
+    # This covers typical JSON property names while preventing SQL injection
+    if not re.match(r"^[a-zA-Z0-9_.-]+$", key):
+        raise ValueError(
+            f"Invalid filter key: '{key}'. Filter keys must contain only alphanumeric characters, underscores, dots, and hyphens."
+        )
+
+
 def _json_loads(content: bytes | str | orjson.Fragment) -> Any:
     if isinstance(content, orjson.Fragment):
         if hasattr(content, "buf"):
@@ -372,6 +390,8 @@ class BaseSqliteStore:
             filter_conditions = []
             if op.filter:
                 for key, value in op.filter.items():
+                    _validate_filter_key(key)
+
                     if isinstance(value, dict):
                         for op_name, val in value.items():
                             condition, filter_params_ = self._get_filter_condition(
@@ -622,6 +642,8 @@ class BaseSqliteStore:
 
     def _get_filter_condition(self, key: str, op: str, value: Any) -> tuple[str, list]:
         """Helper to generate filter conditions."""
+        _validate_filter_key(key)
+
         # We need to properly format values for SQLite JSON extraction comparison
         if op == "$eq":
             if isinstance(value, str):
@@ -858,6 +880,8 @@ class SqliteStore(BaseSqliteStore, BaseStore):
 
     def _get_filter_condition(self, key: str, op: str, value: Any) -> tuple[str, list]:
         """Helper to generate filter conditions."""
+        _validate_filter_key(key)
+
         # We need to properly format values for SQLite JSON extraction comparison
         if op == "$eq":
             if isinstance(value, str):

libs/checkpoint-sqlite/tests/test_store.py

@@ -1047,3 +1047,23 @@ def test_search_items(
         for ns in test_namespaces:
             key = f"item_{ns[-1]}"
             store.delete(ns, key)
+
+
+def test_sql_injection_vulnerability(store: SqliteStore) -> None:
+    """Test that SQL injection via malicious filter keys is prevented."""
+    # Add public and private documents
+    store.put(("docs",), "public", {"access": "public", "data": "public info"})
+    store.put(
+        ("docs",), "private", {"access": "private", "data": "secret", "password": "123"}
+    )
+
+    # Normal query - returns 1 public document
+    normal = store.search(("docs",), filter={"access": "public"})
+    assert len(normal) == 1
+    assert normal[0].value["access"] == "public"
+
+    # SQL injection attempt via malicious key should raise ValueError
+    malicious_key = "access') = 'public' OR '1'='1' OR json_extract(value, '$."
+
+    with pytest.raises(ValueError, match="Invalid filter key"):
+        store.search(("docs",), filter={malicious_key: "dummy"})

libs/langgraph/langgraph/func/__init__.py

@@ -262,6 +262,11 @@ class entrypoint(Generic[ContextT]):
         cache_policy: A cache policy to use for caching the results of the workflow.
         retry_policy: A retry policy (or list of policies) to use for the workflow in case of a failure.
 
+    !!! warning "`config_schema` Deprecated"
+        The `config_schema` parameter is deprecated in v0.6.0 and support will be removed in v2.0.0.
+        Please use `context_schema` instead to specify the schema for run-scoped context.
+
+
     Example: Using entrypoint and tasks
         ```python
         import time

libs/langgraph/langgraph/graph/state.py

@@ -129,6 +129,10 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         input_schema: The schema class that defines the input to the graph.
         output_schema: The schema class that defines the output from the graph.
 
+    !!! warning "`config_schema` Deprecated"
+        The `config_schema` parameter is deprecated in v0.6.0 and support will be removed in v2.0.0.
+        Please use `context_schema` instead to specify the schema for run-scoped context.
+
     Example:
         ```python
         from langchain_core.runnables import RunnableConfig

libs/langgraph/langgraph/pregel/main.py

@@ -602,6 +602,7 @@ class Pregel(
     Defaults to None."""
 
     context_schema: type[ContextT] | None = None
+    """Specifies the schema for the context object that will be passed to the workflow."""
 
     config: RunnableConfig | None = None
 
@@ -2438,6 +2439,8 @@ class Pregel(
         Args:
             input: The input to the graph.
             config: The configuration to use for the run.
+            context: The static context to use for the run.
+                !!! version-added "Added in version 0.6.0."
             stream_mode: The mode to stream output, defaults to `self.stream_mode`.
                 Options are:
 
@@ -2694,6 +2697,8 @@ class Pregel(
         Args:
             input: The input to the graph.
             config: The configuration to use for the run.
+            context: The static context to use for the run.
+                !!! version-added "Added in version 0.6.0."
             stream_mode: The mode to stream output, defaults to `self.stream_mode`.
                 Options are:
 
@@ -2981,6 +2986,8 @@ class Pregel(
         Args:
             input: The input data for the graph. It can be a dictionary or any other type.
             config: Optional. The configuration for the graph run.
+            context: The static context to use for the run.
+                !!! version-added "Added in version 0.6.0."
             stream_mode: Optional[str]. The stream mode for the graph run. Default is "values".
             print_mode: Accepts the same values as `stream_mode`, but only prints the output to the console, for debugging purposes. Does not affect the output of the graph in any way.
             output_keys: Optional. The output keys to retrieve from the graph run.
@@ -3058,6 +3065,8 @@ class Pregel(
         Args:
             input: The input data for the computation. It can be a dictionary or any other type.
             config: Optional. The configuration for the computation.
+            context: The static context to use for the run.
+                !!! version-added "Added in version 0.6.0."
             stream_mode: Optional. The stream mode for the computation. Default is "values".
             print_mode: Accepts the same values as `stream_mode`, but only prints the output to the console, for debugging purposes. Does not affect the output of the graph in any way.
             output_keys: Optional. The output keys to include in the result. Default is None.

libs/langgraph/langgraph/pregel/remote.py

@@ -633,6 +633,7 @@ class RemoteGraph(PregelProtocol):
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         subgraphs: bool = False,
+        headers: dict[str, str] | None = None,
         **kwargs: Any,
     ) -> Iterator[dict[str, Any] | Any]:
         """Create a run and stream the results.
@@ -648,6 +649,7 @@ class RemoteGraph(PregelProtocol):
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
             subgraphs: Stream from subgraphs.
+            headers: Additional headers to pass to the request.
             **kwargs: Additional params to pass to client.runs.stream.
 
         Yields:
@@ -676,7 +678,9 @@ class RemoteGraph(PregelProtocol):
             interrupt_after=interrupt_after,
             stream_subgraphs=subgraphs or stream is not None,
             if_not_exists="create",
-            headers=self._merge_tracing_headers(kwargs.pop("headers", None) or {}),
+            headers=_merge_tracing_headers(headers)
+            if self.distributed_tracing
+            else headers,
             **kwargs,
         ):
             # split mode and ns
@@ -736,6 +740,7 @@ class RemoteGraph(PregelProtocol):
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         subgraphs: bool = False,
+        headers: dict[str, str] | None = None,
         **kwargs: Any,
     ) -> AsyncIterator[dict[str, Any] | Any]:
         """Create a run and stream the results.
@@ -751,6 +756,7 @@ class RemoteGraph(PregelProtocol):
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
             subgraphs: Stream from subgraphs.
+            headers: Additional headers to pass to the request.
             **kwargs: Additional params to pass to client.runs.stream.
 
         Yields:
@@ -779,7 +785,9 @@ class RemoteGraph(PregelProtocol):
             interrupt_after=interrupt_after,
             stream_subgraphs=subgraphs or stream is not None,
             if_not_exists="create",
-            headers=self._merge_tracing_headers(kwargs.pop("headers", None) or {}),
+            headers=_merge_tracing_headers(headers)
+            if self.distributed_tracing
+            else headers,
             **kwargs,
         ):
             # split mode and ns
@@ -853,6 +861,7 @@ class RemoteGraph(PregelProtocol):
         *,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
+        headers: dict[str, str] | None = None,
         **kwargs: Any,
     ) -> dict[str, Any] | Any:
         """Create a run, wait until it finishes and return the final state.
@@ -862,6 +871,7 @@ class RemoteGraph(PregelProtocol):
             config: A `RunnableConfig` for graph invocation.
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
+            headers: Additional headers to pass to the request.
             **kwargs: Additional params to pass to RemoteGraph.stream.
 
         Returns:
@@ -872,6 +882,7 @@ class RemoteGraph(PregelProtocol):
             config=config,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
+            headers=headers,
             stream_mode="values",
             **kwargs,
         ):
@@ -888,6 +899,7 @@ class RemoteGraph(PregelProtocol):
         *,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
+        headers: dict[str, str] | None = None,
         **kwargs: Any,
     ) -> dict[str, Any] | Any:
         """Create a run, wait until it finishes and return the final state.
@@ -897,6 +909,7 @@ class RemoteGraph(PregelProtocol):
             config: A `RunnableConfig` for graph invocation.
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
+            headers: Additional headers to pass to the request.
             **kwargs: Additional params to pass to RemoteGraph.astream.
 
         Returns:
@@ -907,6 +920,7 @@ class RemoteGraph(PregelProtocol):
             config=config,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
+            headers=headers,
             stream_mode="values",
             **kwargs,
         ):
@@ -916,12 +930,16 @@ class RemoteGraph(PregelProtocol):
         except UnboundLocalError:
             return None
 
-    def _merge_tracing_headers(self, headers: dict[str, str]) -> dict[str, str]:
-        if rt := ls.get_current_run_tree():
-            tracing_headers = rt.to_headers()
-            baggage = tracing_headers.pop("baggage")
+
+def _merge_tracing_headers(headers: dict[str, str] | None) -> dict[str, str] | None:
+    if rt := ls.get_current_run_tree():
+        tracing_headers = rt.to_headers()
+        baggage = tracing_headers.pop("baggage")
+        if headers:
             if "baggage" in headers:
                 baggage = headers["baggage"] + "," + baggage
             tracing_headers["baggage"] = baggage
             headers.update(tracing_headers)
-        return headers
+        else:
+            headers = tracing_headers
+    return headers

libs/langgraph/langgraph/runtime.py

@@ -11,6 +11,8 @@ from langgraph.store.base import BaseStore
 from langgraph.types import _DC_KWARGS, StreamWriter
 from langgraph.typing import ContextT
 
+__all__ = ("Runtime", "get_runtime")
+
 
 def _no_op_stream_writer(_: Any) -> None: ...
 
@@ -24,9 +26,61 @@ class _RuntimeOverrides(TypedDict, Generic[ContextT], total=False):
 
 @dataclass(**_DC_KWARGS)
 class Runtime(Generic[ContextT]):
-    """Convenience class that bundles run-scoped context and graph configuration.
+    """Convenience class that bundles run-scoped context and other runtime utilities.
 
-    !!! version-added "Added in version 1.0.0."
+    !!! version-added "Added in version v0.6.0"
+
+    Example:
+
+    ```python
+    from typing import TypedDict
+    from langgraph.graph import StateGraph
+    from dataclasses import dataclass
+    from langgraph.runtime import Runtime
+    from langgraph.store.memory import InMemoryStore
+
+
+    @dataclass
+    class Context:  # (1)!
+        user_id: str
+
+
+    class State(TypedDict, total=False):
+        response: str
+
+
+    store = InMemoryStore()  # (2)!
+    store.put(("users",), "user_123", {"name": "Alice"})
+
+
+    def personalized_greeting(state: State, runtime: Runtime[Context]) -> State:
+        '''Generate personalized greeting using runtime context and store.'''
+        user_id = runtime.context.user_id  # (3)!
+        name = "unknown_user"
+        if runtime.store:
+            if memory := runtime.store.get(("users",), user_id):
+                name = memory.value["name"]
+
+        response = f"Hello {name}! Nice to see you again."
+        return {"response": response}
+
+
+    graph = (
+        StateGraph(state_schema=State, context_schema=Context)
+        .add_node("personalized_greeting", personalized_greeting)
+        .set_entry_point("personalized_greeting")
+        .set_finish_point("personalized_greeting")
+        .compile(store=store)
+    )
+
+    result = graph.invoke({}, context=Context(user_id="user_123"))
+    print(result)
+    # > {'response': 'Hello Alice! Nice to see you again.'}
+    ```
+
+    1. Define a schema for the runtime context.
+    2. Create a store to persist memories and other information.
+    3. Use the runtime context to access the user_id.
     """
 
     context: ContextT = field(default=None)  # type: ignore[assignment]
@@ -76,7 +130,14 @@ DEFAULT_RUNTIME = Runtime(
 
 
 def get_runtime(context_schema: type[ContextT] | None = None) -> Runtime[ContextT]:
-    """Get the runtime for the current graph run."""
+    """Get the runtime for the current graph run.
+
+    Args:
+        context_schema: Optional schema used for type hinting the return type of the runtime.
+
+    Returns:
+        The runtime for the current graph run.
+    """
 
     # TODO: in an ideal world, we would have a context manager for
     # the runtime that's independent of the config. this will follow

libs/langgraph/langgraph/types.py

@@ -149,10 +149,25 @@ class Interrupt:
     """Information about an interrupt that occurred in a node.
 
     !!! version-added "Added in version 0.2.24."
+
+    !!! version-changed "Changed in version v0.4.0"
+        * `interrupt_id` was introduced as a property
+
+    !!! version-changed "Changed in version v0.6.0"
+
+        The following attributes have been removed:
+
+        * `ns`
+        * `when`
+        * `resumable`
+        * `interrupt_id`, deprecated in favor of `id`
     """
 
     value: Any
+    """The value associated with the interrupt."""
+
     id: str
+    """The ID of the interrupt. Can be used to resume the interrupt directly."""
 
     def __init__(
         self,

libs/langgraph/pyproject.toml

@@ -15,7 +15,7 @@ dependencies = [
     "langchain-core>=0.1",
     "langgraph-checkpoint>=2.1.0,<3.0.0",
     "langgraph-sdk>=0.2.0,<0.3.0",
-    "langgraph-prebuilt>=0.5.0,<0.6.0",
+    "langgraph-prebuilt>=0.6.0,<0.7.0",
     "xxhash>=3.5.0",
     "pydantic>=2.7.4",
 ]

libs/langgraph/tests/test_remote_graph.py

@@ -3,6 +3,7 @@ import sys
 from typing import Annotated, Union
 from unittest.mock import AsyncMock, MagicMock
 
+import langsmith as ls
 import pytest
 from langchain_core.messages import AnyMessage, BaseMessage
 from langchain_core.runnables import RunnableConfig
@@ -899,21 +900,20 @@ async def test_langgraph_cloud_integration():
     }
 
     # test invoke
-    response = app.invoke(
+    app.invoke(
         input,
         config={"configurable": {"thread_id": "39a6104a-34e7-4f83-929c-d9eb163003c9"}},
         interrupt_before=["agent"],
     )
-    print("response:", response["messages"][-1].content)
 
     # test stream
-    async for chunk in app.astream(
+    async for _ in app.astream(
         input,
         config={"configurable": {"thread_id": "2dc3e3e7-39ac-4597-aa57-4404b944e82a"}},
         subgraphs=True,
         stream_mode=["debug", "messages"],
     ):
-        print("chunk:", chunk)
+        pass
 
     # test stream events
     async for chunk in remote_pregel.astream_events(
@@ -923,17 +923,16 @@ async def test_langgraph_cloud_integration():
         subgraphs=True,
         stream_mode=[],
     ):
-        print("chunk:", chunk)
+        pass
 
     # test get state
-    state_snapshot = await remote_pregel.aget_state(
+    await remote_pregel.aget_state(
         config={"configurable": {"thread_id": "2dc3e3e7-39ac-4597-aa57-4404b944e82a"}},
         subgraphs=True,
     )
-    print("state snapshot:", state_snapshot)
 
     # test update state
-    response = await remote_pregel.aupdate_state(
+    await remote_pregel.aupdate_state(
         config={"configurable": {"thread_id": "6645e002-ed50-4022-92a3-d0d186fdf812"}},
         values={
             "messages": [
@@ -944,18 +943,16 @@ async def test_langgraph_cloud_integration():
             ]
         },
     )
-    print("response:", response)
 
     # test get history
     async for state in remote_pregel.aget_state_history(
         config={"configurable": {"thread_id": "2dc3e3e7-39ac-4597-aa57-4404b944e82a"}},
     ):
-        print("state snapshot:", state)
+        pass
 
     # test get graph
     remote_pregel.graph_id = "fe096781-5601-53d2-b2f6-0d3403f7e9ca"  # must be UUID
-    graph = await remote_pregel.aget_graph(xray=True)
-    print("graph:", graph)
+    await remote_pregel.aget_graph(xray=True)
 
 
 def test_sanitize_config():
@@ -1181,3 +1178,73 @@ async def test_remote_graph_stream_messages_tuple(
     assert coerced_events == coerced_inmem_events
     # TODO: Fix the namespace matching in the next api release.
     # assert namespaces == inmem_namespaces
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("distributed_tracing", [False, True])
+@pytest.mark.parametrize("stream", [False, True])
+async def test_include_headers(distributed_tracing: bool, stream: bool):
+    mock_async_client = MagicMock()
+    async_iter = MagicMock()
+    return_value = [
+        StreamPart(event="values", data={"chunk": "data1"}),
+    ]
+    async_iter.__aiter__.return_value = return_value
+    astream_mock = mock_async_client.runs.stream
+    astream_mock.return_value = async_iter
+
+    mock_sync_client = MagicMock()
+    sync_iter = MagicMock()
+    sync_iter.__iter__.return_value = return_value
+    stream_mock = mock_sync_client.runs.stream
+    stream_mock.return_value = async_iter
+
+    remote_pregel = RemoteGraph(
+        "test_graph_id",
+        client=mock_async_client,
+        sync_client=mock_sync_client,
+        distributed_tracing=distributed_tracing,
+    )
+
+    config = {"configurable": {"thread_id": "thread_1"}}
+    with ls.tracing_context(enabled=True, client=MagicMock()):
+        with ls.trace("foo"):
+            if stream:
+                async for _ in remote_pregel.astream(
+                    {"input": {"messages": [{"type": "human", "content": "hello"}]}},
+                    config,
+                    headers={"foo": "bar"},
+                ):
+                    pass
+
+            else:
+                await remote_pregel.ainvoke(
+                    {"input": {"messages": [{"type": "human", "content": "hello"}]}},
+                    config,
+                    headers={"foo": "bar"},
+                )
+    expected = {"foo": "bar"}
+    if distributed_tracing:
+        expected["langsmith-trace"] = AnyStr()
+        expected["baggage"] = AnyStr()
+
+    assert astream_mock.call_args.kwargs["headers"] == expected
+    stream_mock.assert_not_called()
+
+    with ls.tracing_context(enabled=True, client=MagicMock()):
+        with ls.trace("foo"):
+            if stream:
+                for _ in remote_pregel.stream(
+                    {"input": {"messages": [{"type": "human", "content": "hello"}]}},
+                    config,
+                    headers={"foo": "bar"},
+                ):
+                    pass
+
+            else:
+                remote_pregel.invoke(
+                    {"input": {"messages": [{"type": "human", "content": "hello"}]}},
+                    config,
+                    headers={"foo": "bar"},
+                )
+    assert stream_mock.call_args.kwargs["headers"] == expected

libs/langgraph/uv.lock

@@ -1394,7 +1394,7 @@ dev = [
 
 [[package]]
 name = "langgraph-cli"
-version = "0.3.5"
+version = "0.3.6"
 source = { editable = "../cli" }
 dependencies = [
     { name = "click", version = "8.1.8", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
@@ -1433,7 +1433,7 @@ dev = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "0.5.2"
+version = "0.6.0"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

@@ -245,6 +245,435 @@ def _validate_chat_history(
     raise ValueError(error_message)
 
 
+class _AgentBuilder:
+    """Internal builder class for constructing React agents with intuitive method-to-node mapping."""
+
+    def __init__(
+        self,
+        model: Union[str, LanguageModelLike],
+        tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
+        *,
+        prompt: Optional[Prompt] = None,
+        response_format: Optional[
+            Union[StructuredResponseSchema, tuple[str, StructuredResponseSchema]]
+        ] = None,
+        pre_model_hook: Optional[RunnableLike] = None,
+        post_model_hook: Optional[RunnableLike] = None,
+        state_schema: Optional[StateSchemaType] = None,
+        context_schema: Optional[Type[Any]] = None,
+        version: Literal["v1", "v2"] = "v2",
+        name: Optional[str] = None,
+    ):
+        # Store all parameters
+        self.model = model
+        self.tools = tools
+        self.prompt = prompt
+        self.response_format = response_format
+        self.pre_model_hook = pre_model_hook
+        self.post_model_hook = post_model_hook
+        self.state_schema = state_schema
+        self.context_schema = context_schema
+        self.version = version
+        self.name = name
+
+        # Setup tools
+        if isinstance(self.tools, ToolNode):
+            self._tool_classes = list(self.tools.tools_by_name.values())
+            self._tool_node = self.tools
+        else:
+            self._llm_builtin_tools = [t for t in self.tools if isinstance(t, dict)]
+            self._tool_node = ToolNode(
+                [t for t in self.tools if not isinstance(t, dict)]
+            )
+            self._tool_classes = list(self._tool_node.tools_by_name.values())
+
+        self._should_return_direct: set[str] = {
+            t.name for t in self._tool_classes if t.return_direct
+        }
+
+        # Setup state schema
+        if self.state_schema is not None:
+            required_keys = {"messages", "remaining_steps"}
+            if self.response_format is not None:
+                required_keys.add("structured_response")
+
+            schema_keys = set(get_type_hints(self.state_schema))
+            if missing_keys := required_keys - schema_keys:
+                raise ValueError(
+                    f"Missing required key(s) {missing_keys} in state_schema"
+                )
+
+            self._final_state_schema = self.state_schema
+        else:
+            self._final_state_schema = (
+                AgentStateWithStructuredResponse
+                if self.response_format is not None
+                else AgentState
+            )
+
+        # Setup model
+        model = self.model
+
+        # Convert string models
+        if isinstance(model, str):
+            try:
+                from langchain.chat_models import init_chat_model  # type: ignore[import-not-found]
+            except ImportError:
+                raise ImportError(
+                    "Please install langchain (`pip install langchain`) to use '<provider>:<model>' string syntax for `model` parameter."
+                )
+            model = cast(BaseChatModel, init_chat_model(model))
+
+        # Bind tools if needed
+        if (
+            _should_bind_tools(
+                model, self._tool_classes, num_builtin=len(self._llm_builtin_tools)
+            )
+            and len(self._tool_classes + self._llm_builtin_tools) > 0
+        ):
+            model = cast(BaseChatModel, model).bind_tools(
+                self._tool_classes + self._llm_builtin_tools
+            )  # type: ignore[operator]
+
+        self._model_runnable = _get_prompt_runnable(self.prompt) | model
+
+    def create_model_node(self) -> RunnableCallable:
+        """Create the 'agent' node that calls the LLM."""
+
+        def _get_model_input_state(state: StateSchema) -> StateSchema:
+            if self.pre_model_hook is not None:
+                messages: Optional[Sequence[BaseMessage]] = (
+                    _get_state_value(state, "llm_input_messages")
+                ) or _get_state_value(state, "messages")
+                error_msg: str = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
+            else:
+                messages = _get_state_value(state, "messages")
+                error_msg = f"Expected input to call_model to have 'messages' key, but got {state}"
+
+            if messages is None:
+                raise ValueError(error_msg)
+
+            _validate_chat_history(messages)
+
+            if isinstance(self._final_state_schema, type) and issubclass(
+                self._final_state_schema, BaseModel
+            ):
+                state.messages = messages  # type: ignore
+            else:
+                state["messages"] = messages  # type: ignore
+            return state
+
+        def _are_more_steps_needed(state: StateSchema, response: BaseMessage) -> bool:
+            has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
+            all_tools_return_direct = (
+                all(
+                    call["name"] in self._should_return_direct
+                    for call in response.tool_calls
+                )
+                if isinstance(response, AIMessage)
+                else False
+            )
+            remaining_steps = _get_state_value(state, "remaining_steps", None)
+            is_last_step = _get_state_value(state, "is_last_step", False)
+            return (
+                (remaining_steps is None and is_last_step and has_tool_calls)
+                or (
+                    remaining_steps is not None
+                    and remaining_steps < 1
+                    and all_tools_return_direct
+                )
+                or (
+                    remaining_steps is not None
+                    and remaining_steps < 2
+                    and has_tool_calls
+                )
+            )
+
+        def call_model(state: StateSchema, config: RunnableConfig) -> StateSchema:
+            state = _get_model_input_state(state)
+            response = cast(AIMessage, self._model_runnable.invoke(state, config))  # type: ignore[union-attr]
+            response.name = self.name
+
+            if _are_more_steps_needed(state, response):
+                return {
+                    "messages": [
+                        AIMessage(
+                            id=response.id,
+                            content="Sorry, need more steps to process this request.",
+                        )
+                    ]
+                }
+            return {"messages": [response]}
+
+        async def acall_model(
+            state: StateSchema, config: RunnableConfig
+        ) -> StateSchema:
+            state = _get_model_input_state(state)
+            response = cast(
+                AIMessage, await self._model_runnable.ainvoke(state, config)
+            )  # type: ignore[union-attr]
+            response.name = self.name
+
+            if _are_more_steps_needed(state, response):
+                return {
+                    "messages": [
+                        AIMessage(
+                            id=response.id,
+                            content="Sorry, need more steps to process this request.",
+                        )
+                    ]
+                }
+            return {"messages": [response]}
+
+        # Determine input schema
+        input_schema = self._final_state_schema
+        if self.pre_model_hook is not None:
+            if isinstance(self._final_state_schema, type) and issubclass(
+                self._final_state_schema, BaseModel
+            ):
+                from pydantic import create_model
+
+                input_schema = create_model(
+                    "CallModelInputSchema",
+                    llm_input_messages=(list[AnyMessage], ...),
+                    __base__=self._final_state_schema,
+                )
+            else:
+
+                class CallModelInputSchema(self._final_state_schema):  # type: ignore
+                    llm_input_messages: list[AnyMessage]
+
+                input_schema = CallModelInputSchema
+
+        return RunnableCallable(call_model, acall_model, input_schema=input_schema)
+
+    def create_structured_response_node(self) -> Optional[RunnableCallable]:
+        """Create the 'generate_structured_response' node if configured."""
+        if self.response_format is None:
+            return None
+
+        def generate_structured_response(
+            state: StateSchema, config: RunnableConfig
+        ) -> StateSchema:
+            messages = _get_state_value(state, "messages")
+            structured_response_schema = self.response_format
+            if isinstance(self.response_format, tuple):
+                system_prompt, structured_response_schema = self.response_format
+                messages = [SystemMessage(content=system_prompt)] + list(messages)
+
+            model_with_structured_output = _get_model(
+                self._model_runnable
+            ).with_structured_output(  # type: ignore[arg-type]
+                cast(StructuredResponseSchema, structured_response_schema)
+            )
+            response = model_with_structured_output.invoke(messages, config)
+            return {"structured_response": response}
+
+        async def agenerate_structured_response(
+            state: StateSchema, config: RunnableConfig
+        ) -> StateSchema:
+            messages = _get_state_value(state, "messages")
+            structured_response_schema = self.response_format
+            if isinstance(self.response_format, tuple):
+                system_prompt, structured_response_schema = self.response_format
+                messages = [SystemMessage(content=system_prompt)] + list(messages)
+
+            model_with_structured_output = _get_model(
+                self._model_runnable
+            ).with_structured_output(  # type: ignore[arg-type]
+                cast(StructuredResponseSchema, structured_response_schema)
+            )
+            response = await model_with_structured_output.ainvoke(messages, config)
+            return {"structured_response": response}
+
+        return RunnableCallable(
+            generate_structured_response, agenerate_structured_response
+        )
+
+    def create_model_router(self) -> Callable[[StateSchema], Union[str, list[Send]]]:
+        """Create routing function for model node conditional edges."""
+
+        def should_continue(state: StateSchema) -> Union[str, list[Send]]:
+            messages = _get_state_value(state, "messages")
+            last_message = messages[-1]
+
+            if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
+                if self.post_model_hook is not None:
+                    return "post_model_hook"
+                elif self.response_format is not None:
+                    return "generate_structured_response"
+                else:
+                    return END
+            else:
+                if self.version == "v1":
+                    return "tools"
+                elif self.version == "v2":
+                    if self.post_model_hook is not None:
+                        return "post_model_hook"
+                    return [
+                        Send(
+                            "tools",
+                            ToolCallWithContext(
+                                __type="tool_call_with_context",
+                                tool_call=tool_call,
+                                state=state,
+                            ),
+                        )
+                        for tool_call in last_message.tool_calls
+                    ]
+
+        return should_continue
+
+    def post_model_hook_router(self, state: StateSchema) -> Union[str, list[Send]]:
+        """Route to the next node after post_model_hook."""
+        messages = _get_state_value(state, "messages")
+        tool_messages = [m.tool_call_id for m in messages if isinstance(m, ToolMessage)]
+        last_ai_message = next(
+            m for m in reversed(messages) if isinstance(m, AIMessage)
+        )
+        pending_tool_calls = [
+            c for c in last_ai_message.tool_calls if c["id"] not in tool_messages
+        ]
+
+        if pending_tool_calls:
+            return [
+                Send(
+                    "tools",
+                    ToolCallWithContext(
+                        __type="tool_call_with_context",
+                        tool_call=tool_call,
+                        state=state,
+                    ),
+                )
+                for tool_call in pending_tool_calls
+            ]
+        elif isinstance(messages[-1], ToolMessage):
+            return self._get_entry_point()
+        elif self.response_format is not None:
+            return "generate_structured_response"
+        else:
+            return END
+
+    def create_tools_router(self) -> Optional[Callable[[StateSchema], str]]:
+        """Create routing function for tools node conditional edges."""
+        if not self._should_return_direct:
+            return None
+
+        def route_tool_responses(state: StateSchema) -> str:
+            messages = _get_state_value(state, "messages")
+            for m in reversed(messages):
+                if not isinstance(m, ToolMessage):
+                    break
+                if m.name in self._should_return_direct:
+                    return END
+
+            if isinstance(m, AIMessage) and m.tool_calls:
+                if any(
+                    call["name"] in self._should_return_direct for call in m.tool_calls
+                ):
+                    return END
+
+            return self._get_entry_point()
+
+        return route_tool_responses
+
+    def _get_entry_point(self) -> str:
+        """Get the workflow entry point."""
+        return "pre_model_hook" if self.pre_model_hook else "agent"
+
+    def _has_tools(self) -> bool:
+        """Check if agent has tools enabled."""
+        return len(self._tool_classes) > 0
+
+    def _get_model_edges(self) -> list[str]:
+        """Get possible edge destinations from model node."""
+        edges = []
+
+        # If post_model_hook exists, we don't add edges here - we use direct edge instead
+        if not self.post_model_hook:
+            if self._has_tools():
+                edges.append("tools")
+            if self.response_format:
+                edges.append("generate_structured_response")
+            if not self._has_tools() and not self.response_format:
+                edges.append(END)
+
+        return edges
+
+    def _get_post_model_hook_edges(self) -> list[str]:
+        """Get possible edge destinations from post_model_hook node."""
+        edges = [self._get_entry_point()]
+        if self._has_tools():
+            edges.append("tools")
+        if self.response_format:
+            edges.append("generate_structured_response")
+        else:
+            edges.append(END)
+        return edges
+
+    def build(self) -> StateGraph:
+        """Build the agent workflow graph (uncompiled)."""
+        # Create workflow
+        workflow = StateGraph(
+            state_schema=self._final_state_schema,  # type: ignore[arg-type]
+            context_schema=self.context_schema,
+        )
+
+        # Add nodes
+        # Always add model node (named 'agent' for backwards compatibility)
+        workflow.add_node("agent", self.create_model_node())
+
+        # Add tools node if needed
+        if self._has_tools():
+            workflow.add_node("tools", self._tool_node)
+
+        # Add hook nodes if configured
+        if self.pre_model_hook:
+            workflow.add_node("pre_model_hook", self.pre_model_hook)  # type: ignore[arg-type]
+        if self.post_model_hook:
+            workflow.add_node("post_model_hook", self.post_model_hook)  # type: ignore[arg-type]
+
+        # Add structured response node if configured
+        structured_node = self.create_structured_response_node()
+        if structured_node:
+            workflow.add_node("generate_structured_response", structured_node)
+
+        # Add edges
+        entry_point = self._get_entry_point()
+        workflow.set_entry_point(entry_point)
+
+        # Pre-model hook edge
+        if self.pre_model_hook:
+            workflow.add_edge("pre_model_hook", "agent")
+
+        # Model node edges
+        if self.post_model_hook:
+            # Direct edge from model node to post_model_hook when post_model_hook exists
+            workflow.add_edge("agent", "post_model_hook")
+            # Post-model hook conditional edges
+            post_hook_edges = self._get_post_model_hook_edges()
+            workflow.add_conditional_edges(
+                "post_model_hook", self.post_model_hook_router, path_map=post_hook_edges
+            )  # type: ignore[arg-type]
+        else:
+            # Conditional edges from model node when no post_model_hook
+            model_router = self.create_model_router()
+            model_edges = self._get_model_edges()
+            workflow.add_conditional_edges("agent", model_router, path_map=model_edges)  # type: ignore[arg-type]
+
+        # Tools edges
+        if self._has_tools():
+            tools_router = self.create_tools_router()
+            if tools_router:
+                workflow.add_conditional_edges(
+                    "tools", tools_router, path_map=[entry_point, END]
+                )
+            else:
+                workflow.add_edge("tools", entry_point)
+
+        return workflow
+
+
 def create_react_agent(
     model: Union[str, LanguageModelLike],
     tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
@@ -364,6 +793,11 @@ def create_react_agent(
             This name will be automatically used when adding ReAct agent graph to another graph as a subgraph node -
             particularly useful for building multi-agent systems.
 
+    !!! warning "`config_schema` Deprecated"
+        The `config_schema` parameter is deprecated in v0.6.0 and support will be removed in v2.0.0.
+        Please use `context_schema` instead to specify the schema for run-scoped context.
+
+
     Returns:
         A compiled LangChain runnable that can be used for chat interactions.
 
@@ -406,6 +840,7 @@ def create_react_agent(
             print(chunk)
         ```
     """
+    # Handle deprecated config_schema parameter
     if (
         config_schema := deprecated_kwargs.pop("config_schema", MISSING)
     ) is not MISSING:
@@ -417,400 +852,29 @@ def create_react_agent(
         if context_schema is not None:
             context_schema = config_schema
 
+    # Validate version
     if version not in ("v1", "v2"):
         raise ValueError(
             f"Invalid version {version}. Supported versions are 'v1' and 'v2'."
         )
 
-    if state_schema is not None:
-        required_keys = {"messages", "remaining_steps"}
-        if response_format is not None:
-            required_keys.add("structured_response")
-
-        schema_keys = set(get_type_hints(state_schema))
-        if missing_keys := required_keys - set(schema_keys):
-            raise ValueError(f"Missing required key(s) {missing_keys} in state_schema")
-
-    if state_schema is None:
-        state_schema = (
-            AgentStateWithStructuredResponse
-            if response_format is not None
-            else AgentState
-        )
-
-    llm_builtin_tools: list[dict] = []
-    if isinstance(tools, ToolNode):
-        tool_classes = list(tools.tools_by_name.values())
-        tool_node = tools
-    else:
-        llm_builtin_tools = [t for t in tools if isinstance(t, dict)]
-        tool_node = ToolNode([t for t in tools if not isinstance(t, dict)])
-        tool_classes = list(tool_node.tools_by_name.values())
-
-    if isinstance(model, str):
-        try:
-            from langchain.chat_models import (  # type: ignore[import-not-found]
-                init_chat_model,
-            )
-        except ImportError:
-            raise ImportError(
-                "Please install langchain (`pip install langchain`) to use '<provider>:<model>' string syntax for `model` parameter."
-            )
-
-        model = cast(BaseChatModel, init_chat_model(model))
-
-    tool_calling_enabled = len(tool_classes) > 0
-
-    if (
-        _should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools))
-        and len(tool_classes + llm_builtin_tools) > 0
-    ):
-        model = cast(BaseChatModel, model).bind_tools(tool_classes + llm_builtin_tools)  # type: ignore[operator]
-
-    model_runnable = _get_prompt_runnable(prompt) | model
-
-    # If any of the tools are configured to return_directly after running,
-    # our graph needs to check if these were called
-    should_return_direct = {t.name for t in tool_classes if t.return_direct}
-
-    def _are_more_steps_needed(state: StateSchema, response: BaseMessage) -> bool:
-        has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
-        all_tools_return_direct = (
-            all(call["name"] in should_return_direct for call in response.tool_calls)
-            if isinstance(response, AIMessage)
-            else False
-        )
-        remaining_steps = _get_state_value(state, "remaining_steps", None)
-        is_last_step = _get_state_value(state, "is_last_step", False)
-        return (
-            (remaining_steps is None and is_last_step and has_tool_calls)
-            or (
-                remaining_steps is not None
-                and remaining_steps < 1
-                and all_tools_return_direct
-            )
-            or (remaining_steps is not None and remaining_steps < 2 and has_tool_calls)
-        )
-
-    def _get_model_input_state(state: StateSchema) -> StateSchema:
-        if pre_model_hook is not None:
-            messages = (
-                _get_state_value(state, "llm_input_messages")
-            ) or _get_state_value(state, "messages")
-            error_msg = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
-        else:
-            messages = _get_state_value(state, "messages")
-            error_msg = (
-                f"Expected input to call_model to have 'messages' key, but got {state}"
-            )
-
-        if messages is None:
-            raise ValueError(error_msg)
-
-        _validate_chat_history(messages)
-        # we're passing messages under `messages` key, as this is expected by the prompt
-        if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
-            state.messages = messages  # type: ignore
-        else:
-            state["messages"] = messages  # type: ignore
-
-        return state
-
-    # Define the function that calls the model
-    def call_model(state: StateSchema, config: RunnableConfig) -> StateSchema:
-        state = _get_model_input_state(state)
-        response = cast(AIMessage, model_runnable.invoke(state, config))
-        # add agent name to the AIMessage
-        response.name = name
-
-        if _are_more_steps_needed(state, response):
-            return {
-                "messages": [
-                    AIMessage(
-                        id=response.id,
-                        content="Sorry, need more steps to process this request.",
-                    )
-                ]
-            }
-        # We return a list, because this will get added to the existing list
-        return {"messages": [response]}
-
-    async def acall_model(state: StateSchema, config: RunnableConfig) -> StateSchema:
-        state = _get_model_input_state(state)
-        response = cast(AIMessage, await model_runnable.ainvoke(state, config))
-        # add agent name to the AIMessage
-        response.name = name
-        if _are_more_steps_needed(state, response):
-            return {
-                "messages": [
-                    AIMessage(
-                        id=response.id,
-                        content="Sorry, need more steps to process this request.",
-                    )
-                ]
-            }
-        # We return a list, because this will get added to the existing list
-        return {"messages": [response]}
-
-    input_schema: StateSchemaType
-    if pre_model_hook is not None:
-        # Dynamically create a schema that inherits from state_schema and adds 'llm_input_messages'
-        if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
-            # For Pydantic schemas
-            from pydantic import create_model
-
-            input_schema = create_model(
-                "CallModelInputSchema",
-                llm_input_messages=(list[AnyMessage], ...),
-                __base__=state_schema,
-            )
-        else:
-            # For TypedDict schemas
-            class CallModelInputSchema(state_schema):  # type: ignore
-                llm_input_messages: list[AnyMessage]
-
-            input_schema = CallModelInputSchema
-    else:
-        input_schema = state_schema
-
-    def generate_structured_response(
-        state: StateSchema, config: RunnableConfig
-    ) -> StateSchema:
-        messages = _get_state_value(state, "messages")
-        structured_response_schema = response_format
-        if isinstance(response_format, tuple):
-            system_prompt, structured_response_schema = response_format
-            messages = [SystemMessage(content=system_prompt)] + list(messages)
-
-        model_with_structured_output = _get_model(model).with_structured_output(
-            cast(StructuredResponseSchema, structured_response_schema)
-        )
-        response = model_with_structured_output.invoke(messages, config)
-        return {"structured_response": response}
-
-    async def agenerate_structured_response(
-        state: StateSchema, config: RunnableConfig
-    ) -> StateSchema:
-        messages = _get_state_value(state, "messages")
-        structured_response_schema = response_format
-        if isinstance(response_format, tuple):
-            system_prompt, structured_response_schema = response_format
-            messages = [SystemMessage(content=system_prompt)] + list(messages)
-
-        model_with_structured_output = _get_model(model).with_structured_output(
-            cast(StructuredResponseSchema, structured_response_schema)
-        )
-        response = await model_with_structured_output.ainvoke(messages, config)
-        return {"structured_response": response}
-
-    if not tool_calling_enabled:
-        # Define a new graph
-        workflow = StateGraph(state_schema=state_schema, context_schema=context_schema)
-        workflow.add_node(
-            "agent",
-            RunnableCallable(call_model, acall_model),
-            input_schema=input_schema,
-        )
-        if pre_model_hook is not None:
-            workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
-            workflow.add_edge("pre_model_hook", "agent")
-            entrypoint = "pre_model_hook"
-        else:
-            entrypoint = "agent"
-
-        workflow.set_entry_point(entrypoint)
-
-        if post_model_hook is not None:
-            workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
-            workflow.add_edge("agent", "post_model_hook")
-
-        if response_format is not None:
-            workflow.add_node(
-                "generate_structured_response",
-                RunnableCallable(
-                    generate_structured_response,
-                    agenerate_structured_response,
-                ),
-            )
-            if post_model_hook is not None:
-                workflow.add_edge("post_model_hook", "generate_structured_response")
-            else:
-                workflow.add_edge("agent", "generate_structured_response")
-
-        return workflow.compile(
-            checkpointer=checkpointer,
-            store=store,
-            interrupt_before=interrupt_before,
-            interrupt_after=interrupt_after,
-            debug=debug,
-            name=name,
-        )
-
-    # Define the function that determines whether to continue or not
-    def should_continue(state: StateSchema) -> Union[str, list[Send]]:
-        messages = _get_state_value(state, "messages")
-        last_message = messages[-1]
-        # If there is no function call, then we finish
-        if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
-            if post_model_hook is not None:
-                return "post_model_hook"
-            elif response_format is not None:
-                return "generate_structured_response"
-            else:
-                return END
-        # Otherwise if there is, we continue
-        else:
-            if version == "v1":
-                return "tools"
-            elif version == "v2":
-                if post_model_hook is not None:
-                    return "post_model_hook"
-                return [
-                    Send(
-                        "tools",
-                        ToolCallWithContext(
-                            __type="tool_call_with_context",
-                            tool_call=tool_call,
-                            state=state,
-                        ),
-                    )
-                    for tool_call in last_message.tool_calls
-                ]
-
-    # Define a new graph
-    workflow = StateGraph(
-        state_schema=state_schema or AgentState, context_schema=context_schema
+    # Build the graph using the internal builder
+    builder = _AgentBuilder(
+        model=model,
+        tools=tools,
+        prompt=prompt,
+        response_format=response_format,
+        pre_model_hook=pre_model_hook,
+        post_model_hook=post_model_hook,
+        state_schema=state_schema,
+        context_schema=context_schema,
+        version=version,
+        name=name,
     )
 
-    # Define the two nodes we will cycle between
-    workflow.add_node(
-        "agent",
-        RunnableCallable(call_model, acall_model),
-        input_schema=input_schema,
-    )
-    workflow.add_node("tools", tool_node)
+    workflow = builder.build()
 
-    # Optionally add a pre-model hook node that will be called
-    # every time before the "agent" (LLM-calling node)
-    if pre_model_hook is not None:
-        workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
-        workflow.add_edge("pre_model_hook", "agent")
-        entrypoint = "pre_model_hook"
-    else:
-        entrypoint = "agent"
-
-    # Set the entrypoint as `agent`
-    # This means that this node is the first one called
-    workflow.set_entry_point(entrypoint)
-
-    agent_paths = []
-    post_model_hook_paths = [entrypoint, "tools"]
-
-    # Add a post model hook node if post_model_hook is provided
-    if post_model_hook is not None:
-        workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
-        agent_paths.append("post_model_hook")
-        workflow.add_edge("agent", "post_model_hook")
-    else:
-        agent_paths.append("tools")
-
-    # Add a structured output node if response_format is provided
-    if response_format is not None:
-        workflow.add_node(
-            "generate_structured_response",
-            RunnableCallable(
-                generate_structured_response,
-                agenerate_structured_response,
-            ),
-        )
-        if post_model_hook is not None:
-            post_model_hook_paths.append("generate_structured_response")
-        else:
-            agent_paths.append("generate_structured_response")
-    else:
-        if post_model_hook is not None:
-            post_model_hook_paths.append(END)
-        else:
-            agent_paths.append(END)
-
-    if post_model_hook is not None:
-
-        def post_model_hook_router(state: StateSchema) -> Union[str, list[Send]]:
-            """Route to the next node after post_model_hook.
-
-            Routes to one of:
-            * "tools": if there are pending tool calls without a corresponding message.
-            * "generate_structured_response": if no pending tool calls exist and response_format is specified.
-            * END: if no pending tool calls exist and no response_format is specified.
-            """
-
-            messages = _get_state_value(state, "messages")
-            tool_messages = [
-                m.tool_call_id for m in messages if isinstance(m, ToolMessage)
-            ]
-            last_ai_message = next(
-                m for m in reversed(messages) if isinstance(m, AIMessage)
-            )
-            pending_tool_calls = [
-                c for c in last_ai_message.tool_calls if c["id"] not in tool_messages
-            ]
-
-            if pending_tool_calls:
-                return [
-                    Send(
-                        "tools",
-                        ToolCallWithContext(
-                            __type="tool_call_with_context",
-                            tool_call=tool_call,
-                            state=state,
-                        ),
-                    )
-                    for tool_call in pending_tool_calls
-                ]
-            elif isinstance(messages[-1], ToolMessage):
-                return entrypoint
-            elif response_format is not None:
-                return "generate_structured_response"
-            else:
-                return END
-
-        workflow.add_conditional_edges(
-            "post_model_hook",
-            post_model_hook_router,  # type: ignore[arg-type]
-            path_map=post_model_hook_paths,
-        )
-
-    workflow.add_conditional_edges(
-        "agent",
-        should_continue,  # type: ignore[arg-type]
-        path_map=agent_paths,
-    )
-
-    def route_tool_responses(state: StateSchema) -> str:
-        for m in reversed(_get_state_value(state, "messages")):
-            if not isinstance(m, ToolMessage):
-                break
-            if m.name in should_return_direct:
-                return END
-
-        # handle a case of parallel tool calls where
-        # the tool w/ `return_direct` was executed in a different `Send`
-        if isinstance(m, AIMessage) and m.tool_calls:
-            if any(call["name"] in should_return_direct for call in m.tool_calls):
-                return END
-
-        return entrypoint
-
-    if should_return_direct:
-        workflow.add_conditional_edges(
-            "tools", route_tool_responses, path_map=[entrypoint, END]
-        )
-    else:
-        workflow.add_edge("tools", entrypoint)
-
-    # Finally, we compile it!
-    # This compiles it into a LangChain Runnable,
-    # meaning you can use it as you would any other runnable
+    # Compile and return the graph
     return workflow.compile(
         checkpointer=checkpointer,
         store=store,

libs/prebuilt/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-prebuilt"
-version = "0.5.2"
+version = "0.6.0"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.9"

libs/prebuilt/uv.lock

@@ -460,7 +460,7 @@ dev = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "0.5.2"
+version = "0.6.0"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },