langchain-ai/langchain · error · NotImplementedError

{self!r} only supports sync methods.

Error message

{self!r} only supports sync methods.

What it means

`RunnableGenerator.atransform` requires an async `_atransform`, set only when the constructor received an async generator function. If the generator wraps a sync generator, the async path is absent and `.astream()`/`.atransform()` raises `NotImplementedError`. Sync generator code cannot be safely driven inside an event loop's async iteration protocol by this class.

Source

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

    @override
    def invoke(
        self, input: Input, config: RunnableConfig | None = None, **kwargs: Any
    ) -> Output:
        final: Output | None = None
        for output in self.stream(input, config, **kwargs):
            final = output if final is None else final + output  # type: ignore[operator]
        return cast("Output", final)

    @override
    def atransform(
        self,
        input: AsyncIterator[Input],
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Output]:
        if not hasattr(self, "_atransform"):
            msg = f"{self!r} only supports sync methods."
            raise NotImplementedError(msg)

        return self._atransform_stream_with_config(
            input, self._atransform, config, defers_inputs=True, **kwargs
        )

    @override
    def astream(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Output]:
        async def input_aiter() -> AsyncIterator[Input]:
            yield input

        return self.atransform(input_aiter(), config, **kwargs)

    @override

View on GitHub (pinned to e32fa9a52e)

Solutions

1. Call the sync API from a worker: `await run_in_executor(None, lambda: list(rg.stream(x)))`.
2. Rewrite the transform as an async generator (`async def` + `yield`, `async for` over input).
3. Run sync streaming in a thread and bridge chunks via an `asyncio.Queue` if you need true async streaming.
4. Standardize team code on async generators for anything used in async servers.

Example fix

// before
def transform(chunks):
    for c in chunks:
        yield c * 2
rg = RunnableGenerator(transform)
async for c in rg.astream([1]):  # NotImplementedError
    ...

// after
async def atransform(chunks):
    async for c in chunks:
        yield c * 2
rg = RunnableGenerator(atransform)
async for c in rg.astream([1]):
    ...

def supports_async_stream(rg) -> bool:
    return hasattr(rg, '_atransform')

def is_async_generator_runnable(rg) -> bool:
    return hasattr(rg, '_atransform')

try:
    async for chunk in rg.astream(x):
        ...
except NotImplementedError as e:
    if 'only supports sync' in str(e):
        chunks = await asyncio.to_thread(lambda: list(rg.stream(x)))
    else:
        raise

When it happens

Trigger: `RunnableGenerator(sync_generator)` then `async for chunk in rg.astream(x)` or `await rg.atransform(...)`; also `await rg.ainvoke(x)` since it consumes the async stream.

Common situations: Reusing a sync streaming transformer inside an async FastAPI/LangServe endpoint; upgrading a sync pipeline to `asyncio` without rewriting the generator.

Related errors

• {self!r} only supports async methods.
• Cannot stream from a generator function asynchronously.Use .
• Expected a generator function type for `transform`.Instead g
• Cannot invoke a coroutine function synchronously.Use `ainvok
• Cannot stream a coroutine function synchronously.Use `astrea

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