fix(guardrails): check Anthropic-format request tools in tool_permission pre_call

litellm/proxy/guardrails/guardrail_hooks/tool_permission.py

@@ -14,6 +14,7 @@
     CustomGuardrail,
     log_guardrail_information,
 )
+from litellm.llms.base_llm.guardrail_translation.utils import anthropic_tool_names
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.common_utils.callback_utils import (
     add_guardrail_to_applied_guardrails_header,
@@ -40,6 +41,7 @@
 )
 
 GUARDRAIL_NAME: Final = "tool_permission"
+_RESPONSES_CALL_TYPES: Final = frozenset({"responses", "aresponses", "_aresponses_websocket"})
 
 
 def _object_mapping(value: object) -> Mapping[str, object] | None:
@@ -603,27 +605,33 @@ def _denied_message(block: object) -> str | None:
         if not any(_is_tool_use_block(block) for block in kept_blocks):
             response["stop_reason"] = "end_turn"  # rebind-ok: dropping every tool_use ends the turn
 
-    def _get_request_tool_name(self, tool: object) -> tuple[str | None, str | None]:
+    def _get_request_tool_targets(
+        self, tool: object, call_type: CallTypesLiteral
+    ) -> tuple[tuple[str, str | None], ...]:
         tool_type: Final = self._get_mapping_value(tool, "type")
-        if tool_type != "function":
-            return None, tool_type
-
-        function: Final = self._get_mapping_value(tool, "function")
-        tool_name: Final = self._get_mapping_value(function, "name")
-        return tool_name, tool_type
+        normalized_type: Final = (
+            tool_type if call_type in _RESPONSES_CALL_TYPES or tool_type not in (None, "custom") else "function"
+        )
+        return tuple((name, normalized_type) for name in anthropic_tool_names(tool))
 
     def _get_legacy_function_name(self, function: object) -> str | None:
         return self._get_mapping_value(function, "name")
 
-    def _get_named_tool_choice(self, data: dict) -> str | None:
+    def _get_named_tool_choice(self, data: Mapping[str, object]) -> str | None:
         tool_choice: Final = data.get("tool_choice")
         if not tool_choice or tool_choice in ("auto", "none", "required"):
             return None
         if isinstance(tool_choice, str):
             return tool_choice
-        if self._get_mapping_value(tool_choice, "type") != "function":
+        if self._get_mapping_value(tool_choice, "type") not in ("tool", "function"):
             return None
-        return self._get_mapping_value(self._get_mapping_value(tool_choice, "function"), "name")
+        function_name: Final = self._get_mapping_value(self._get_mapping_value(tool_choice, "function"), "name")
+        return function_name or self._get_mapping_value(tool_choice, "name")
+
+    @staticmethod
+    def _is_anthropic_tool_choice(data: Mapping[str, object]) -> bool:
+        tool_choice: Final = _object_mapping(data.get("tool_choice"))
+        return tool_choice is not None and tool_choice.get("type") == "tool"
 
     def _get_named_function_call(self, data: dict) -> str | None:
         function_call: Final = data.get("function_call")
@@ -633,13 +641,11 @@ def _get_named_function_call(self, data: dict) -> str | None:
             return function_call
         return self._get_mapping_value(function_call, "name")
 
-    def _collect_request_tools(self, data: dict) -> list[tuple[str, str | None]]:
+    def _collect_request_tools(self, data: dict, call_type: CallTypesLiteral) -> list[tuple[str, str | None]]:
         request_tools: Final[list[tuple[str, str | None]]] = []
 
         for tool in data.get("tools") or []:
-            tool_name, tool_type = self._get_request_tool_name(tool)
-            if tool_name is not None:
-                request_tools.append((tool_name, tool_type))
+            request_tools.extend(self._get_request_tool_targets(tool, call_type))
 
         for function in data.get("functions") or []:
             function_name = self._get_legacy_function_name(function)
@@ -679,13 +685,9 @@ def _modify_request_with_permission_errors(
 
         tools: Final[list[ChatCompletionToolParam] | None] = data.get("tools")
         if tools is not None:
-            new_tools: Final = []
-            for tool in tools:
-                tool_name, tool_type = self._get_request_tool_name(tool)
-                if tool_type == "function" and tool_name in error_tool_names:
-                    continue
-                new_tools.append(tool)
-            data["tools"] = new_tools
+            data["tools"] = [
+                tool for tool in tools if not any(name in error_tool_names for name in anthropic_tool_names(tool))
+            ]
 
         functions: Final = data.get("functions")
         if functions is not None:
@@ -695,7 +697,7 @@ def _modify_request_with_permission_errors(
 
         named_tool_choice: Final = self._get_named_tool_choice(data)
         if named_tool_choice in error_tool_names:
-            data["tool_choice"] = "none"
+            data["tool_choice"] = {"type": "none"} if self._is_anthropic_tool_choice(data) else "none"
 
         named_function_call: Final = self._get_named_function_call(data)
         if named_function_call in error_tool_names:
@@ -805,7 +807,7 @@ async def async_pre_call_hook(
         if self.should_run_guardrail(data=data, event_type=event_type) is not True:
             return data
 
-        new_tools: Final = self._collect_request_tools(data)
+        new_tools: Final = self._collect_request_tools(data, call_type)
         if not new_tools:
             verbose_proxy_logger.debug(
                 "Tool Permission Guardrail: not running guardrail. No tools or functions in data"

tests/e2e/coverage_registry/guardrail.yaml

@@ -25,8 +25,8 @@
 - {id: guardrail.ibm_guardrails.post_call.blocks, module: guardrail, tier: P1, hook_point: post_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/ibm_guardrails/ibm_detector.py", rationale: "Output policy validation"}
 - {id: guardrail.semantic_guard.pre_call.blocks, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/semantic_guard", rationale: "Semantic policy compliance"}
 - {id: guardrail.block_code_execution.pre_call.blocks, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/block_code_execution", rationale: "Code-injection prevention"}
-- {id: guardrail.tool_permission.pre_call.allows, module: guardrail, tier: P1, hook_point: pre_call, assertions: [allows], exercised_on: [chat_completions], source: "guardrail_hooks/tool_permission.py", rationale: "Grant allowed tools"}
-- {id: guardrail.tool_permission.pre_call.blocks, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/tool_permission.py", rationale: "Block unauthorized tools"}
+- {id: guardrail.tool_permission.pre_call.allows, module: guardrail, tier: P1, hook_point: pre_call, assertions: [allows], exercised_on: [chat_completions, messages, responses], source: "guardrail_hooks/tool_permission.py", rationale: "Grant allowed tools"}
+- {id: guardrail.tool_permission.pre_call.blocks, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions, messages, responses], source: "guardrail_hooks/tool_permission.py", rationale: "Block unauthorized tools"}
 - {id: guardrail.microsoft_purview.pre_call.blocks, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/microsoft_purview/purview_dlp.py", rationale: "DLP sensitive-data disclosure"}
 - {id: guardrail.headroom.pre_call.blocks, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/headroom/headroom.py", rationale: "Anomaly detection threshold"}
 - {id: guardrail.generic_guardrail_api.pre_call.blocks, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/generic_guardrail_api/generic_guardrail_api.py", rationale: "Vendor-agnostic custom API"}

tests/e2e/guardrails/test_tool_permission_guardrail_e2e.py

@@ -19,10 +19,9 @@
 
 from __future__ import annotations
 
-from typing import Final
+from typing import Final, Literal
 
 import pytest
-
 from e2e_config import unique_marker
 from e2e_http import StreamingResponse, UnknownApiError
 from e2e_metadata import Capability, Domain, Mode, Provider, Subject, meta
@@ -31,14 +30,29 @@
     ToolPermissionParamsBody,
     ToolPermissionRuleBody,
     poll_until_blocked,
+    poll_until_blocked_stream,
     poll_until_guardrail_applied,
 )
 from lifecycle import ResourceManager
-from models import ChatResponse, ChatTool, ChatToolFunction
+from models import (
+    AnthropicCustomTool,
+    AnthropicMessagesBody,
+    AnthropicMessagesResponse,
+    AnthropicToolChoice,
+    ChatMessage,
+    ChatResponse,
+    ChatTool,
+    ChatToolFunction,
+    JsonSchemaProperty,
+    ResponsesFunctionTool,
+    ResponsesToolBody,
+    ResponsesToolOutput,
+    ToolInputSchema,
+)
 
-pytestmark = pytest.mark.e2e
+pytestmark: Final = pytest.mark.e2e
 
-MODEL = "gemini-2.5-flash"
+MODEL: Final = "gemini-2.5-flash"
 
 #: The one tool the guardrail permits, and one it does not. Both are declared by
 #: the caller in the request body; the guardrail reads them there.
@@ -63,11 +77,15 @@
 
 TOOL_PROMPT: Final = "What is the weather in Paris right now?"
 
+type FlatToolEndpoint = Literal["messages", "responses"]
+ALLOWED_TOOL_SCHEMA: Final = ToolInputSchema(properties={"city": JsonSchemaProperty(type="string")}, required=["city"])
+DENIED_TOOL_SCHEMA: Final = ToolInputSchema()
+
 
 def _register_tool_permission(client: GuardrailsClient, resources: ResourceManager, *, name: str) -> None:
     """Allow-list exactly one tool: everything else falls to `default_action=deny`
     and, with `on_disallowed_action=block`, is rejected outright."""
-    guardrail_id = client.register(
+    guardrail_id: Final = client.register(
         name,
         ToolPermissionParamsBody(
             mode="pre_call",
@@ -100,6 +118,57 @@ def _tool_call_names(response: ChatResponse) -> tuple[str, ...]:
     )
 
 
+def _send_flat_tool(
+    client: GuardrailsClient,
+    key: str,
+    endpoint: FlatToolEndpoint,
+    *,
+    guardrail: str,
+    tool: ChatTool,
+    schema: ToolInputSchema,
+    require_tool: bool = False,
+) -> StreamingResponse:
+    name: Final = tool.function.name
+    match endpoint:
+        case "messages":
+            return client.proxy.transport.send(
+                "/v1/messages",
+                headers=client.proxy.transport.bearer(key),
+                json=AnthropicMessagesBody(
+                    model=MODEL,
+                    messages=[ChatMessage(role="user", content=TOOL_PROMPT)],
+                    max_tokens=128,
+                    tools=[
+                        AnthropicCustomTool(name=name, description=tool.function.description or "", input_schema=schema)
+                    ],
+                    tool_choice=AnthropicToolChoice(type="any") if require_tool else None,
+                    guardrails=[guardrail],
+                ),
+            )
+        case "responses":
+            return client.proxy.transport.send(
+                "/v1/responses",
+                headers=client.proxy.transport.bearer(key),
+                json=ResponsesToolBody(
+                    model=MODEL,
+                    input=TOOL_PROMPT,
+                    tools=[ResponsesFunctionTool(name=name, description=tool.function.description, parameters=schema)],
+                    tool_choice="required" if require_tool else None,
+                    guardrails=[guardrail],
+                ),
+            )
+
+
+def _flat_tool_call_names(endpoint: FlatToolEndpoint, body: str) -> tuple[str, ...]:
+    match endpoint:
+        case "messages":
+            content: Final = AnthropicMessagesResponse.model_validate_json(body).content or []
+            return tuple(block.name for block in content if block.type == "tool_use" and block.name)
+        case "responses":
+            output: Final = ResponsesToolOutput.model_validate_json(body).output
+            return tuple(item.name for item in output if item.type == "function_call" and item.name)
+
+
 class TestToolPermissionPreCall:
     @pytest.mark.covers("guardrail.tool_permission.pre_call.blocks", exercised_on=["chat_completions"])
     @meta(
@@ -118,10 +187,10 @@ def test_pre_call_blocks_tool_outside_the_allow_list(
         rejected with a 400 that names the denied tool. An unauthorized tool that
         merely reaches the model is the whole failure mode this guardrail exists
         to prevent, so a 200 here is a hard failure."""
-        name = f"e2e-toolperm-block-{unique_marker()}"
+        name: Final = f"e2e-toolperm-block-{unique_marker()}"
         _register_tool_permission(client, resources, name=name)
 
-        result = poll_until_blocked(
+        result: Final = poll_until_blocked(
             lambda: client.chat(
                 scoped_key,
                 MODEL,
@@ -161,10 +230,10 @@ def test_pre_call_allows_permitted_tool(
         and the model calls it. Without the header check a guardrail that never
         attached would pass this test for the wrong reason, so the 200 alone is
         not the contract."""
-        name = f"e2e-toolperm-allow-{unique_marker()}"
+        name: Final = f"e2e-toolperm-allow-{unique_marker()}"
         _register_tool_permission(client, resources, name=name)
 
-        outcome = poll_until_guardrail_applied(
+        outcome: Final = poll_until_guardrail_applied(
             lambda: client.chat_raw(
                 scoped_key,
                 MODEL,
@@ -178,13 +247,117 @@ def test_pre_call_allows_permitted_tool(
         )
 
         assert outcome.ok, f"the permitted tool must be served, got {outcome.status_code}: {outcome.body[:400]}"
-        applied = _applied_guardrails(outcome)
+        applied: Final = _applied_guardrails(outcome)
         assert name in applied, (
             "the allowed call must carry x-litellm-applied-guardrails naming the guardrail; "
             f"without it the 200 only proves the guardrail never ran. Got {applied!r}"
         )
 
-        called = _tool_call_names(ChatResponse.model_validate_json(outcome.body))
+        called: Final = _tool_call_names(ChatResponse.model_validate_json(outcome.body))
+        assert called == (ALLOWED_TOOL.function.name,), (
+            f"the served call must carry one tool call for the permitted tool, got {called!r}: {outcome.body[:400]}"
+        )
+
+    @pytest.mark.parametrize(
+        "endpoint",
+        [
+            pytest.param(
+                "messages",
+                marks=pytest.mark.covers("guardrail.tool_permission.pre_call.blocks", exercised_on=["messages"]),
+            ),
+            pytest.param(
+                "responses",
+                marks=pytest.mark.covers("guardrail.tool_permission.pre_call.blocks", exercised_on=["responses"]),
+            ),
+        ],
+    )
+    @meta(
+        Subject(
+            domain=Domain.GUARDRAILS,
+            providers=(Provider.GEMINI,),
+            models=(MODEL,),
+            capabilities=(Capability.FUNCTION_CALLING,),
+            mode=Mode.NONSTREAM,
+        )
+    )
+    def test_pre_call_blocks_flat_tool_outside_the_allow_list(
+        self,
+        client: GuardrailsClient,
+        resources: ResourceManager,
+        scoped_key: str,
+        endpoint: FlatToolEndpoint,
+    ) -> None:
+        name: Final = f"e2e-toolperm-block-{endpoint}-{unique_marker()}"
+        _register_tool_permission(client, resources, name=name)
+
+        outcome: Final = poll_until_blocked_stream(
+            lambda: _send_flat_tool(
+                client, scoped_key, endpoint, guardrail=name, tool=DENIED_TOOL, schema=DENIED_TOOL_SCHEMA
+            )
+        )
+
+        assert outcome.status_code == 400, (
+            f"tool_permission let a tool outside the allow-list through /{endpoint}; "
+            f"got {outcome.status_code}: {outcome.body[:400]}"
+        )
+        assert DENIED_TOOL.function.name in outcome.body, (
+            f"the block must name the denied tool so the caller can fix the request; got: {outcome.body[:400]}"
+        )
+        assert "Violated guardrail policy" in outcome.body, (
+            f"the block must be a guardrail verdict, got: {outcome.body[:400]}"
+        )
+
+    @pytest.mark.parametrize(
+        "endpoint",
+        [
+            pytest.param(
+                "messages",
+                marks=pytest.mark.covers("guardrail.tool_permission.pre_call.allows", exercised_on=["messages"]),
+            ),
+            pytest.param(
+                "responses",
+                marks=pytest.mark.covers("guardrail.tool_permission.pre_call.allows", exercised_on=["responses"]),
+            ),
+        ],
+    )
+    @meta(
+        Subject(
+            domain=Domain.GUARDRAILS,
+            providers=(Provider.GEMINI,),
+            models=(MODEL,),
+            capabilities=(Capability.FUNCTION_CALLING,),
+            mode=Mode.NONSTREAM,
+        )
+    )
+    def test_pre_call_allows_permitted_flat_tool(
+        self,
+        client: GuardrailsClient,
+        resources: ResourceManager,
+        scoped_key: str,
+        endpoint: FlatToolEndpoint,
+    ) -> None:
+        name: Final = f"e2e-toolperm-allow-{endpoint}-{unique_marker()}"
+        _register_tool_permission(client, resources, name=name)
+
+        outcome: Final = poll_until_guardrail_applied(
+            lambda: _send_flat_tool(
+                client,
+                scoped_key,
+                endpoint,
+                guardrail=name,
+                tool=ALLOWED_TOOL,
+                schema=ALLOWED_TOOL_SCHEMA,
+                require_tool=True,
+            ),
+            name,
+        )
+
+        assert outcome.ok, f"the permitted tool must be served, got {outcome.status_code}: {outcome.body[:400]}"
+        applied: Final = _applied_guardrails(outcome)
+        assert name in applied, (
+            f"the allowed call must carry x-litellm-applied-guardrails naming the guardrail; got {applied!r}"
+        )
+        called: Final = _flat_tool_call_names(endpoint, outcome.body)
         assert called == (ALLOWED_TOOL.function.name,), (
             f"the served call must carry one tool call for the permitted tool, got {called!r}: {outcome.body[:400]}"
         )