diff --git a/libs/langgraph/langgraph/graph/message.py b/libs/langgraph/langgraph/graph/message.py
index 3018d21da..64f375ad2 100644
--- a/libs/langgraph/langgraph/graph/message.py
+++ b/libs/langgraph/langgraph/graph/message.py
@@ -250,14 +250,16 @@ def _messages_delta_reducer(
     """**Experimental.** Batch reducer for use with `DeltaChannel`.
 
     Processes all writes in one pass — dedup by ID, `RemoveMessage`
-    tombstoning — without calling `add_messages`. Assumes writes contain
-    already-typed `BaseMessage` objects (no raw-dict coercion).
+    tombstoning — without calling `add_messages`.
 
     This reducer is batching-invariant, as required by `DeltaChannel`:
     `reducer(reducer(state, xs), ys) == reducer(state, xs + ys)`.
 
-    Use `add_messages` as the reducer for `BinaryOperatorAggregate` or
-    anywhere raw message dicts / strings need to be coerced first.
+    Raw dict / string / tuple inputs are coerced to typed `BaseMessage`
+    objects so that HTTP-driven graphs work without a separate coercion
+    step. This is not full `add_messages` parity — `REMOVE_ALL_MESSAGES`,
+    unknown-id `RemoveMessage` errors, missing-id UUID assignment, and
+    `BaseMessageChunk` conversion are not handled here.
 
     Example::
 
@@ -268,13 +270,30 @@ def _messages_delta_reducer(
         class State(TypedDict):
             messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
     """
-    from itertools import chain
 
-    index: dict[str, int] = {m.id: i for i, m in enumerate(state) if m.id is not None}
-    result: list[AnyMessage | None] = list(state)
-    for msg in chain.from_iterable(
-        [w] if isinstance(w, BaseMessage) else w for w in writes
-    ):
+    # Each write is either a list of message-likes or a single message-like
+    # (BaseMessage / dict / str / tuple). Only lists flatten; everything
+    # else is one message.
+    flat: list[Any] = []
+    for w in writes:
+        if isinstance(w, list):
+            flat.extend(w)
+        else:
+            flat.append(w)
+    # Steady state: the reducer's own output is already typed, so skip
+    # `convert_to_messages` on state when the first element is a BaseMessage.
+    # Only raw input (initial dicts, deserialized blobs) hits the slow path.
+    if state and isinstance(state[0], BaseMessage):
+        state_msgs = state
+    else:
+        state_msgs = cast("list[AnyMessage]", convert_to_messages(state))
+    msgs = cast("list[AnyMessage]", convert_to_messages(flat))
+
+    index: dict[str, int] = {
+        m.id: i for i, m in enumerate(state_msgs) if m.id is not None
+    }
+    result: list[AnyMessage | None] = list(state_msgs)
+    for msg in msgs:
         mid = msg.id
         if mid is None:
             result.append(msg)
diff --git a/libs/langgraph/tests/test_channels.py b/libs/langgraph/tests/test_channels.py
index fd2e35e33..6c15f3307 100644
--- a/libs/langgraph/tests/test_channels.py
+++ b/libs/langgraph/tests/test_channels.py
@@ -233,6 +233,58 @@ def test_delta_channel_update_by_id_and_replay() -> None:
     assert ch2.get()[0].content == "updated"
 
 
+def test_delta_channel_dict_coercion() -> None:
+    """_messages_delta_reducer coerces dict writes to BaseMessage objects.
+
+    HTTP-driven input always arrives as JSON dicts.  The reducer must coerce
+    them (same contract as add_messages) so graphs work without a separate
+    coercion step.
+    """
+    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+
+    # dict input — simulates what arrives from the HTTP API
+    ch.update([{"role": "human", "content": "hello", "id": "h1"}])
+    assert len(ch.get()) == 1
+    assert isinstance(ch.get()[0], HumanMessage)
+    assert ch.get()[0].content == "hello"
+    assert ch.get()[0].id == "h1"
+
+    # update by ID via dict
+    ch.update([{"role": "ai", "content": "world", "id": "h1"}])
+    assert len(ch.get()) == 1
+    assert ch.get()[0].content == "world"
+
+    # remove via RemoveMessage instance (same contract as add_messages)
+    ch.update([RemoveMessage(id="h1")])
+    assert ch.get() == []
+
+
+def test_messages_delta_reducer_coerces_state() -> None:
+    """State (left side) is coerced when raw — supports raw initial input
+    and deserialized blobs. The steady-state path (state already typed)
+    short-circuits and skips coercion.
+    """
+    state = [{"role": "human", "content": "hello", "id": "h1"}]
+    writes = [[{"role": "ai", "content": "world", "id": "h1"}]]
+    result = _messages_delta_reducer(state, writes)  # type: ignore[arg-type]
+    assert len(result) == 1
+    assert isinstance(result[0], AIMessage)
+    assert result[0].content == "world"
+    assert result[0].id == "h1"
+
+
+def test_messages_delta_reducer_tuple_write_is_one_message() -> None:
+    """A top-level tuple write is one message-like, not a sequence to flatten.
+
+    `("user", "hi")` is a valid `MessageLikeRepresentation`; flattening it
+    would produce two HumanMessages ("user", "hi") instead of one.
+    """
+    result = _messages_delta_reducer([], [("user", "hi")])  # type: ignore[arg-type]
+    assert len(result) == 1
+    assert isinstance(result[0], HumanMessage)
+    assert result[0].content == "hi"
+
+
 def test_delta_channel_checkpoint_returns_sentinel() -> None:
     """checkpoint() always returns DELTA_SENTINEL regardless of state."""
     ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)