langchain-ai/langchain · error · NotImplementedError

{self!r} only supports async methods.

Error message

{self!r} only supports async methods.

What it means

`RunnableGenerator.transform` (the sync streaming path) requires a sync `_transform`, which is only set when the constructor received a sync generator function. If the generator was created with an async transform, only `_atransform` exists, so calling `.stream()`/`.transform()`/`.invoke()` raises `NotImplementedError`. Async-only generators cannot be driven from synchronous code.

Source

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

            return False
        return False

    __hash__ = None  # type: ignore[assignment]

    @override
    def __repr__(self) -> str:
        return f"RunnableGenerator({self.name})"

    @override
    def transform(
        self,
        input: Iterator[Input],
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> Iterator[Output]:
        if not hasattr(self, "_transform"):
            msg = f"{self!r} only supports async methods."
            raise NotImplementedError(msg)
        return self._transform_stream_with_config(
            input,
            self._transform,  # type: ignore[arg-type]
            config,
            defers_inputs=True,
            **kwargs,
        )

    @override
    def stream(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> Iterator[Output]:
        return self.transform(iter([input]), config, **kwargs)

    @override

View on GitHub (pinned to e32fa9a52e)

Solutions

1. Use the async API: `await runnable.ainvoke(...)`, `async for chunk in runnable.astream(...)`.
2. Provide both a sync generator and an async generator if you need both paths (`RunnableGenerator` only stores one, so expose two separate instances or a sync one).
3. In sync-only code, define the transform as a plain generator function.
4. In shared libraries, branch on `asyncio.get_event_loop().is_running()` or expose explicit sync/async constructors.

Example fix

// before
async def atransform(chunks):
    async for c in chunks:
        yield c * 2
rg = RunnableGenerator(atransform)
list(rg.stream([1, 2]))  # NotImplementedError

// after
def transform(chunks):
    for c in chunks:
        yield c * 2
rg = RunnableGenerator(transform)
list(rg.stream([1, 2]))  # ok
// or keep async and call: async for chunk in rg.astream([1, 2]): ...

import inspect
from langchain_core.runnables import RunnableGenerator

def supports_sync_stream(rg: RunnableGenerator) -> bool:
    return inspect.isgeneratorfunction(getattr(rg, '_transform', None)) if hasattr(rg, '_transform') else False

def is_sync_generator_runnable(rg) -> bool:
    return hasattr(rg, '_transform')

try:
    chunks = list(rg.stream(x))
except NotImplementedError as e:
    if 'only supports async' in str(e):
        chunks = asyncio.run(collect(rg.astream(x)))
    else:
        raise

When it happens

Trigger: `RunnableGenerator(async_transform)` where `async_transform` is an `async def` with `yield`, then calling `.stream(input)`, `.transform(iter([input]))`, or `.invoke(input)` (invoke consumes the sync stream).

Common situations: Writing an async streaming transformer for an async app and later reusing it in a script/notebook sync path; test suites that call `.invoke()` on runnables that were built async-only.

Related errors

• {self!r} only supports sync methods.
• Expected a generator function type for `transform`.Instead g
• Cannot stream a coroutine function synchronously.Use `astrea
• Cannot stream from a generator function asynchronously.Use .
• Func was provided as a coroutine function, but afunc was als

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