langchain-ai/langchain · error · NotImplementedError

stream_events(version={version!r}) is not supported. Use ast

Error message

stream_events(version={version!r}) is not supported. Use astream_events() for v1/v2, or stream_events(version='v3') on a supported subclass.

What it means

Raised by the base `Runnable.stream_events` when `version` is 'v1' or 'v2' (or anything unrecognized besides the v3 branch). The synchronous `stream_events` API only implements v3; v1/v2 event streaming exists solely in the async `astream_events` path. The message states this directly: use `astream_events()` for v1/v2, or `stream_events(version='v3')` on a supported subclass.

Source

Thrown at libs/core/langchain_core/runnables/base.py:1746

        """
        # Base impl always raises; consume args so they don't trip ARG002.
        del input, config, include_names, include_types, include_tags
        del exclude_names, exclude_types, exclude_tags, kwargs
        if version == "v3":
            msg = (
                "stream_events(version='v3') is only supported on Runnable subclasses "
                "that implement the v3 streaming protocol "
                "(BaseChatModel, CompiledGraph). "
                f"Got: {type(self).__name__}"
            )
            raise NotImplementedError(msg)
        msg = (
            f"stream_events(version={version!r}) is not supported. "
            "Use astream_events() for v1/v2, or stream_events(version='v3') "
            "on a supported subclass."
        )
        raise NotImplementedError(msg)

    def transform(
        self,
        input: Iterator[Input],
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> Iterator[Output]:
        """Transform inputs to outputs.

        Default implementation of transform, which buffers input and calls `astream`.

        Subclasses must override this method if they can start producing output while
        input is still being generated.

        Args:
            input: An iterator of inputs to the `Runnable`.
            config: The config to use for the `Runnable`.
            **kwargs: Additional keyword arguments to pass to the `Runnable`.

View on GitHub (pinned to e32fa9a52e)

Solutions

  1. Switch to the async API: `async for ev in chain.astream_events(inp, version="v2")`
  2. If code must stay synchronous, use `stream_events(inp, version="v3")` on a BaseChatModel/CompiledGraph (a non-v3 subclass will raise the companion v3 error)
  3. If no event granularity is needed, use plain `chain.stream(inp)` / `chain.invoke(inp)` instead

Example fix

# before
for ev in chain.stream_events(inp, version="v2"):
    ...  # NotImplementedError: not supported

# after
async def run():
    async for ev in chain.astream_events(inp, version="v2"):
        ...
Defensive patterns

Strategy: validation

Validate before calling

def checked_sync_version(v: str) -> Literal["v3"]:
    if v != "v3":
        msg = "sync stream_events supports only version='v3'; use astream_events for v1/v2"
        raise ValueError(msg)
    return "v3"

Type guard

def is_sync_stream_version(v: object) -> bool:
    return v == "v3"

Try / catch

try:
    for ev in chain.stream_events(inp, version=version):
        ...
except NotImplementedError as e:
    if "astream_events" in str(e):
        # route to the async API instead
        raise RuntimeError("use astream_events(version='v1'|'v2')") from e
    raise

Prevention

When it happens

Trigger: `chain.stream_events(inp, version="v2")` — mirroring the familiar `astream_events(version="v2")` default — or `version="v1"`. Any non-'v3' value reaches the second raise in the base implementation.

Common situations: Copy-pasting async streaming code and only dropping the leading `a`; assuming parity between sync and async event APIs; defaults carried over from `astream_events` call sites where v2 is standard.

Related errors

AI-assisted analysis of langchain-ai/langchain@e32fa9a52e (2026-08-14). Data as JSON: /api/errors/a66e34e5848a429f. Report an issue: GitHub.