Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/_content_utils.py: 6%

1""" 

2Shared helpers for guardrail hooks: extract text from a request body 

3regardless of whether it uses Chat Completions ``messages``, Responses-API 

4``input``, or multimodal list-format ``content`` parts. 

5 

6Hooks that only check ``data["messages"]`` for string content silently 

7skip the other shapes — these helpers normalise that so every hook sees 

8every text fragment. 

9""" 

10 

11from collections.abc import Callable, Iterator, Mapping, Sequence 

12from typing import Any, Final 

13 

14# Call types whose body carries free-form chat / prompt text that 

15# text-content guardrails (banned keywords, content moderation, secret 

16# detection, …) should inspect. The proxy ingress passes ``route_type`` 

17# straight through as ``call_type``, so the literal values here are 

18# what the guardrail dispatcher actually receives: 

19# 

20# /v1/chat/completions -> "acompletion" 

21# /v1/responses -> "aresponses" 

22# 

23# ``"completion"`` is included for SDK / internal callers that invoke 

24# ``pre_call_hook`` directly with the sync name. Embedding, moderation, 

25# audio, and transcription endpoints are deliberately excluded — text 

26# guardrails on those paths are a separate scope. 

27TEXT_CONTENT_CALL_TYPES: Final[frozenset[str]] = frozenset({"completion", "acompletion", "aresponses"}) 

28 

29 

30def is_text_content_call_type(call_type: str) -> bool: 

31 """Return True if ``call_type`` carries free-form text that text 

32 guardrails should inspect (Chat Completions or Responses API).""" 

33 return call_type in TEXT_CONTENT_CALL_TYPES 

34 

35 

36# Call types whose request body carries no conversation at all. Embeddings carry 

37# ``input`` — documents being indexed, not a prompt — which 

38# :func:`build_inspection_messages` would lift into synthetic chat messages. 

39# 

40# Deny-list on purpose: ``TEXT_CONTENT_CALL_TYPES`` above omits conversational 

41# call types (``anthropic_messages``, ``responses``, ``call_mcp_tool``), so a 

42# blocking guardrail gated on that allow-list would stop inspecting real chat 

43# traffic. Testing this instead leaves an unrecognised call type inspected. 

44NON_CONVERSATIONAL_CALL_TYPES: Final[frozenset[str]] = frozenset({"embedding", "aembedding"}) 

45 

46 

47def is_non_conversational_call_type(call_type: str) -> bool: 

48 """Return True if ``call_type``'s body carries no conversation to inspect.""" 

49 return call_type in NON_CONVERSATIONAL_CALL_TYPES 

50 

51 

52TEXT_PART_TYPES: Final[frozenset[str]] = frozenset( 

53 {"text", "input_text", "output_text", "summary_text", "reasoning_text"} 

54) 

55 

56# Responses-API item types whose ``output`` field carries user/tool text 

57# that guardrails should inspect. ``function_call_output`` is the 

58# built-in shape; ``custom_tool_call_output`` is the custom-tool 

59# counterpart (see ``ChatCompletionCustomToolCallOutput``). 

60_OUTPUT_ITEM_TYPES: Final[frozenset[str]] = frozenset({"function_call_output", "custom_tool_call_output"}) 

61 

62 

63def _part_text(part: Mapping[str, object]) -> str | None: 

64 """Return non-empty plaintext from any content part that carries ``text``.""" 

65 if not isinstance(part, dict): 

66 return None 

67 text = part.get("text") 

68 if isinstance(text, str) and text: 

69 return text 

70 return None 

71 

72 

73def _iter_text_parts_in_content(content: object) -> Iterator[str]: 

74 """Yield text fragments from a ``message.content`` value (string or 

75 multimodal list). Non-text parts (images, audio, …) are skipped.""" 

76 if isinstance(content, str): 

77 if content: 

78 yield content 

79 elif isinstance(content, list): 

80 for part in content: 

81 if isinstance(part, str): 

82 # A bare string in a content/input list is itself a text 

83 # fragment (Responses-API mixed-list shape). 

84 if part: 

85 yield part 

86 continue 

87 if not isinstance(part, dict): 

88 continue 

89 text = _part_text(part) 

90 if text is not None: 

91 yield text 

92 

93 

94def _coerce_input_to_messages(input_value: object) -> list[dict[str, object]]: 

95 """Coerce a Responses-API ``data["input"]`` value into chat-style messages.""" 

96 if isinstance(input_value, str): 

97 return [{"role": "user", "content": input_value}] 

98 if not isinstance(input_value, list): 

99 return [] 

100 messages: Final[list[dict[str, object]]] = [] 

101 for item in input_value: 

102 if isinstance(item, str): 

103 messages.append({"role": "user", "content": item}) 

104 elif isinstance(item, dict): 

105 if _part_text(item) is not None: 

106 messages.append({"role": item.get("role") or "user", "content": [item]}) 

107 elif item.get("type") == "reasoning": 

108 if "content" in item: 

109 messages.append( 

110 { # mutable-ok: append reasoning content 

111 "role": item.get("role") or "assistant", 

112 "content": item["content"], 

113 } 

114 ) 

115 if isinstance(item.get("summary"), list): 

116 messages.append( 

117 { # mutable-ok: append reasoning summary 

118 "role": item.get("role") or "assistant", 

119 "content": item["summary"], 

120 } 

121 ) 

122 elif "content" in item: 

123 messages.append({"role": item.get("role") or "user", "content": item["content"]}) 

124 elif item.get("type") in _OUTPUT_ITEM_TYPES and "output" in item: 

125 messages.append({"role": item.get("role") or "tool", "content": item["output"]}) 

126 return messages 

127 

128 

129def _iter_inspection_messages(data: Mapping[str, object]) -> Iterator[object]: 

130 """Yield every message-like dict, walking ``messages`` AND ``input``.""" 

131 messages: Final = data.get("messages") 

132 if isinstance(messages, list): 

133 yield from messages 

134 yield from _coerce_input_to_messages(data.get("input")) 

135 

136 

137def iter_message_text(data: Mapping[str, object]) -> Iterator[str]: 

138 """Yield every text fragment from ``messages`` AND ``input``. 

139 

140 Walks every role (user, assistant, system, …) — guardrails inspect 

141 the entire conversation, not just user turns. 

142 """ 

143 for message in _iter_inspection_messages(data): 

144 if not isinstance(message, dict): 

145 continue 

146 yield from _iter_text_parts_in_content(message.get("content")) 

147 

148 

149def walk_user_text(data: dict[str, Any], visit: Callable[[str], str]) -> int: 

150 """Rewrite every text fragment in place via ``visit``. 

151 

152 Mutates ``data["messages"]`` and ``data["input"]``. Returns the number 

153 of fragments visited so callers can short-circuit when nothing was 

154 inspected. 

155 """ 

156 visited = 0 

157 

158 def _rewrite_content(content: object) -> object: 

159 nonlocal visited 

160 if isinstance(content, str): 

161 if content: 

162 visited += 1 

163 return visit(content) 

164 return content 

165 if isinstance(content, list): 

166 new_parts: Final[list[object]] = [] 

167 for part in content: 

168 if isinstance(part, str) and part: 

169 visited += 1 

170 new_parts.append(visit(part)) 

171 elif isinstance(part, dict) and _part_text(part) is not None: 

172 visited += 1 

173 new_parts.append({**part, "text": visit(part["text"])}) 

174 else: 

175 new_parts.append(part) 

176 return new_parts 

177 return content 

178 

179 messages: Final = data.get("messages") 

180 if isinstance(messages, list): 

181 for message in messages: 

182 if isinstance(message, dict) and "content" in message: 

183 message["content"] = _rewrite_content(message["content"]) 

184 

185 input_value: Final = data.get("input") 

186 if isinstance(input_value, str): 

187 if input_value: 

188 visited += 1 

189 data["input"] = visit(input_value) 

190 return visited 

191 if isinstance(input_value, list): 

192 for idx, item in enumerate(input_value): 

193 if isinstance(item, str): 

194 if item: 

195 visited += 1 

196 input_value[idx] = visit(item) 

197 elif isinstance(item, dict): 

198 if _part_text(item) is not None: 

199 visited += 1 

200 input_value[idx] = {**item, "text": visit(item["text"])} # mutable-ok: rewrite text part in place 

201 elif item.get("type") == "reasoning": 

202 if "content" in item: 

203 item["content"] = _rewrite_content(item["content"]) 

204 if isinstance(item.get("summary"), list): 

205 item["summary"] = _rewrite_content(item["summary"]) 

206 elif "content" in item: 

207 item["content"] = _rewrite_content(item["content"]) 

208 elif item.get("type") in _OUTPUT_ITEM_TYPES and "output" in item: 

209 item["output"] = _rewrite_content(item["output"]) 

210 return visited 

211 

212 return visited 

213 

214 

215def is_string_batch_input(data: Mapping[str, object]) -> bool: 

216 """Return True when the only inspected content is an ``input`` list of plain 

217 strings, the /embeddings batch shape, which :func:`apply_redacted_messages_back` 

218 rewrites element-wise.""" 

219 if "messages" in data: 

220 return False 

221 input_value: Final = data.get("input") 

222 return isinstance(input_value, list) and bool(input_value) and all(isinstance(item, str) for item in input_value) 

223 

224 

225def apply_redacted_messages_back(data: dict[str, Any], redacted_messages: Sequence[object]) -> bool: 

226 """Write redacted messages back to whichever field(s) the caller used. 

227 

228 Mask/anonymize paths take a synthesised messages list (from 

229 :func:`build_inspection_messages`), get a redacted version back from a 

230 third-party guardrail, and need to rewrite the request body. Writing 

231 only to ``data["messages"]`` leaves the Responses-API ``data["input"]`` 

232 field untouched, so the unredacted text still reaches the LLM. 

233 

234 This helper updates both fields when both are present. A string batch 

235 (``/embeddings`` ``input`` list) is rewritten element-wise: the n-th 

236 redacted message replaces the n-th non-empty element, because 

237 :func:`build_inspection_messages` emits one message per non-empty string. 

238 

239 Returns False, leaving ``data`` untouched, when a batch response does not 

240 carry exactly one message per inspected element: a partial rewrite would 

241 forward the remaining originals unredacted. Callers must block on False. 

242 """ 

243 if is_string_batch_input(data): 

244 batch: Final = data["input"] 

245 inspected_indices: Final = tuple(idx for idx, item in enumerate(batch) if item) 

246 if len(redacted_messages) != len(inspected_indices): 

247 return False 

248 if any(not isinstance(message, Mapping) or message.get("content") is None for message in redacted_messages): 

249 return False 

250 redacted_texts: Final = tuple( 

251 "\n".join(_iter_text_parts_in_content(message["content"])) for message in redacted_messages 

252 ) 

253 for idx, text in zip(inspected_indices, redacted_texts): 

254 batch[idx] = text 

255 return True 

256 if "messages" in data: 

257 data["messages"] = redacted_messages 

258 input_value: Final = data.get("input") 

259 if isinstance(input_value, str): 

260 text_parts: Final[list[str]] = [] 

261 for msg in redacted_messages: 

262 if not isinstance(msg, dict): 

263 continue 

264 text_parts.extend(_iter_text_parts_in_content(msg.get("content"))) 

265 data["input"] = "\n".join(text_parts) 

266 return True 

267 

268 

269def has_non_string_content(data: Mapping[str, object]) -> bool: 

270 """Return True if any inspected content is not a plain string. 

271 

272 Used by hooks whose mask/redact path operates on string offsets and 

273 therefore cannot preserve multimodal non-text parts. Such hooks should 

274 degrade to block-on-detect when this returns True so image/audio parts 

275 are not silently stripped during in-place masking. 

276 """ 

277 messages: Final = data.get("messages") 

278 if isinstance(messages, list): 

279 for message in messages: 

280 if isinstance(message, dict) and not isinstance(message.get("content"), str): 

281 if message.get("content") is not None: 

282 return True 

283 input_value: Final = data.get("input") 

284 if input_value is not None and not isinstance(input_value, str): 

285 return True 

286 return False 

287 

288 

289def build_inspection_messages(data: dict[str, Any]) -> list[dict[str, str]]: 

290 """Synthesize a chat-style messages list for posting to a guardrail API. 

291 

292 Each returned message has a plain-string ``content`` — multimodal text 

293 parts are joined with newlines and Responses-API ``input`` is lifted 

294 into synthetic messages. Messages with no inspectable text are dropped. 

295 

296 Hooks that POST ``{"messages": [...]}`` to an external service should 

297 call this instead of ``data.get("messages", [])`` so the Responses API 

298 and multimodal content are covered. 

299 """ 

300 flattened: Final[list[dict[str, str]]] = [] 

301 for message in _iter_inspection_messages(data): 

302 if not isinstance(message, dict): 

303 continue 

304 text = "\n".join(_iter_text_parts_in_content(message.get("content"))) 

305 if not text: 

306 continue 

307 role = message.get("role", "user") or "user" 

308 flattened.append({"role": role, "content": text}) 

309 return flattened