_chain_async_tool_call_wrappers() — langchain Function Reference

Architecture documentation for the _chain_async_tool_call_wrappers() function in factory.py from the langchain codebase.

Function python LangChainCore ApiManagement called by 1

Entity Profile

LangChainCore→ ApiManagement→ _chain_async_tool_call_wrappers() — langchain Function Reference

Dependency Diagram

graph TD
  655c790d_8e18_faf9_28c0_bb4d668f853a["_chain_async_tool_call_wrappers()"]
  fd7a28b1_3772_169b_6524_1342f35143b1["factory.py"]
  655c790d_8e18_faf9_28c0_bb4d668f853a -->|defined in| fd7a28b1_3772_169b_6524_1342f35143b1
  f4b66c38_651c_807a_caca_41f73fbbe516["create_agent()"]
  f4b66c38_651c_807a_caca_41f73fbbe516 -->|calls| 655c790d_8e18_faf9_28c0_bb4d668f853a
  style 655c790d_8e18_faf9_28c0_bb4d668f853a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/factory.py lines 589–650

def _chain_async_tool_call_wrappers(
    wrappers: Sequence[
        Callable[
            [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],
            Awaitable[ToolMessage | Command[Any]],
        ]
    ],
) -> (
    Callable[
        [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],
        Awaitable[ToolMessage | Command[Any]],
    ]
    | None
):
    """Compose async wrappers into middleware stack (first = outermost).

    Args:
        wrappers: Async wrappers in middleware order.

    Returns:
        Composed async wrapper, or `None` if empty.
    """
    if not wrappers:
        return None

    if len(wrappers) == 1:
        return wrappers[0]

    def compose_two(
        outer: Callable[
            [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],
            Awaitable[ToolMessage | Command[Any]],
        ],
        inner: Callable[
            [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],
            Awaitable[ToolMessage | Command[Any]],
        ],
    ) -> Callable[
        [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],
        Awaitable[ToolMessage | Command[Any]],
    ]:
        """Compose two async wrappers where outer wraps inner."""

        async def composed(
            request: ToolCallRequest,
            execute: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
        ) -> ToolMessage | Command[Any]:
            # Create an async callable that invokes inner with the original execute
            async def call_inner(req: ToolCallRequest) -> ToolMessage | Command[Any]:
                return await inner(req, execute)

            # Outer can call call_inner multiple times
            return await outer(request, call_inner)

        return composed

    # Chain all wrappers: first -> second -> ... -> last
    result = wrappers[-1]
    for wrapper in reversed(wrappers[:-1]):
        result = compose_two(wrapper, result)

    return result