feat: transformation.py

proper responses api tool handling for guardrail translation layer

Author: Krrish Dholakia

litellm/llms/openai/responses/guardrail_translation/handler.py

@@ -96,10 +96,11 @@ async def process_input_messages(
         # Handle simple string input
         if isinstance(input_data, str):
             inputs = GenericGuardrailAPIInputs(texts=[input_data])
+            original_tools: List[Dict[str, Any]] = []
 
             # Extract and transform tools if present
-
             if "tools" in data and data["tools"]:
+                original_tools = list(data["tools"])
                 self._extract_and_transform_tools(data["tools"], tools_to_check)
                 if tools_to_check:
                     inputs["tools"] = tools_to_check
@@ -118,9 +119,9 @@ async def process_input_messages(
             )
             guardrailed_texts = guardrailed_inputs.get("texts", [])
             data["input"] = guardrailed_texts[0] if guardrailed_texts else input_data
-            guardrailed_tools = guardrailed_inputs.get("tools")
-            if guardrailed_tools is not None:
-                data["tools"] = guardrailed_tools
+            self._apply_guardrailed_tools_to_data(
+                data, original_tools, guardrailed_inputs.get("tools")
+            )
             verbose_proxy_logger.debug("OpenAI Responses API: Processed string input")
             return data
 
@@ -131,8 +132,7 @@ async def process_input_messages(
         texts_to_check: List[str] = []
         images_to_check: List[str] = []
         task_mappings: List[Tuple[int, Optional[int]]] = []
-        # Track (message_index, content_index) for each text
-        # content_index is None for string content, int for list content
+        original_tools_list: List[Dict[str, Any]] = list(data.get("tools") or [])
 
         # Step 1: Extract all text content, images, and tools
         for msg_idx, message in enumerate(input_data):
@@ -169,9 +169,11 @@ async def process_input_messages(
             )
 
             guardrailed_texts = guardrailed_inputs.get("texts", [])
-            guardrailed_tools = guardrailed_inputs.get("tools")
-            if guardrailed_tools is not None:
-                data["tools"] = guardrailed_tools
+            self._apply_guardrailed_tools_to_data(
+                data,
+                original_tools_list,
+                guardrailed_inputs.get("tools"),
+            )
 
             # Step 3: Map guardrail responses back to original input structure
             await self._apply_guardrail_responses_to_input(
@@ -209,6 +211,53 @@ def _extract_and_transform_tools(
                 cast(List[ChatCompletionToolParam], transformed_tools)
             )
 
+    def _remap_tools_to_responses_api_format(
+        self, guardrailed_tools: List[Any]
+    ) -> List[Dict[str, Any]]:
+        """
+        Remap guardrail-returned tools (Chat Completion format) back to
+        Responses API request tool format.
+        """
+        return LiteLLMCompletionResponsesConfig.transform_chat_completion_tool_params_to_responses_api_tools(
+            guardrailed_tools  # type: ignore
+        )
+
+    def _merge_tools_after_guardrail(
+        self,
+        original_tools: List[Dict[str, Any]],
+        remapped: List[Dict[str, Any]],
+    ) -> List[Dict[str, Any]]:
+        """
+        Merge remapped guardrailed tools with original tools that were not sent
+        to the guardrail (e.g. web_search, web_search_preview), preserving order.
+        """
+        if not original_tools:
+            return remapped
+        result: List[Dict[str, Any]] = []
+        j = 0
+        for tool in original_tools:
+            if isinstance(tool, dict) and tool.get("type") in (
+                "web_search",
+                "web_search_preview",
+            ):
+                result.append(tool)
+            else:
+                if j < len(remapped):
+                    result.append(remapped[j])
+                    j += 1
+        return result
+
+    def _apply_guardrailed_tools_to_data(
+        self,
+        data: dict,
+        original_tools: List[Dict[str, Any]],
+        guardrailed_tools: Optional[List[Any]],
+    ) -> None:
+        """Remap guardrailed tools to Responses API format and merge with original, then set data['tools']."""
+        if guardrailed_tools is not None:
+            remapped = self._remap_tools_to_responses_api_format(guardrailed_tools)
+            data["tools"] = self._merge_tools_after_guardrail(original_tools, remapped)
+
     def _extract_input_text_and_images(
         self,
         message: Any,  # Can be Dict[str, Any] or ResponseInputParam
@@ -413,7 +462,10 @@ async def process_output_streaming_response(
                     List[ChatCompletionToolCallChunk], tool_calls
                 )
                 # Include model information if available
-                if hasattr(model_response_stream, "model") and model_response_stream.model:
+                if (
+                    hasattr(model_response_stream, "model")
+                    and model_response_stream.model
+                ):
                     inputs["model"] = model_response_stream.model
                 _guardrailed_inputs = await guardrail_to_apply.apply_guardrail(
                     inputs=inputs,
@@ -454,15 +506,21 @@ async def process_output_streaming_response(
                     )
                     return responses_so_far
             else:
-                verbose_proxy_logger.debug("Skipping output guardrail - model response has no choices")
+                verbose_proxy_logger.debug(
+                    "Skipping output guardrail - model response has no choices"
+                )
         # model_response_stream = OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(final_chunk)
         # tool_calls = model_response_stream.choices[0].tool_calls
         # convert openai response to model response
         string_so_far = self.get_streaming_string_so_far(responses_so_far)
         inputs = GenericGuardrailAPIInputs(texts=[string_so_far])
         # Try to get model from the final chunk if available
         if isinstance(final_chunk, dict):
-            response_model = final_chunk.get("response", {}).get("model") if isinstance(final_chunk.get("response"), dict) else None
+            response_model = (
+                final_chunk.get("response", {}).get("model")
+                if isinstance(final_chunk.get("response"), dict)
+                else None
+            )
             if response_model:
                 inputs["model"] = response_model
         _guardrailed_inputs = await guardrail_to_apply.apply_guardrail(
@@ -597,8 +655,8 @@ def _extract_output_text_and_images(
                     content = generic_response_output_item.content
             except Exception:
                 # Try to extract content directly from output_item if validation fails
-                if hasattr(output_item, "content") and output_item.content: # type: ignore
-                    content = output_item.content # type: ignore
+                if hasattr(output_item, "content") and output_item.content:  # type: ignore
+                    content = output_item.content  # type: ignore
                 else:
                     return
         elif isinstance(output_item, dict):
@@ -675,10 +733,10 @@ async def _apply_guardrail_responses_to_output(
                         if isinstance(content_item, OutputText):
                             content_item.text = guardrail_response
                             # Update the original response output
-                            if hasattr(output_item, "content") and output_item.content: # type: ignore
-                                original_content = output_item.content[content_idx] # type: ignore
+                            if hasattr(output_item, "content") and output_item.content:  # type: ignore
+                                original_content = output_item.content[content_idx]  # type: ignore
                                 if hasattr(original_content, "text"):
-                                    original_content.text = guardrail_response # type: ignore
+                                    original_content.text = guardrail_response  # type: ignore
                 except Exception:
                     pass
             elif isinstance(output_item, dict):

litellm/proxy/_new_secret_config.yaml

@@ -13,3 +13,25 @@ model_list:
   - model_name: gpt-4.1-mini
     litellm_params:
       model: openai/gpt-4.1-mini
+  - model_name: gpt-5-mini
+    litellm_params:
+      model: openai/gpt-5-mini
+
+
+guardrails:
+  - guardrail_name: mcp-user-permissions
+    litellm_params:
+      guardrail: mcp_end_user_permission
+      mode: pre_call
+      default_on: true
+
+mcp_servers:
+  my_http_server:
+    url: "http://0.0.0.0:8001/mcp"
+    transport: "http"
+    description: "My custom MCP server"
+    available_on_public_internet: true
+
+general_settings:
+  store_model_in_db: true
+  store_prompts_in_spend_logs: true

litellm/proxy/guardrails/guardrail_hooks/mcp_end_user_permission/mcp_end_user_permission.py

@@ -9,6 +9,7 @@
 - end_user_id, no mcp_servers  → allow all (default)
 - end_user_id + mcp_servers    → allow only those servers
 """
+
 from typing import TYPE_CHECKING, Any, List, Literal, Optional, Type
 
 from litellm._logging import verbose_proxy_logger
@@ -78,8 +79,10 @@ async def _check_request_tools(
         if not tools:
             return inputs
 
-        allowed_mcp_servers = await self._get_allowed_mcp_servers_from_object_permission(
-            object_permission
+        allowed_mcp_servers = (
+            await self._get_allowed_mcp_servers_from_object_permission(
+                object_permission
+            )
         )
         if allowed_mcp_servers is None:
             return inputs  # No restrictions → pass through unchanged
@@ -93,7 +96,9 @@ async def _check_request_tools(
 
         for tool in tools:
             tool_name = self._get_tool_name_from_definition(tool)
-            server_name = self._extract_mcp_server_name(tool_name) if tool_name else None
+            server_name = (
+                self._extract_mcp_server_name(tool_name) if tool_name else None
+            )
 
             if server_name is None:
                 # Not an MCP tool (no prefix) or unrecognised format → keep
@@ -138,7 +143,9 @@ async def _resolve_end_user_object_permission(
         end_user_object = await MCPEndUserPermissionGuardrail._fetch_end_user_object(
             end_user_id
         )
-        return end_user_object.object_permission if end_user_object is not None else None
+        return (
+            end_user_object.object_permission if end_user_object is not None else None
+        )
 
     @staticmethod
     def _get_end_user_id_from_request_data(request_data: dict) -> Optional[str]:

litellm/responses/litellm_completion_transformation/transformation.py

@@ -6,6 +6,7 @@
 from typing import Any, Dict, List, Literal, Optional, Set, Tuple, Union, cast
 
 from openai.types.responses import ResponseFunctionToolCall
+from openai.types.responses.response_create_params import ResponseInputParam
 from openai.types.responses.tool_param import FunctionToolParam
 from typing_extensions import TypedDict
 
@@ -32,7 +33,6 @@
     OpenAIWebSearchUserLocation,
     OutputTokensDetails,
     ResponseAPIUsage,
-    ResponseInputParam,
     ResponsesAPIOptionalRequestParams,
     ResponsesAPIResponse,
     ResponsesAPIStatus,
@@ -738,9 +738,25 @@ def _add_tool_call_to_assistant(
 
     @staticmethod
     def _ensure_tool_results_have_corresponding_tool_calls(
-        messages: List[Union[AllMessageValues, GenericChatCompletionMessage, ChatCompletionResponseMessage]],
+        messages: Sequence[
+            Union[
+                AllMessageValues,
+                GenericChatCompletionMessage,
+                ChatCompletionResponseMessage,
+                ChatCompletionMessageToolCall,
+                Message,
+            ]
+        ],
         tools: Optional[List[Any]] = None,
-    ) -> List[Union[AllMessageValues, GenericChatCompletionMessage, ChatCompletionResponseMessage]]:
+    ) -> List[
+        Union[
+            AllMessageValues,
+            GenericChatCompletionMessage,
+            ChatCompletionResponseMessage,
+            ChatCompletionMessageToolCall,
+            Message,
+        ]
+    ]:
         """
         Ensure that tool_result messages have corresponding tool_calls in the previous assistant message.
         
@@ -755,11 +771,19 @@ def _ensure_tool_results_have_corresponding_tool_calls(
             List of messages with tool_calls added to assistant messages when needed
         """
         if not messages:
-            return messages
-        
-        # Create a deep copy to avoid modifying the original
+            return list(messages)
+
+        # Create a deep copy to avoid modifying the original (use list() so we can mutate and return List)
         import copy
-        fixed_messages = copy.deepcopy(messages)
+        fixed_messages: List[
+            Union[
+                AllMessageValues,
+                GenericChatCompletionMessage,
+                ChatCompletionResponseMessage,
+                ChatCompletionMessageToolCall,
+                Message,
+            ]
+        ] = list(copy.deepcopy(messages))
         messages_to_remove = []
 
         # Count non-tool messages to avoid removing all messages
@@ -1306,6 +1330,50 @@ def transform_responses_api_tools_to_chat_completion_tools(
                 chat_completion_tools.append(cast(Union[ChatCompletionToolParam, OpenAIMcpServerTool], tool))
         return chat_completion_tools, web_search_options
 
+    @staticmethod
+    def transform_chat_completion_tool_params_to_responses_api_tools(
+        chat_completion_tools: Optional[
+            List[Union[ChatCompletionToolParam, OpenAIMcpServerTool]]
+        ],
+    ) -> List[Dict[str, Any]]:
+        """
+        Transform Chat Completion tool params (e.g. from guardrail output) back to
+        Responses API request tool format. Inverse of
+        transform_responses_api_tools_to_chat_completion_tools for the tools list.
+        """
+        if chat_completion_tools is None or not chat_completion_tools:
+            return []
+        result: List[Dict[str, Any]] = []
+        for tool in chat_completion_tools:
+            if not isinstance(tool, dict):
+                result.append(tool)  # type: ignore
+                continue
+            if tool.get("type") == "function":
+                fn = tool.get("function") or {}
+                parameters = dict(fn.get("parameters", {}) or {})
+                if not parameters or "type" not in parameters:
+                    parameters["type"] = "object"
+                responses_tool: Dict[str, Any] = {
+                    "type": "function",
+                    "name": fn.get("name") or "",
+                    "description": fn.get("description") or "",
+                    "parameters": parameters,
+                    "strict": fn.get("strict", False) or False,
+                }
+                if tool.get("cache_control") is not None:
+                    responses_tool["cache_control"] = tool.get("cache_control")
+                if tool.get("defer_loading") is not None:
+                    responses_tool["defer_loading"] = tool.get("defer_loading")
+                if tool.get("allowed_callers") is not None:
+                    responses_tool["allowed_callers"] = tool.get("allowed_callers")
+                if tool.get("input_examples") is not None:
+                    responses_tool["input_examples"] = tool.get("input_examples")
+                result.append(responses_tool)
+            else:
+                # mcp or other: pass through unchanged
+                result.append(dict(tool))
+        return result
+
     @staticmethod
     def transform_chat_completion_tools_to_responses_tools(
         chat_completion_response: ModelResponse,