type constraints on channels, serde

Author: evalstate

src/fast_agent/llm/provider/anthropic/llm_anthropic.py

@@ -781,7 +781,20 @@ async def _anthropic_completion(
         if raw_thinking_blocks:
             if channels is None:
                 channels = {}
-            channels[ANTHROPIC_THINKING_BLOCKS] = raw_thinking_blocks
+            serialized_blocks = []
+            for block in raw_thinking_blocks:
+                try:
+                    payload = block.model_dump()
+                except Exception:
+                    payload = {"type": getattr(block, "type", "thinking")}
+                    if isinstance(block, ThinkingBlock):
+                        payload.update(
+                            {"thinking": block.thinking, "signature": block.signature}
+                        )
+                    elif isinstance(block, RedactedThinkingBlock):
+                        payload.update({"data": block.data})
+                serialized_blocks.append(TextContent(type="text", text=json.dumps(payload)))
+            channels[ANTHROPIC_THINKING_BLOCKS] = serialized_blocks
 
         return PromptMessageExtended(
             role="assistant",

src/fast_agent/llm/provider/anthropic/multipart_converter_anthropic.py

@@ -1,3 +1,4 @@
+import json
 import re
 from typing import Literal, Sequence, Union, cast
 from urllib.parse import urlparse
@@ -11,6 +12,7 @@
     MessageParam,
     PlainTextSourceParam,
     RedactedThinkingBlock,
+    RedactedThinkingBlockParam,
     TextBlockParam,
     ThinkingBlock,
     ThinkingBlockParam,
@@ -106,6 +108,28 @@ def convert_to_anthropic(multipart_msg: PromptMessageExtended) -> MessageParam:
                             # Redacted thinking blocks are passed as-is
                             # They contain encrypted data that the API can verify
                             all_content_blocks.append(thinking_block)
+                        elif isinstance(thinking_block, TextContent):
+                            try:
+                                payload = json.loads(thinking_block.text)
+                            except (TypeError, json.JSONDecodeError):
+                                payload = None
+                            if isinstance(payload, dict):
+                                block_type = payload.get("type")
+                                if block_type == "thinking":
+                                    all_content_blocks.append(
+                                        ThinkingBlockParam(
+                                            type="thinking",
+                                            thinking=payload.get("thinking", ""),
+                                            signature=payload.get("signature", ""),
+                                        )
+                                    )
+                                elif block_type == "redacted_thinking":
+                                    all_content_blocks.append(
+                                        RedactedThinkingBlockParam(
+                                            type="redacted_thinking",
+                                            data=payload.get("data", ""),
+                                        )
+                                    )
 
             for tool_use_id, req in multipart_msg.tool_calls.items():
                 sanitized_id = AnthropicConverter._sanitize_tool_id(tool_use_id)

src/fast_agent/mcp/prompt_message_extended.py

@@ -1,4 +1,4 @@
-from typing import Any, Mapping, Sequence
+from typing import Mapping, Sequence
 
 from mcp.types import (
     CallToolRequest,
@@ -27,11 +27,11 @@ class PromptMessageExtended(BaseModel):
     content: list[ContentBlock] = []
     tool_calls: dict[str, CallToolRequest] | None = None
     tool_results: dict[str, CallToolResult] | None = None
-    # Channels can carry provider-specific payloads (e.g., raw Anthropic thinking blocks).
-    channels: Mapping[str, Sequence[Any]] | None = None
+    channels: Mapping[str, Sequence[ContentBlock]] | None = None
     stop_reason: LlmStopReason | None = None
     is_template: bool = False
 
+
     @classmethod
     def to_extended(cls, messages: list[PromptMessage]) -> list["PromptMessageExtended"]:
         """Convert a sequence of PromptMessages into PromptMessageExtended objects."""

tests/unit/fast_agent/llm/providers/test_multipart_converter_anthropic.py

@@ -1,8 +1,11 @@
 import base64
+import json
 import unittest
 
 from mcp.types import (
     BlobResourceContents,
+    CallToolRequest,
+    CallToolRequestParams,
     CallToolResult,
     EmbeddedResource,
     ImageContent,
@@ -12,6 +15,7 @@
 )
 from pydantic import AnyUrl
 
+from fast_agent.constants import ANTHROPIC_THINKING_BLOCKS
 from fast_agent.llm.provider.anthropic.multipart_converter_anthropic import (
     AnthropicConverter,
 )
@@ -715,6 +719,42 @@ def test_assistant_multiple_text_blocks(self):
         self.assertEqual(anthropic_msg["content"][1]["type"], "text")
         self.assertEqual(anthropic_msg["content"][1]["text"], "Second part of response")
 
+    def test_assistant_thinking_blocks_deserialized_from_channel(self):
+        """Ensure thinking channel JSON is converted to Anthropic thinking params."""
+        thinking_payload = {
+            "type": "thinking",
+            "thinking": "Reasoning summary.",
+            "signature": "sig123",
+        }
+        redacted_payload = {"type": "redacted_thinking", "data": "opaque"}
+        channels = {
+            ANTHROPIC_THINKING_BLOCKS: [
+                TextContent(type="text", text=json.dumps(thinking_payload)),
+                TextContent(type="text", text=json.dumps(redacted_payload)),
+            ]
+        }
+        tool_calls = {
+            "toolu_1": CallToolRequest(
+                method="tools/call",
+                params=CallToolRequestParams(name="test_tool", arguments={"x": 1}),
+            )
+        }
+        multipart = PromptMessageExtended(
+            role="assistant", content=[], tool_calls=tool_calls, channels=channels
+        )
+
+        anthropic_msg = AnthropicConverter.convert_to_anthropic(multipart)
+
+        self.assertEqual(anthropic_msg["role"], "assistant")
+        self.assertEqual(len(anthropic_msg["content"]), 3)
+        self.assertEqual(anthropic_msg["content"][0]["type"], "thinking")
+        self.assertEqual(anthropic_msg["content"][0]["thinking"], "Reasoning summary.")
+        self.assertEqual(anthropic_msg["content"][0]["signature"], "sig123")
+        self.assertEqual(anthropic_msg["content"][1]["type"], "redacted_thinking")
+        self.assertEqual(anthropic_msg["content"][1]["data"], "opaque")
+        self.assertEqual(anthropic_msg["content"][2]["type"], "tool_use")
+        self.assertEqual(anthropic_msg["content"][2]["name"], "test_tool")
+
     def test_assistant_non_text_content_stripped(self):
         """Test that non-text content is stripped from assistant messages."""
         # Create a mixed content message with text and image