langchain-ai/langchain · error · NotImplementedError

StructuredTool does not support sync invocation.

Error message

StructuredTool does not support sync invocation.

What it means

Identical contract to Tool: StructuredTool._run raises NotImplementedError when only a coroutine was supplied. The structured schema exists, but there is no sync function, so .invoke/.run cannot execute the tool.

Source

Thrown at libs/core/langchain_core/tools/structured.py:99

        """Use the tool.

        Args:
            *args: Positional arguments to pass to the tool
            config: Configuration for the run
            run_manager: Optional callback manager to use for the run
            **kwargs: Keyword arguments to pass to the tool

        Returns:
            The result of the tool execution
        """
        if self.func:
            if run_manager and signature(self.func).parameters.get("callbacks"):
                kwargs["callbacks"] = run_manager.get_child()
            if config_param := _get_runnable_config_param(self.func):
                kwargs[config_param] = config
            return self.func(*args, **kwargs)
        msg = "StructuredTool does not support sync invocation."
        raise NotImplementedError(msg)

    async def _arun(
        self,
        *args: Any,
        config: RunnableConfig,
        run_manager: AsyncCallbackManagerForToolRun | None = None,
        **kwargs: Any,
    ) -> Any:
        """Use the tool asynchronously.

        Args:
            *args: Positional arguments to pass to the tool
            config: Configuration for the run
            run_manager: Optional callback manager to use for the run
            **kwargs: Keyword arguments to pass to the tool

        Returns:
            The result of the tool execution

View on GitHub (pinned to e32fa9a52e)

Solutions

  1. Run the workflow asynchronously: await tool.ainvoke(...) / await agent.ainvoke(...)
  2. Add a sync func implementation (or a bridge) alongside coroutine
  3. If a third-party tool is async-only, wrap it: StructuredTool.from_function(func=lambda **kw: asyncio.run(async_fn(**kw)), coroutine=async_fn, ...)

Example fix

# before
async def fetch(url: str) -> str: ...
tool = StructuredTool.from_function(coroutine=fetch, name="fetch", description="Fetch")
tool.invoke({"url": "https://x"})  # NotImplementedError

# after
result = asyncio.run(tool.ainvoke({"url": "https://x"}))
Defensive patterns

Strategy: type-guard

Validate before calling

def call_tool(tool, payload):
    if getattr(tool, "func", None) is not None:
        return tool.invoke(payload)
    import asyncio
    return asyncio.run(tool.ainvoke(payload))

Type guard

def is_sync_tool(tool: object) -> bool:
    return callable(getattr(tool, "func", None))

Try / catch

try:
    out = tool.invoke(payload)
except NotImplementedError as e:
    if "sync invocation" in str(e):
        out = asyncio.run(tool.ainvoke(payload))
    else:
        raise

Prevention

When it happens

Trigger: StructuredTool.from_function(coroutine=async_fn) then tool.invoke({...}); agent/ToolNode driven synchronously (agent.invoke) with async-only StructuredTools.

Common situations: Async-native integrations (DB, HTTP tools) used inside sync notebook scripts; mixing sync agent runners with async tools; frameworks calling tool.func directly.

Related errors


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