fix(core): add tool_call_id to on_tool_error event data

libs/core/langchain_core/runnables/schema.py

@@ -45,6 +45,15 @@ class EventData(TypedDict, total=False):
     chunks support addition in general, and adding them up should result
     in the output of the `Runnable` that generated the event.
     """
+    tool_call_id: NotRequired[str | None]
+    """The tool call ID associated with the tool execution.
+
+    This field is only available for tool-related events (e.g., `on_tool_error`)
+    and can be used to link errors to specific tool calls in stateless agent
+    implementations.
+
+    !!! version-added "Added in version 1.0.0"
+    """
 
 
 class BaseStreamEvent(TypedDict):

libs/core/langchain_core/tools/base.py

@@ -805,6 +805,7 @@ def run(
             # but if it is we will send a `None` value to the callback instead
             # TODO: will need to address issue via a patch.
             inputs=tool_input if isinstance(tool_input, dict) else None,
+            tool_call_id=tool_call_id,
             **kwargs,
         )
 
@@ -852,7 +853,7 @@ def run(
             error_to_raise = e
 
         if error_to_raise:
-            run_manager.on_tool_error(error_to_raise)
+            run_manager.on_tool_error(error_to_raise, tool_call_id=tool_call_id)
             raise error_to_raise
         output = _format_output(content, artifact, tool_call_id, self.name, status)
         run_manager.on_tool_end(output, color=color, name=self.name, **kwargs)
@@ -916,6 +917,7 @@ async def arun(
             # but if it is we will send a `None` value to the callback instead
             # TODO: will need to address issue via a patch.
             inputs=tool_input if isinstance(tool_input, dict) else None,
+            tool_call_id=tool_call_id,
             **kwargs,
         )
         content = None
@@ -965,7 +967,7 @@ async def arun(
             error_to_raise = e
 
         if error_to_raise:
-            await run_manager.on_tool_error(error_to_raise)
+            await run_manager.on_tool_error(error_to_raise, tool_call_id=tool_call_id)
             raise error_to_raise
 
         output = _format_output(content, artifact, tool_call_id, self.name, status)

libs/core/langchain_core/tracers/event_stream.py

@@ -72,6 +72,8 @@ class RunInfo(TypedDict):
     """The inputs to the run."""
     parent_run_id: UUID | None
     """The ID of the parent run."""
+    tool_call_id: NotRequired[str | None]
+    """The tool call ID associated with the run."""
 
 
 def _assign_name(name: str | None, serialized: dict[str, Any] | None) -> str:
@@ -300,6 +302,10 @@ def _write_run_start_info(
             # vs. None value.
             info["inputs"] = kwargs["inputs"]
 
+        if "tool_call_id" in kwargs:
+            # Store tool_call_id in run info for linking errors to tool calls
+            info["tool_call_id"] = kwargs["tool_call_id"]
+
         self.run_map[run_id] = info
         self.parent_map[run_id] = parent_run_id
 
@@ -658,6 +664,7 @@ async def on_tool_start(
             name_=name_,
             run_type="tool",
             inputs=inputs,
+            tool_call_id=kwargs.get("tool_call_id"),
         )
 
         self._send(
@@ -686,23 +693,27 @@ async def on_tool_error(
         **kwargs: Any,
     ) -> None:
         """Run when tool errors."""
+        # Extract tool_call_id from kwargs first, fallback to run_info if available
+        tool_call_id = kwargs.get("tool_call_id")
         run_info, inputs = self._get_tool_run_info_with_inputs(run_id)
-
-        self._send(
-            {
-                "event": "on_tool_error",
-                "data": {
-                    "error": error,
-                    "input": inputs,
-                },
-                "run_id": str(run_id),
-                "name": run_info["name"],
-                "tags": run_info["tags"],
-                "metadata": run_info["metadata"],
-                "parent_ids": self._get_parent_ids(run_id),
+        # If not in kwargs, check run_info (fallback for backward compatibility)
+        if tool_call_id is None:
+            tool_call_id = run_info.get("tool_call_id")
+
+        event: StandardStreamEvent = {
+            "event": "on_tool_error",
+            "data": {
+                "error": error,
+                "input": inputs,
+                "tool_call_id": tool_call_id,
             },
-            "tool",
-        )
+            "run_id": str(run_id),
+            "name": run_info["name"],
+            "tags": run_info["tags"],
+            "metadata": run_info["metadata"],
+            "parent_ids": self._get_parent_ids(run_id),
+        }
+        self._send(event, "tool")
 
     @override
     async def on_tool_end(self, output: Any, *, run_id: UUID, **kwargs: Any) -> None: