feat(langchain_v1): add on_tool_call middleware hook

[eyurtsev]

Adds generator-based middleware for intercepting tool execution in agents. Middleware can retry on errors, cache results, modify requests, or short-circuit execution.

Implementation

Middleware Protocol

class AgentMiddleware:
    def on_tool_call(
        self,
        request: ToolCallRequest,
        state: StateT,
        runtime: Runtime[ContextT],
    ) -> Generator[ToolCallRequest | ToolMessage | Command, ToolMessage | Command, None]:
        """
        Yields: ToolCallRequest (execute), ToolMessage (cached result), or Command (control flow)
        Receives: ToolMessage or Command via .send()
        Returns: None (final result is last value sent to handler)
        """
        yield request  # passthrough

Composition
Multiple middleware compose automatically (first = outermost), with _chain_tool_call_handlers() stacking them like nested function calls.

Examples

Retry on error:

class RetryMiddleware(AgentMiddleware):
    def on_tool_call(self, request, state, runtime):
        for attempt in range(3):
            response = yield request
            if not isinstance(response, ToolMessage) or response.status != "error":
                return
            if attempt == 2:
                return  # Give up

Cache results:

class CacheMiddleware(AgentMiddleware):
    def on_tool_call(self, request, state, runtime):
        cache_key = (request.tool_call["name"], tuple(request.tool_call["args"].items()))
        if cached := self.cache.get(cache_key):
            yield ToolMessage(content=cached, tool_call_id=request.tool_call["id"])
        else:
            response = yield request
            self.cache[cache_key] = response.content

Emulate tools with LLM

class ToolEmulator(AgentMiddleware):
    def on_tool_call(self, request, state, runtime):
        prompt = f"""Emulate: {request.tool_call["name"]}
Description: {request.tool.description}
Args: {request.tool_call["args"]}
Return ONLY the tool's output."""

        response = emulator_model.invoke([HumanMessage(prompt)])
        yield ToolMessage(
            content=response.content,
            tool_call_id=request.tool_call["id"],
            name=request.tool_call["name"],
        )

Modify requests:

class ScalingMiddleware(AgentMiddleware):
    def on_tool_call(self, request, state, runtime):
        if "value" in request.tool_call["args"]:
            request.tool_call["args"]["value"] *= 2
        yield request

[sydney-runkle]

Why this part?

[ToolCallRequest, Any, Any]

[eyurtsev]

Left a note, but in tool node those things don't behave as state and runtime necessarily... We can update the 3rd one to guaranteed runtime in a bit

[sydney-runkle]

Makes sense we don't want to import here

[sydney-runkle]

Similar notes as previous PR, would love to look at tracing more carefully for nested handlers.

[sydney-runkle]

can we confirm the exact contents + order of the call log?

[sydney-runkle]

we do in the test below, so this seems not relevant

[sydney-runkle]

ah yeah no async support right?

[eyurtsev]

Not right now. It would be post v1 thing. And you can get really far without async logic with this interceptor pattern -- you need async if the actual interceptor needs to do IO (e.g., access redis)

[eyurtsev]

@sydney-runkle , i'll likely need to fix this in 2 separate changes:

1. We need a fix to expose actual state here (similar to the patch I made a few weeks back -- i'll port it to this code)
2. We need to update the unit tests for ToolNode to always run within langgraph runtime (as it can only be used from within langgraph runtime), that'll allow us to guarantee that the runtime exists (except for python 3.10 async)

[eyurtsev]

let's check if we're OK with functionality everywhere, and can then do a quick follow through to fix types

[sydney-runkle]

Exciting pattern.

Want to further investigate tracing. Happy to help document!