wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resolve pyright errors in Anthropic get_messages_for_logging

Browse Source 
Content items in MessageParam have a heterogeneous union type (Pydantic
ContentBlock variants and TypedDict *BlockParam variants), neither of
which supports the dict-style access and mutation this sanitizer does.
Treat the deepcopied message as a plain dict and guard each content
item with isinstance(item, dict) — matches the runtime shape produced
by _from_standard_message and avoids crashing if a non-dict ever flows
through the LLMSpecificMessage path.

Drops anthropic_adapter.py from 115 to 23 pyright errors.
This commit is contained in:
Paul Kompfner
2026-04-27 14:14:57 -04:00
parent f1eef9ba0a
commit 70aeb5c7c2
1 changed files with 13 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/pipecat/adapters/services/anthropic_adapter.py
View File

@@ -121,16 +121,20 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
messages = self._from_universal_context_messages(self.get_messages(context)).messages messages = self._from_universal_context_messages(self.get_messages(context)).messages
# Sanitize messages for logging # Sanitize messages for logging
messages_for_logging = [] messages_for_logging: list[dict[str, Any]] = []
for message in messages: for message in messages:
msg = copy.deepcopy(message) msg: dict[str, Any] = copy.deepcopy(dict(message))
if "content" in msg: content = msg.get("content")
if isinstance(msg["content"], list): if isinstance(content, list):
for item in msg["content"]: for item in content:
if item["type"] == "image": if not isinstance(item, dict):
item["source"]["data"] = "..." continue
if item["type"] == "thinking" and item.get("signature"): if item.get("type") == "image":
item["signature"] = "..." source = item.get("source")
if isinstance(source, dict):
source["data"] = "..."
if item.get("type") == "thinking" and item.get("signature"):
item["signature"] = "..."
messages_for_logging.append(msg) messages_for_logging.append(msg)
return messages_for_logging return messages_for_logging