Remove Anthropic-derived cache hints that Gemini/OpenAI transport cannot accept.
(body: dict[str, Any])
| 2053 | |
| 2054 | |
| 2055 | def _strip_openai_cache_hints_for_google(body: dict[str, Any]) -> None: |
| 2056 | """Remove Anthropic-derived cache hints that Gemini/OpenAI transport cannot accept.""" |
| 2057 | for tool in body.get("tools", []) or []: |
| 2058 | if isinstance(tool, dict): |
| 2059 | tool.pop("cache_control", None) |
| 2060 | |
| 2061 | for message in body.get("messages", []) or []: |
| 2062 | if not isinstance(message, dict): |
| 2063 | continue |
| 2064 | content = message.get("content") |
| 2065 | if not isinstance(content, list): |
| 2066 | continue |
| 2067 | |
| 2068 | flattened_text_parts: list[str] = [] |
| 2069 | normalized_parts: list[dict[str, Any]] = [] |
| 2070 | only_text_parts = True |
| 2071 | |
| 2072 | for item in content: |
| 2073 | if not isinstance(item, dict): |
| 2074 | continue |
| 2075 | normalized = dict(item) |
| 2076 | normalized.pop("cache_control", None) |
| 2077 | item_type = normalized.get("type") |
| 2078 | if item_type in {"text", "input_text"}: |
| 2079 | flattened_text_parts.append(str(normalized.get("text", ""))) |
| 2080 | else: |
| 2081 | only_text_parts = False |
| 2082 | normalized_parts.append(normalized) |
| 2083 | |
| 2084 | if only_text_parts: |
| 2085 | message["content"] = "\n".join(part for part in flattened_text_parts if part) |
| 2086 | continue |
| 2087 | |
| 2088 | if flattened_text_parts: |
| 2089 | normalized_parts.insert(0, { |
| 2090 | "type": "text", |
| 2091 | "text": "\n".join(part for part in flattened_text_parts if part), |
| 2092 | }) |
| 2093 | message["content"] = normalized_parts |
| 2094 | |
| 2095 | |
| 2096 | def _anthropic_messages_url(base_url: str) -> str: |
no test coverage detected