add msg-to-anthropic-msg converter for streamedresponse

src/magentic/chat_model/anthropic_chat_model.py

@@ -6,6 +6,7 @@
 from typing import Any, Generic, TypeVar, cast, overload
 
 from magentic._parsing import contains_parallel_function_call_type, contains_string_type
+from magentic._streamed_response import StreamedResponse
 from magentic.chat_model.base import ChatModel, aparse_stream, parse_stream
 from magentic.chat_model.function_schema import (
     BaseFunctionSchema,
@@ -34,6 +35,7 @@
     StreamState,
 )
 from magentic.function_call import FunctionCall, ParallelFunctionCall, _create_unique_id
+from magentic.streaming import StreamedStr
 from magentic.vision import UserImageMessage
 
 try:
@@ -48,6 +50,7 @@
         ToolChoiceParam,
         ToolChoiceToolParam,
         ToolParam,
+        ToolUseBlockParam,
     )
 except ImportError as error:
     msg = "To use AnthropicChatModel you must install the `anthropic` package using `pip install 'magentic[anthropic]'`."
@@ -177,7 +180,26 @@ def _(message: AssistantMessage[Any]) -> MessageParam:
             ],
         }
 
-    # TODO: Add support for StreamedResponse here
+    if isinstance(message.content, StreamedResponse):
+        content_blocks: list[TextBlockParam | ToolUseBlockParam] = []
+        for item in message.content:
+            if isinstance(item, StreamedStr):
+                content_blocks.append({"type": "text", "text": str(item)})
+            elif isinstance(item, FunctionCall):
+                function_schema = FunctionCallFunctionSchema(item.function)
+                content_blocks.append(
+                    {
+                        "type": "tool_use",
+                        "id": item._unique_id,
+                        "name": function_schema.name,
+                        "input": json.loads(function_schema.serialize_args(item)),
+                    }
+                )
+
+        return {
+            "role": AnthropicMessageRole.ASSISTANT.value,
+            "content": content_blocks,
+        }
 
     function_schema = function_schema_for_type(type(message.content))
     return {

tests/chat_model/test_anthropic_chat_model.py

@@ -1,4 +1,5 @@
 from typing import Annotated
+from unittest.mock import ANY
 
 import pytest
 from inline_snapshot import snapshot
@@ -11,7 +12,9 @@
 )
 from magentic.chat_model.base import ToolSchemaParseError
 from magentic.chat_model.message import (
+    AssistantMessage,
     DocumentBytes,
+    FunctionResultMessage,
     ImageBytes,
     Message,
     Usage,
@@ -25,6 +28,103 @@
 from magentic.streaming import AsyncStreamedStr, StreamedStr
 
 
+def plus(a: int, b: int) -> int:
+    return a + b
+
+
+@pytest.mark.parametrize(
+    ("message", "expected_anthropic_message"),
+    [
+        (UserMessage("Hello"), {"role": "user", "content": "Hello"}),
+        (AssistantMessage("Hello"), {"role": "assistant", "content": "Hello"}),
+        (
+            AssistantMessage(42),
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": ANY,
+                        "name": "return_int",
+                        "input": {"value": 42},
+                    }
+                ],
+            },
+        ),
+        (
+            AssistantMessage(FunctionCall(plus, 1, 2)),
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": ANY,
+                        "name": "plus",
+                        "input": {"a": 1, "b": 2},
+                    }
+                ],
+            },
+        ),
+        (
+            AssistantMessage(
+                ParallelFunctionCall(
+                    [FunctionCall(plus, 1, 2), FunctionCall(plus, 3, 4)]
+                )
+            ),
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": ANY,
+                        "name": "plus",
+                        "input": {"a": 1, "b": 2},
+                    },
+                    {
+                        "type": "tool_use",
+                        "id": ANY,
+                        "name": "plus",
+                        "input": {"a": 3, "b": 4},
+                    },
+                ],
+            },
+        ),
+        (
+            AssistantMessage(
+                StreamedResponse([StreamedStr(["Hello"]), FunctionCall(plus, 1, 2)])
+            ),
+            {
+                "role": "assistant",
+                "content": [
+                    {"type": "text", "text": "Hello"},
+                    {
+                        "type": "tool_use",
+                        "id": ANY,
+                        "name": "plus",
+                        "input": {"a": 1, "b": 2},
+                    },
+                ],
+            },
+        ),
+        (
+            FunctionResultMessage(3, FunctionCall(plus, 1, 2)),
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": ANY,
+                        "content": {"value": 3},
+                    }
+                ],
+            },
+        ),
+    ],
+)
+def test_message_to_anthropic_message(message, expected_anthropic_message):
+    assert message_to_anthropic_message(message) == expected_anthropic_message
+
+
 def test_message_to_anthropic_message_user_image_document_bytes_pdf(document_bytes_pdf):
     image_message = UserMessage([DocumentBytes(document_bytes_pdf)])
     assert message_to_anthropic_message(image_message) == snapshot(