fix(mypy): Fix type errors across multiple files

- vertex_ai/gemini/transformation.py: Fix TypedDict assignment via dict alias
- mcp_server/server.py: Convert ASGI scope to dict for type compatibility
- pass_through_endpoints.py: Add explicit Optional[dict] type annotation
- vector_store_endpoints/endpoints.py: Add Any type for dynamic proxy hook
- responses transformation.py: Use dict(Reasoning()) and setattr for compatibility
- zscaler_ai_guard.py: Add assert for api_base nullability
This commit is contained in:
OpenClaw 2026-02-14 02:17:43 +00:00
parent df54e1b452
commit 29a3345d05
6 changed files with 13 additions and 14 deletions

View file

@ -533,11 +533,12 @@ def _pop_and_merge_extra_body(data: RequestBody, optional_params: dict) -> None:
"""Pop extra_body from optional_params and shallow-merge into data, deep-merging dict values."""
extra_body: Optional[dict] = optional_params.pop("extra_body", None)
if extra_body is not None:
data_dict: dict = data # type: ignore[assignment]
for k, v in extra_body.items():
if k in data and isinstance(data[k], dict) and isinstance(v, dict):
data[k].update(v)
if k in data_dict and isinstance(data_dict[k], dict) and isinstance(v, dict):
data_dict[k].update(v)
else:
data[k] = v
data_dict[k] = v
def _transform_request_body(

View file

@ -2029,7 +2029,7 @@ if MCP_AVAILABLE:
# Inject masked debug headers when client sends x-litellm-mcp-debug: true
_debug_headers = MCPDebug.maybe_build_debug_headers(
raw_headers=raw_headers,
scope=scope,
scope=dict(scope),
mcp_servers=mcp_servers,
mcp_auth_header=mcp_auth_header,
mcp_server_auth_headers=mcp_server_auth_headers,

View file

@ -1193,14 +1193,11 @@ def create_pass_through_route(
final_query_params.update(query_params)
# When a caller (e.g. bedrock_proxy_route) supplies a pre-built
# body, use it instead of the body parsed from the raw request.
final_custom_body: Optional[dict] = None
if custom_body is not None:
final_custom_body = custom_body
else:
final_custom_body = (
custom_body_data
if isinstance(custom_body_data, dict) or custom_body_data is None
else None
)
elif isinstance(custom_body_data, dict):
final_custom_body = custom_body_data
return await pass_through_request( # type: ignore
request=request,

View file

@ -1,4 +1,4 @@
from typing import Dict, Optional
from typing import Any, Dict, Optional
from fastapi import APIRouter, Depends, HTTPException, Request, Response
@ -230,7 +230,7 @@ async def vector_store_create(
)
# Get managed vector stores hook
managed_vector_stores = proxy_logging_obj.get_proxy_hook("managed_vector_stores")
managed_vector_stores: Any = proxy_logging_obj.get_proxy_hook("managed_vector_stores")
if managed_vector_stores is None:
raise HTTPException(
status_code=500,

View file

@ -1500,7 +1500,7 @@ class LiteLLMCompletionResponsesConfig:
previous_response_id=getattr(
chat_completion_response, "previous_response_id", None
),
reasoning=Reasoning(),
reasoning=dict(Reasoning()),
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
finish_reason
),
@ -1516,7 +1516,7 @@ class LiteLLMCompletionResponsesConfig:
# Surface provider-specific fields (generic passthrough from any provider)
provider_fields = responses_api_response._hidden_params.get("provider_specific_fields")
if provider_fields:
responses_api_response.provider_specific_fields = provider_fields
setattr(responses_api_response, "provider_specific_fields", provider_fields)
return responses_api_response

View file

@ -106,6 +106,7 @@ class ZscalerAIGuardConfigModel(GuardrailConfigModel):
)
# Check for configuration issues
assert api_base is not None # always set via env default above
is_resolve_policy = api_base.endswith("/resolve-and-execute-policy")
is_execute_policy = api_base.endswith("/execute-policy") and not is_resolve_policy