fix(mypy): fix presidio, panw, perplexity, and mcp hook type issues

Task 6: Fix Presidio guardrail type issues (5 errors)
- Cast event_hook lists to List[GuardrailEventHooks]
- Cast response to dict for _process_anthropic_response_for_pii
- Remove bytes from async_post_call_streaming_iterator_hook return type

Task 7: Fix PANW Prisma AIRS type issues (3 errors)
- Annotate contents as List[Dict[str, Any]]
- Add type annotation and type: ignore for error_obj dict

Task 8: Fix Perplexity responses type issues (5 errors)
- Change _ensure_message_type return type to Union[str, ResponseInputParam]
- Add explicit List[Any] annotation for result

Task 9: Fix MCP semantic filter hook override (1 error)
- Add litellm_call_info parameter to match superclass signature

Co-authored-by: yuneng-jiang <yuneng-jiang@users.noreply.github.com>
This commit is contained in:
Cursor Agent 2026-03-13 00:01:26 +00:00
parent b3a30a15c4
commit d6bb2946bc
No known key found for this signature in database
4 changed files with 10 additions and 8 deletions

View file

@ -63,12 +63,12 @@ class PerplexityResponsesConfig(OpenAIResponsesAPIConfig):
def _ensure_message_type(
self, input: Union[str, ResponseInputParam]
) -> Union[str, List[Dict[str, Any]]]:
) -> Union[str, ResponseInputParam]:
"""Ensure list input items have type='message' (required by Perplexity)."""
if isinstance(input, str):
return input
if isinstance(input, list):
result = []
result: List[Any] = []
for item in input:
if isinstance(item, dict) and "type" not in item:
item = {**item, "type": "message"}

View file

@ -315,6 +315,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
panw_metadata["litellm_trace_id"] = metadata["litellm_trace_id"]
# Build contents: tool_event takes priority, else prompt/response text
contents: List[Dict[str, Any]]
if tool_event is not None:
contents = [{"tool_event": tool_event}]
else:
@ -1485,7 +1486,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
detail = (
e.detail if isinstance(e.detail, dict) else {"message": str(e.detail)}
)
error_obj = dict(detail.get("error", detail))
error_obj: Dict[str, Any] = dict(detail.get("error", detail)) # type: ignore[arg-type]
error_obj["code"] = e.status_code
yield f"data: {json.dumps({'error': error_obj})}\n\n"
except Exception as e:

View file

@ -106,9 +106,9 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
if (self.output_parse_pii or self.apply_to_output) and not logging_only:
current_hook = self.event_hook
if isinstance(current_hook, str) and current_hook != "post_call":
self.event_hook = [current_hook, "post_call"]
self.event_hook = cast(List[GuardrailEventHooks], [current_hook, "post_call"])
elif isinstance(current_hook, list) and "post_call" not in current_hook:
self.event_hook = current_hook + ["post_call"]
self.event_hook = cast(List[GuardrailEventHooks], current_hook + ["post_call"])
self.pii_entities_config: Dict[Union[PiiEntityType, str], PiiAction] = (
pii_entities_config or {}
)
@ -908,7 +908,7 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
if self.apply_to_output is True:
if self._is_anthropic_message_response(response):
return await self._process_anthropic_response_for_pii(
response=response, request_data=data, mode="mask"
response=cast(dict, response), request_data=data, mode="mask"
)
return await self._mask_output_response(
response=response, request_data=data
@ -927,7 +927,7 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
)
elif self._is_anthropic_message_response(response):
await self._process_anthropic_response_for_pii(
response=response, request_data=data, mode="unmask"
response=cast(dict, response), request_data=data, mode="unmask"
)
return response
@ -1234,7 +1234,7 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
user_api_key_dict: UserAPIKeyAuth,
response: Any,
request_data: dict,
) -> AsyncGenerator[Union[ModelResponseStream, bytes], None]:
) -> AsyncGenerator[ModelResponseStream, None]:
"""
Process streaming response chunks to unmask PII tokens when needed.
"""

View file

@ -259,6 +259,7 @@ class SemanticToolFilterHook(CustomLogger):
user_api_key_dict: "UserAPIKeyAuth",
response: Any,
request_headers: Optional[Dict[str, str]] = None,
litellm_call_info: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, str]]:
"""Add semantic filter stats and tool names to response headers."""
from litellm.constants import MAX_MCP_SEMANTIC_FILTER_TOOLS_HEADER_LENGTH