feat(anthropic): native content-block streaming events

libs/partners/anthropic/langchain_anthropic/_stream_events.py

@@ -0,0 +1,147 @@
+"""Native content-block streaming-event converter for Anthropic.
+
+Maps the raw Anthropic SDK stream (``message_start`` /
+``content_block_start`` / ``content_block_delta`` / ``content_block_stop`` /
+``message_delta`` / ``message_stop``) to the protocol ``MessagesData``
+event lifecycle, reusing the shared ``BlockStreamTracker`` for block
+mechanics and the existing per-event chunk builder for content extraction.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any
+
+from langchain_core.language_models.stream_events import (
+    BlockStreamTracker,
+    accumulate_usage,
+    build_message_finish,
+    iter_protocol_blocks,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator, Iterator
+
+    from langchain_core.messages import AIMessageChunk
+    from langchain_protocol.protocol import (
+        MessageMetadata,
+        MessagesData,
+        MessageStartData,
+    )
+
+# The bound method ``ChatAnthropic._make_message_chunk_from_anthropic_event``.
+MakeChunk = Callable[..., "tuple[AIMessageChunk | None, Any]"]
+
+
+def _message_start(event: Any, message_id: str | None) -> MessageStartData:
+    msg_obj = getattr(event, "message", None)
+    # Do not use the provider message id (`msg_...`) here: on the v3 path core
+    # seeds the stream with the LangChain run id, and an empty id lets that
+    # stand (matching the compat bridge). Only an explicit `message_id` wins.
+    resolved_id = message_id or ""
+    metadata: MessageMetadata = {"provider": "anthropic"}
+    model = getattr(msg_obj, "model", None)
+    if model:
+        metadata["model"] = model
+    return {
+        "event": "message-start",
+        "role": "ai",
+        "id": resolved_id,
+        "metadata": metadata,
+    }
+
+
+def convert_anthropic_stream(
+    raw: Iterator[Any],
+    make_chunk: MakeChunk,
+    *,
+    stream_usage: bool = True,
+    message_id: str | None = None,
+) -> Iterator[MessagesData]:
+    """Convert a raw Anthropic event stream to protocol events.
+
+    Args:
+        raw: Raw Anthropic SDK stream events.
+        make_chunk: ``ChatAnthropic._make_message_chunk_from_anthropic_event``,
+            injected so the converter stays pure and unit-testable.
+        stream_usage: Forwarded to ``make_chunk``.
+        message_id: Overrides the provider message id on ``message-start``.
+
+    Yields:
+        Protocol ``MessagesData`` lifecycle events.
+    """
+    tracker = BlockStreamTracker()
+    started = False
+    block_start_event: Any = None
+    usage: dict[str, Any] | None = None
+    response_metadata: dict[str, Any] = {"model_provider": "anthropic"}
+
+    for event in raw:
+        etype = getattr(event, "type", None)
+        chunk_msg, block_start_event = make_chunk(
+            event,
+            stream_usage=stream_usage,
+            coerce_content_to_string=False,
+            block_start_event=block_start_event,
+        )
+        if not started:
+            started = True
+            yield _message_start(event, message_id)
+        if chunk_msg is not None:
+            for key, block in iter_protocol_blocks(chunk_msg):
+                yield from tracker.feed(key, block)
+            if chunk_msg.usage_metadata:
+                usage = accumulate_usage(usage, chunk_msg.usage_metadata)
+            if chunk_msg.response_metadata:
+                response_metadata.update(chunk_msg.response_metadata)
+        if etype == "content_block_stop":
+            yield from tracker.finish_block(getattr(event, "index", None))
+
+    if not started:
+        return
+    yield from tracker.finish_all()
+    yield build_message_finish(usage=usage, response_metadata=response_metadata)
+
+
+async def aconvert_anthropic_stream(
+    raw: AsyncIterator[Any],
+    make_chunk: MakeChunk,
+    *,
+    stream_usage: bool = True,
+    message_id: str | None = None,
+) -> AsyncIterator[MessagesData]:
+    """Async twin of `convert_anthropic_stream`. `make_chunk` is sync."""
+    tracker = BlockStreamTracker()
+    started = False
+    block_start_event: Any = None
+    usage: dict[str, Any] | None = None
+    response_metadata: dict[str, Any] = {"model_provider": "anthropic"}
+
+    async for event in raw:
+        etype = getattr(event, "type", None)
+        chunk_msg, block_start_event = make_chunk(
+            event,
+            stream_usage=stream_usage,
+            coerce_content_to_string=False,
+            block_start_event=block_start_event,
+        )
+        if not started:
+            started = True
+            yield _message_start(event, message_id)
+        if chunk_msg is not None:
+            for key, block in iter_protocol_blocks(chunk_msg):
+                for ev in tracker.feed(key, block):
+                    yield ev
+            if chunk_msg.usage_metadata:
+                usage = accumulate_usage(usage, chunk_msg.usage_metadata)
+            if chunk_msg.response_metadata:
+                response_metadata.update(chunk_msg.response_metadata)
+        if etype == "content_block_stop":
+            for ev in tracker.finish_block(getattr(event, "index", None)):
+                yield ev
+
+    if not started:
+        return
+    for ev in tracker.finish_all():
+        yield ev
+    yield build_message_finish(usage=usage, response_metadata=response_metadata)

libs/partners/anthropic/langchain_anthropic/chat_models.py

@@ -11,7 +11,7 @@ import warnings
 from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence
 from functools import cached_property
 from operator import itemgetter
-from typing import Any, Final, Literal, cast
+from typing import TYPE_CHECKING, Any, Final, Literal, cast
 
 import anthropic
 from langchain_core.callbacks import (
@@ -64,9 +64,16 @@ from langchain_anthropic._client_utils import (
     _get_default_httpx_client,
 )
 from langchain_anthropic._compat import _convert_from_v1_to_anthropic
+from langchain_anthropic._stream_events import (
+    aconvert_anthropic_stream,
+    convert_anthropic_stream,
+)
 from langchain_anthropic.data._profiles import _PROFILES
 from langchain_anthropic.output_parsers import extract_tool_calls
 
+if TYPE_CHECKING:
+    from langchain_protocol.protocol import ContentBlockDelta, MessagesData
+
 _message_type_lookups = {
     "human": "user",
     "ai": "assistant",
@@ -874,6 +881,13 @@ def _handle_anthropic_bad_request(e: anthropic.BadRequestError) -> None:
     raise
 
 
+def _delta_text_for_callback(delta: ContentBlockDelta) -> str:
+    """Text increment to report via `on_llm_new_token` (empty for non-text)."""
+    if delta.get("type") == "text-delta":
+        return cast("str", delta.get("text", ""))
+    return ""
+
+
 class ChatAnthropic(BaseChatModel):
     """Anthropic (Claude) chat models.
 
@@ -1509,6 +1523,74 @@ class ChatAnthropic(BaseChatModel):
         except anthropic.BadRequestError as e:
             _handle_anthropic_bad_request(e)
 
+    def _stream_chat_model_events(
+        self,
+        messages: list[BaseMessage],
+        stop: list[str] | None = None,
+        run_manager: CallbackManagerForLLMRun | None = None,
+        *,
+        stream_usage: bool | None = None,
+        message_id: str | None = None,
+        **kwargs: Any,
+    ) -> Iterator[MessagesData]:
+        """Emit Anthropic-native content-block protocol events.
+
+        Detected by `langchain-core`'s `_iter_v2_events`; powers
+        `stream_events(version="v3")`. Falls through to the compat bridge
+        only if this method is absent. `message_id` is threaded from the
+        stream so `message-start` matches the bridge's LangChain run id.
+        """
+        if stream_usage is None:
+            stream_usage = self.stream_usage
+        kwargs["stream"] = True
+        payload = self._get_request_payload(messages, stop=stop, **kwargs)
+        try:
+            raw = self._create(payload)
+            for event in convert_anthropic_stream(
+                raw,
+                self._make_message_chunk_from_anthropic_event,
+                stream_usage=stream_usage,
+                message_id=message_id,
+            ):
+                if run_manager is not None and event["event"] == "content-block-delta":
+                    token = _delta_text_for_callback(event["delta"])
+                    if token:
+                        run_manager.on_llm_new_token(token)
+                yield event
+        except anthropic.BadRequestError as e:
+            _handle_anthropic_bad_request(e)
+
+    async def _astream_chat_model_events(
+        self,
+        messages: list[BaseMessage],
+        stop: list[str] | None = None,
+        run_manager: AsyncCallbackManagerForLLMRun | None = None,
+        *,
+        stream_usage: bool | None = None,
+        message_id: str | None = None,
+        **kwargs: Any,
+    ) -> AsyncIterator[MessagesData]:
+        """Async twin of `_stream_chat_model_events`."""
+        if stream_usage is None:
+            stream_usage = self.stream_usage
+        kwargs["stream"] = True
+        payload = self._get_request_payload(messages, stop=stop, **kwargs)
+        try:
+            raw = await self._acreate(payload)
+            async for event in aconvert_anthropic_stream(
+                raw,
+                self._make_message_chunk_from_anthropic_event,
+                stream_usage=stream_usage,
+                message_id=message_id,
+            ):
+                if run_manager is not None and event["event"] == "content-block-delta":
+                    token = _delta_text_for_callback(event["delta"])
+                    if token:
+                        await run_manager.on_llm_new_token(token)
+                yield event
+        except anthropic.BadRequestError as e:
+            _handle_anthropic_bad_request(e)
+
     def _make_message_chunk_from_anthropic_event(
         self,
         event: anthropic.types.RawMessageStreamEvent,

libs/partners/anthropic/tests/unit_tests/test_chat_models.py

@@ -3222,19 +3222,12 @@ def test_no_task_budget_no_beta() -> None:
         assert "task-budgets-2026-03-13" not in betas
 
 
-def test_anthropic_stream_events_v3_lifecycle() -> None:
-    """Validate lifecycle events across a thinking + text + tool_use stream.
+def _lifecycle_events() -> list[Any]:
+    """Build a raw thinking + text + tool_use Anthropic stream fixture.
 
-    Anthropic emits raw `content_block_start` / `content_block_delta` /
-    `content_block_stop` events with integer `index` fields, interleaved
-    with `message_start` and `message_delta`. This test threads a
-    realistic event sequence through `_stream` via a mocked raw client
-    and asserts that `stream_events(version="v3")` produces a spec-conformant
-    event stream: paired start/finish per block, no interleaving, sequential
-    `uint` wire indices.
+    Shared by the sync and async v3 lifecycle parity tests so both exercise
+    the identical event sequence through the native hooks.
     """
-    from unittest.mock import patch
-
     from anthropic.types import (
         InputJSONDelta,
         RawContentBlockDeltaEvent,
@@ -3252,7 +3245,6 @@ def test_anthropic_stream_events_v3_lifecycle() -> None:
     from anthropic.types.raw_message_delta_event import (
         MessageDeltaUsage as RawMessageDeltaUsage,
     )
-    from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream
 
     msg = Message(
         id="msg_1",
@@ -3265,7 +3257,7 @@ def test_anthropic_stream_events_v3_lifecycle() -> None:
         type="message",
     )
 
-    events = [
+    return [
         RawMessageStartEvent(message=msg, type="message_start"),
         # thinking block (index=0)
         RawContentBlockStartEvent(
@@ -3337,6 +3329,24 @@ def test_anthropic_stream_events_v3_lifecycle() -> None:
         RawMessageStopEvent(type="message_stop"),
     ]
 
+
+def test_anthropic_stream_events_v3_lifecycle() -> None:
+    """Validate lifecycle events across a thinking + text + tool_use stream.
+
+    Anthropic emits raw `content_block_start` / `content_block_delta` /
+    `content_block_stop` events with integer `index` fields, interleaved
+    with `message_start` and `message_delta`. This test threads a
+    realistic event sequence through `_stream` via a mocked raw client
+    and asserts that `stream_events(version="v3")` produces a spec-conformant
+    event stream: paired start/finish per block, no interleaving, sequential
+    `uint` wire indices.
+    """
+    from unittest.mock import patch
+
+    from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream
+
+    events = _lifecycle_events()
+
     # Enable thinking so `coerce_content_to_string=False` in `_stream`,
     # which gives every content block an integer `index` field — the
     # structured path the protocol bridge actually exercises.  Default
@@ -3355,6 +3365,14 @@ def test_anthropic_stream_events_v3_lifecycle() -> None:
 
     assert_valid_event_stream(stream_events)
 
+    # `message-start` must carry the stream's LangChain run id (threaded from
+    # core), matching the bridge path — not the provider message id ("msg_1")
+    # and not an empty string.
+    message_start = cast("dict[str, Any]", stream_events[0])
+    assert message_start["event"] == "message-start"
+    assert message_start["id"]
+    assert message_start["id"] != "msg_1"
+
     finishes = [e for e in stream_events if e["event"] == "content-block-finish"]
     types = [f["content"]["type"] for f in finishes]
     assert types == ["reasoning", "text", "tool_call"]
@@ -3378,3 +3396,36 @@ def test_anthropic_stream_events_v3_lifecycle() -> None:
     message_finish = cast("dict[str, Any]", stream_events[-1])
     assert message_finish["event"] == "message-finish"
     assert message_finish["metadata"]["stop_reason"] == "tool_use"
+
+
+async def test_anthropic_astream_events_v3_lifecycle() -> None:
+    """Async twin of `test_anthropic_stream_events_v3_lifecycle`.
+
+    Exercises `_astream_chat_model_events` end-to-end via the
+    `AsyncChatModelStream` producer task.
+    """
+    from unittest.mock import patch
+
+    events = _lifecycle_events()
+    llm = ChatAnthropic(
+        model=MODEL_NAME, thinking={"type": "enabled", "budget_tokens": 1024}
+    )
+
+    async def mock_acreate(_payload: Any) -> Any:
+        async def _gen() -> Any:
+            for event in events:
+                yield event
+
+        return _gen()
+
+    with patch.object(llm, "_acreate", mock_acreate):
+        stream = await llm.astream_events("Test query", version="v3")
+        collected = [e async for e in stream]
+
+    finishes = [e for e in collected if e["event"] == "content-block-finish"]
+    assert [f["content"]["type"] for f in finishes] == [
+        "reasoning",
+        "text",
+        "tool_call",
+    ]
+    assert collected[-1]["metadata"]["stop_reason"] == "tool_use"

libs/partners/anthropic/tests/unit_tests/test_stream_events.py

@@ -0,0 +1,182 @@
+"""Unit tests for the Anthropic native stream-events converter."""
+
+from typing import Any, cast
+
+from anthropic.types import (
+    InputJSONDelta,
+    Message,
+    RawContentBlockDeltaEvent,
+    RawContentBlockStartEvent,
+    RawContentBlockStopEvent,
+    RawMessageDeltaEvent,
+    RawMessageStartEvent,
+    RawMessageStopEvent,
+    TextBlock,
+    TextDelta,
+    ThinkingBlock,
+    ThinkingDelta,
+    ToolUseBlock,
+    Usage,
+)
+from anthropic.types.raw_message_delta_event import Delta as RawMessageDelta
+from anthropic.types.raw_message_delta_event import (
+    MessageDeltaUsage as RawMessageDeltaUsage,
+)
+from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream
+
+from langchain_anthropic import ChatAnthropic
+from langchain_anthropic._stream_events import convert_anthropic_stream
+
+MODEL_NAME = "claude-haiku-4-5-20251001"
+
+
+def _events() -> list[Any]:
+    msg = Message(
+        id="msg_1",
+        content=[],
+        model=MODEL_NAME,
+        role="assistant",
+        stop_reason=None,
+        stop_sequence=None,
+        usage=Usage(input_tokens=10, output_tokens=0),
+        type="message",
+    )
+    return [
+        RawMessageStartEvent(message=msg, type="message_start"),
+        RawContentBlockStartEvent(
+            content_block=ThinkingBlock(signature="", thinking="", type="thinking"),
+            index=0,
+            type="content_block_start",
+        ),
+        RawContentBlockDeltaEvent(
+            delta=ThinkingDelta(thinking="Let me ", type="thinking_delta"),
+            index=0,
+            type="content_block_delta",
+        ),
+        RawContentBlockDeltaEvent(
+            delta=ThinkingDelta(thinking="think.", type="thinking_delta"),
+            index=0,
+            type="content_block_delta",
+        ),
+        RawContentBlockStopEvent(index=0, type="content_block_stop"),
+        RawContentBlockStartEvent(
+            content_block=TextBlock(text="", type="text"),
+            index=1,
+            type="content_block_start",
+        ),
+        RawContentBlockDeltaEvent(
+            delta=TextDelta(text="The answer ", type="text_delta"),
+            index=1,
+            type="content_block_delta",
+        ),
+        RawContentBlockDeltaEvent(
+            delta=TextDelta(text="is 42.", type="text_delta"),
+            index=1,
+            type="content_block_delta",
+        ),
+        RawContentBlockStopEvent(index=1, type="content_block_stop"),
+        RawContentBlockStartEvent(
+            content_block=ToolUseBlock(
+                id="toolu_1", input={}, name="search", type="tool_use"
+            ),
+            index=2,
+            type="content_block_start",
+        ),
+        RawContentBlockDeltaEvent(
+            delta=InputJSONDelta(partial_json='{"q":', type="input_json_delta"),
+            index=2,
+            type="content_block_delta",
+        ),
+        RawContentBlockDeltaEvent(
+            delta=InputJSONDelta(partial_json=' "weather"}', type="input_json_delta"),
+            index=2,
+            type="content_block_delta",
+        ),
+        RawContentBlockStopEvent(index=2, type="content_block_stop"),
+        RawMessageDeltaEvent(
+            delta=RawMessageDelta(stop_reason="tool_use", stop_sequence=None),
+            type="message_delta",
+            usage=RawMessageDeltaUsage(
+                output_tokens=50,
+                input_tokens=10,
+                cache_read_input_tokens=0,
+                cache_creation_input_tokens=0,
+            ),
+        ),
+        RawMessageStopEvent(type="message_stop"),
+    ]
+
+
+def test_convert_anthropic_stream_lifecycle() -> None:
+    llm = ChatAnthropic(model=MODEL_NAME)
+    events: list[Any] = list(
+        convert_anthropic_stream(
+            iter(_events()), llm._make_message_chunk_from_anthropic_event
+        )
+    )
+
+    assert_valid_event_stream(events)
+
+    assert events[0]["event"] == "message-start"
+    # The provider message id (`msg_1`) is deliberately NOT used: on the v3 path
+    # core seeds the stream with the LangChain run id, and an empty id here lets
+    # that stand (matching the compat bridge). Only an explicit `message_id`
+    # overrides it.
+    assert events[0]["id"] == ""
+    assert events[0]["metadata"]["provider"] == "anthropic"
+    assert events[0]["metadata"]["model"] == MODEL_NAME
+
+    finishes = [e for e in events if e["event"] == "content-block-finish"]
+    assert [f["content"]["type"] for f in finishes] == [
+        "reasoning",
+        "text",
+        "tool_call",
+    ]
+    assert [f["index"] for f in finishes] == [0, 1, 2]
+    reasoning = cast("dict[str, Any]", finishes[0]["content"])
+    text = cast("dict[str, Any]", finishes[1]["content"])
+    assert reasoning["reasoning"] == "Let me think."
+    assert text["text"] == "The answer is 42."
+    tool = cast("dict[str, Any]", finishes[2]["content"])
+    assert tool["args"] == {"q": "weather"}
+    assert tool["name"] == "search"
+
+    message_finish = events[-1]
+    assert message_finish["event"] == "message-finish"
+    assert message_finish["metadata"]["stop_reason"] == "tool_use"
+    # content-block-finish for index 0 arrives before index 1 starts
+    # (true stop boundaries, not all-at-end).
+    first_finish = next(
+        i for i, e in enumerate(events) if e["event"] == "content-block-finish"
+    )
+    first_idx1_start = next(
+        i
+        for i, e in enumerate(events)
+        if e["event"] == "content-block-start" and e["index"] == 1
+    )
+    assert first_finish < first_idx1_start
+
+
+async def test_aconvert_anthropic_stream_lifecycle() -> None:
+    llm = ChatAnthropic(model=MODEL_NAME)
+
+    async def _araw() -> Any:
+        for event in _events():
+            yield event
+
+    from langchain_anthropic._stream_events import aconvert_anthropic_stream
+
+    events: list[Any] = [
+        e
+        async for e in aconvert_anthropic_stream(
+            _araw(), llm._make_message_chunk_from_anthropic_event
+        )
+    ]
+    assert_valid_event_stream(events)
+    finishes = [e for e in events if e["event"] == "content-block-finish"]
+    assert [f["content"]["type"] for f in finishes] == [
+        "reasoning",
+        "text",
+        "tool_call",
+    ]
+    assert events[-1]["metadata"]["stop_reason"] == "tool_use"