langchain-ai/langchain · error · TypeError

Cannot stream a coroutine function synchronously.Use `astrea

Error message

Cannot stream a coroutine function synchronously.Use `astream` instead.

What it means

`RunnableLambda.transform`/`stream` delegate to `_transform_stream_with_config` only when a sync implementation (`self.func`) exists. For an async-only `RunnableLambda`, sync streaming is impossible, so a `TypeError` is raised telling you to use `astream`. This is the streaming counterpart of the invoke-time sync/async guard.

Source

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

    def transform(
        self,
        input: Iterator[Input],
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> Iterator[Output]:
        if hasattr(self, "func"):
            yield from self._transform_stream_with_config(
                input,
                self._transform,
                ensure_config(config),
                **kwargs,
            )
        else:
            msg = (
                "Cannot stream a coroutine function synchronously."
                "Use `astream` instead."
            )
            raise TypeError(msg)

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

    async def _atransform(
        self,
        chunks: AsyncIterator[Input],
        run_manager: AsyncCallbackManagerForChainRun,
        config: RunnableConfig,
        **kwargs: Any,
    ) -> AsyncIterator[Output]:
        final: Input

View on GitHub (pinned to e32fa9a52e)

Solutions

1. Switch to async streaming: `async for chunk in r.astream(x): ...`.
2. Construct with both paths: `RunnableLambda(sync_fn, afunc=async_fn)`.
3. Bridge sync: `asyncio.run(collect_async(r.astream(x)))` outside a loop.
4. Add sync wrappers for any lambda used by sync consumers.

Example fix

# before
for chunk in RunnableLambda(async_fn).stream(x):  # TypeError
    print(chunk)

# after
async def main():
    async for chunk in RunnableLambda(async_fn).astream(x):
        print(chunk)
asyncio.run(main())

def supports_sync_stream(r) -> bool:
    return hasattr(r, 'func')

def can_stream_sync(r) -> bool:
    return hasattr(r, 'func')

try:
    for chunk in r.stream(x):
        ...
except TypeError as e:
    if 'astream' in str(e):
        chunks = asyncio.run(collect(r.astream(x)))
    else:
        raise

When it happens

Trigger: `RunnableLambda(async_fn).stream(x)` or `.transform(iter([x]))`; a parent `RunnableSequence.stream()` containing an async-only lambda; sync consuming code over an async-built chain.

Common situations: Reusing async pipeline definitions in CLI scripts; LangServe/eval harnesses that call `.stream()`; partially migrated codebases mixing sync streaming with async lambdas.

Related errors

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

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