test_tool_error_event_includes_tool_call_id() — langchain Function Reference

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 2842–2879

async def test_tool_error_event_includes_tool_call_id() -> None:
    """Test that on_tool_error event includes tool_call_id when provided."""

    @tool
    def failing_tool(x: int) -> str:  # noqa: ARG001
        """A tool that always fails."""
        msg = "Tool execution failed"
        raise ValueError(msg)

    tool_call_id = "test-tool-call-id-123"

    # Invoke the tool with a tool call dict that includes the tool_call_id
    tool_call = {
        "name": "failing_tool",
        "args": {"x": 42},
        "id": tool_call_id,
        "type": "tool_call",
    }

    events: list[StreamEvent] = []

    # Need to use async for loop to collect events before exception is raised.
    # List comprehension would fail entirely when exception occurs.
    async def collect_events() -> None:
        async for event in failing_tool.astream_events(tool_call, version="v2"):
            events.append(event)  # noqa: PERF401

    with pytest.raises(ValueError, match="Tool execution failed"):
        await collect_events()

    # Find the on_tool_error event
    error_events = [e for e in events if e["event"] == "on_tool_error"]
    assert len(error_events) == 1

    error_event = error_events[0]
    assert error_event["name"] == "failing_tool"
    assert "tool_call_id" in error_event["data"]
    assert error_event["data"]["tool_call_id"] == tool_call_id