codekingpro/portable-devtools

1"""Message processing and content serialization for Claude Agent SDK."""2 3from typing import Any4 5 6def _extract_tool_result_text(content: Any) -> str:7    """Extract text content from tool result content blocks."""8    if content is None:9        return ""10    if isinstance(content, str):11        return content12    if isinstance(content, list):13        texts = []14        for item in content:15            if isinstance(item, dict):16                if item.get("type") == "text":17                    texts.append(item.get("text", ""))18            elif hasattr(item, "text"):19                texts.append(getattr(item, "text", ""))20        return "\n".join(texts) if texts else str(content)21    return str(content)22 23 24def flatten_content_blocks(content: Any) -> Any:25    """Convert SDK content blocks into serializable dicts using explicit type checks."""26    if not isinstance(content, list):27        return content28 29    result = []30    for block in content:31        block_type = type(block).__name__32 33        # Handle known Claude SDK block types34        if block_type == "TextBlock":35            result.append(36                {37                    "type": "text",38                    "text": getattr(block, "text", ""),39                }40            )41        elif block_type == "ThinkingBlock":42            result.append(43                {44                    "type": "thinking",45                    "thinking": getattr(block, "thinking", ""),46                    "signature": getattr(block, "signature", ""),47                }48            )49        elif block_type == "ToolUseBlock":50            result.append(51                {52                    "type": "tool_use",53                    "id": getattr(block, "id", None),54                    "name": getattr(block, "name", None),55                    "input": getattr(block, "input", None),56                }57            )58        elif block_type == "ToolResultBlock":59            # Extract text from nested content for tool results60            tool_content = getattr(block, "content", None)61            content_text = _extract_tool_result_text(tool_content)62            result.append(63                {64                    "type": "tool_result",65                    "tool_use_id": getattr(block, "tool_use_id", None),66                    "content": content_text,67                    "is_error": getattr(block, "is_error", False),68                }69            )70        else:71            result.append(block)72    return result73 74 75def unwrap_message_dicts(messages: list[Any]) -> list[dict[str, Any]]:76    """Normalize SDK message dicts into ``{role, content}`` form.77 78    The Claude SDK wraps messages in ``{"message": {"role": ..., "content": ...}}``79    envelopes.  This function unwraps them into a flat list.80    """81    result: list[dict[str, Any]] = []82    for msg in messages:83        if not isinstance(msg, dict):84            result.append(msg)85            continue86        if "message" in msg:87            inner = msg["message"]88            if isinstance(inner, dict):89                result.append(90                    {91                        "role": inner.get("role", "user"),92                        "content": inner.get("content", ""),93                    }94                )95            else:96                result.append(msg)97        else:98            result.append(msg)99    return result100 101 102def build_llm_input(prompt: Any, history: list[dict[str, Any]]) -> list[dict[str, Any]]:103    """Construct a combined prompt + history message list."""104    if isinstance(prompt, str):105        entry = {"content": prompt, "role": "user"}106        return [entry, *history] if history else [entry]107 108    if isinstance(prompt, list):109        formatted = unwrap_message_dicts(prompt)110        return [*formatted, *history] if history else formatted111 112    return list(history) if history else []113