# -*- coding: utf-8 -*-
"""Comprehensive formatter unit tests for AnthropicChatFormatter and
AnthropicMultiAgentFormatter, following the reference test style with exact
ground-truth comparisons.
"""
from unittest import IsolatedAsyncioTestCase
from agentscope.formatter import (
AnthropicChatFormatter,
AnthropicMultiAgentFormatter,
)
from agentscope.message import (
UserMsg,
AssistantMsg,
SystemMsg,
TextBlock,
DataBlock,
ToolCallBlock,
ToolResultBlock,
ToolResultState,
Base64Source,
ThinkingBlock,
HintBlock,
)
class TestAnthropicFormatter(IsolatedAsyncioTestCase):
"""Comprehensive tests for Anthropic Chat and MultiAgent formatters."""
async def asyncSetUp(self) -> None:
"""Set up shared message fixtures and expected ground-truth dicts."""
self.image_b64 = "ZmFrZSBpbWFnZSBkYXRh"
# ---------------------------------------------------------------
# Message fixtures
# (No URL images: Anthropic URL handling downloads from the network)
# ---------------------------------------------------------------
self.msgs_system = [
SystemMsg(
name="system",
content="You're a helpful assistant.",
),
]
self.msgs_conversation = [
UserMsg(
name="user",
content="What is the capital of France?",
),
AssistantMsg(
name="assistant",
content="The capital of France is Paris.",
),
UserMsg(
name="user",
content="What is the capital of Germany?",
),
AssistantMsg(
name="assistant",
content="The capital of Germany is Berlin.",
),
UserMsg(
name="user",
content="What is the capital of Japan?",
),
]
self.msgs_tools = [
AssistantMsg(
name="assistant",
content=[
ToolCallBlock(
id="call_1",
name="get_capital",
input='{"country": "Japan"}',
),
ToolResultBlock(
id="call_1",
name="get_capital",
output=[
TextBlock(text="The capital of Japan is Tokyo."),
],
state=ToolResultState.SUCCESS,
),
TextBlock(text="The capital of Japan is Tokyo."),
],
),
]
# ---------------------------------------------------------------
# Ground truth: AnthropicChatFormatter
# - No "name" field.
# - Content is always a list of {"type": ..., ...} dicts.
# - ToolResultBlock forces role to "user".
# - ToolCallBlock "input" is a dict (parsed from JSON string).
# ---------------------------------------------------------------
self.gt_chat = [
{
"role": "system",
"content": [
{"type": "text", "text": "You're a helpful assistant."},
],
},
{
"role": "user",
"content": [
{
"type": "text",
"text": "What is the capital of France?",
},
],
},
{
"role": "assistant",
"content": [
{
"type": "text",
"text": "The capital of France is Paris.",
},
],
},
{
"role": "user",
"content": [
{
"type": "text",
"text": "What is the capital of Germany?",
},
],
},
{
"role": "assistant",
"content": [
{
"type": "text",
"text": "The capital of Germany is Berlin.",
},
],
},
{
"role": "user",
"content": [
{
"type": "text",
"text": "What is the capital of Japan?",
},
],
},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_1",
"name": "get_capital",
"input": {"country": "Japan"},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_1",
"content": [
{
"type": "text",
"text": "The capital of Japan is Tokyo.",
},
],
},
],
},
{
"role": "assistant",
"content": [
{
"type": "text",
"text": "The capital of Japan is Tokyo.",
},
],
},
]
# ---------------------------------------------------------------
# Ground truth: AnthropicMultiAgentFormatter
# - System: {"role": "system", "content": [{"type": "text", ...}]}
# - Agent messages (is_first=True): wrapped in hist_prompt +
# ....
# - Agent messages (is_first=False): no wrapping at all.
# ---------------------------------------------------------------
_hist_prompt = (
AnthropicMultiAgentFormatter().conversation_history_prompt
)
_conv_text = (
"user: What is the capital of France?\n"
"assistant: The capital of France is Paris.\n"
"user: What is the capital of Germany?\n"
"assistant: The capital of Germany is Berlin.\n"
"user: What is the capital of Japan?"
)
self._gt_trailing_asst = {
"role": "assistant",
"content": [
{
"type": "text",
"text": "The capital of Japan is Tokyo.",
},
],
}
self._gt_tool_call = {
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_1",
"name": "get_capital",
"input": {"country": "Japan"},
},
],
}
self._gt_tool_result = {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_1",
"content": [
{
"type": "text",
"text": "The capital of Japan is Tokyo.",
},
],
},
],
}
self.gt_multiagent = [
{
"role": "system",
"content": [
{"type": "text", "text": "You're a helpful assistant."},
],
},
{
"role": "user",
"content": [
{
"type": "text",
"text": (
_hist_prompt
+ "\n"
+ _conv_text
+ "\n"
),
},
],
},
self._gt_tool_call,
self._gt_tool_result,
self._gt_trailing_asst,
]
# -------------------------------------------------------------------
# AnthropicChatFormatter tests
# -------------------------------------------------------------------
async def test_chat_formatter(self) -> None:
"""Chat formatter produces exact output for various subsets."""
fmt = AnthropicChatFormatter()
# Full history
res = await fmt.format(
[*self.msgs_system, *self.msgs_conversation, *self.msgs_tools],
)
self.assertListEqual(self.gt_chat, res)
# Without system
res = await fmt.format([*self.msgs_conversation, *self.msgs_tools])
self.assertListEqual(self.gt_chat[1:], res)
# Without conversation
n_tools_gt = len(self.gt_chat) - 1 - len(self.msgs_conversation)
res = await fmt.format([*self.msgs_system, *self.msgs_tools])
self.assertListEqual(
[self.gt_chat[0]] + self.gt_chat[-n_tools_gt:],
res,
)
# Without tools
res = await fmt.format([*self.msgs_system, *self.msgs_conversation])
self.assertListEqual(self.gt_chat[:-n_tools_gt], res)
# Empty
res = await fmt.format([])
self.assertListEqual([], res)
async def test_chat_formatter_base64_image(self) -> None:
"""Base64-encoded image is formatted as Anthropic image source."""
fmt = AnthropicChatFormatter()
msgs = [
UserMsg(
name="user",
content=[
TextBlock(text="What's in this image?"),
DataBlock(
source=Base64Source(
data=self.image_b64,
media_type="image/png",
),
),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "user",
"content": [
{"type": "text", "text": "What's in this image?"},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": self.image_b64,
},
},
],
},
],
res,
)
async def test_chat_formatter_thinking_preserved(self) -> None:
"""ThinkingBlock with a signature is passed back as a thinking
content block."""
fmt = AnthropicChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
ThinkingBlock(
thinking="inner thoughts",
signature="sig_abc",
),
TextBlock(text="reply"),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": [
{
"type": "thinking",
"thinking": "inner thoughts",
"signature": "sig_abc",
},
{"type": "text", "text": "reply"},
],
},
],
res,
)
async def test_chat_formatter_thinking_without_signature_dropped(
self,
) -> None:
"""ThinkingBlock from another provider (no signature) is dropped
rather than forwarded with an empty signature, which Anthropic
rejects with `Invalid signature in thinking block`. An empty-string
signature is treated the same as a missing one."""
fmt = AnthropicChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
ThinkingBlock(thinking="from another provider"),
ThinkingBlock(thinking="empty sig", signature=""),
TextBlock(text="reply"),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": [
{"type": "text", "text": "reply"},
],
},
],
res,
)
# Thinking-only message produces no output (no empty assistant msg).
res = await fmt.format(
[
AssistantMsg(
name="assistant",
content=[ThinkingBlock(thinking="only thinking")],
),
],
)
self.assertListEqual([], res)
async def test_chat_formatter_tool_result_role_forced_to_user(
self,
) -> None:
"""Anthropic forces tool_result messages to role='user'."""
fmt = AnthropicChatFormatter()
res = await fmt.format(
[*self.msgs_system, *self.msgs_conversation, *self.msgs_tools],
)
tool_result_roles = [
m["role"]
for m in res
if any(
b.get("type") == "tool_result"
for b in (m.get("content") or [])
)
]
self.assertListEqual(tool_result_roles, ["user"])
async def test_chat_formatter_tool_result_with_image(self) -> None:
"""Tool result containing an image DataBlock inlines the image in the
tool_result content without crashing on TextBlock system-reminders."""
fmt = AnthropicChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
ToolCallBlock(
id="call_img",
name="get_chart",
input="{}",
),
ToolResultBlock(
id="call_img",
name="get_chart",
output=[
TextBlock(text="Here is the chart."),
DataBlock(
source=Base64Source(
data=self.image_b64,
media_type="image/png",
),
),
],
state=ToolResultState.SUCCESS,
),
TextBlock(text="Here is the chart analysis."),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_img",
"name": "get_chart",
"input": {},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_img",
"content": [
{
"type": "text",
"text": "Here is the chart.",
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": self.image_b64,
},
},
],
},
],
},
{
"role": "assistant",
"content": [
{
"type": "text",
"text": "Here is the chart analysis.",
},
],
},
],
res,
)
# -------------------------------------------------------------------
# AnthropicMultiAgentFormatter tests
# -------------------------------------------------------------------
async def test_multiagent_formatter(self) -> None:
"""MultiAgent formatter produces exact output for various subsets."""
fmt = AnthropicMultiAgentFormatter()
# Full history
res = await fmt.format(
[*self.msgs_system, *self.msgs_conversation, *self.msgs_tools],
)
self.assertListEqual(self.gt_multiagent, res)
# Without system
res = await fmt.format([*self.msgs_conversation, *self.msgs_tools])
self.assertListEqual(self.gt_multiagent[1:], res)
# Without tools
res = await fmt.format([*self.msgs_system, *self.msgs_conversation])
self.assertListEqual(self.gt_multiagent[:2], res)
# System only
res = await fmt.format(self.msgs_system)
self.assertListEqual([self.gt_multiagent[0]], res)
# Conversation only
res = await fmt.format(self.msgs_conversation)
self.assertListEqual([self.gt_multiagent[1]], res)
# Tools only
res = await fmt.format(self.msgs_tools)
self.assertListEqual(
[
self._gt_tool_call,
self._gt_tool_result,
self._gt_trailing_asst,
],
res,
)
# System + tools (no conversation)
res = await fmt.format([*self.msgs_system, *self.msgs_tools])
self.assertListEqual(
[
self.gt_multiagent[0],
self._gt_tool_call,
self._gt_tool_result,
self._gt_trailing_asst,
],
res,
)
# Empty
res = await fmt.format([])
self.assertListEqual([], res)
async def test_chat_formatter_complex_multi_step(self) -> None:
"""Complex multi-step sequence with interleaved thinking, text,
tool calls, and tool results."""
fmt = AnthropicChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
ThinkingBlock(thinking="thinking_1", signature="sig_1"),
TextBlock(text="text_1"),
ToolCallBlock(
id="call_1",
name="func_1",
input='{"arg": "value1"}',
),
ToolCallBlock(
id="call_2",
name="func_2",
input='{"arg": "value2"}',
),
ToolResultBlock(
id="call_1",
name="func_1",
output=[TextBlock(text="result_1")],
state=ToolResultState.SUCCESS,
),
ToolResultBlock(
id="call_2",
name="func_2",
output=[TextBlock(text="result_2")],
state=ToolResultState.SUCCESS,
),
ThinkingBlock(thinking="thinking_2", signature="sig_2"),
TextBlock(text="text_2"),
ToolCallBlock(
id="call_3",
name="func_3",
input='{"arg": "value3"}',
),
ToolResultBlock(
id="call_3",
name="func_3",
output=[TextBlock(text="result_3")],
state=ToolResultState.SUCCESS,
),
ToolCallBlock(
id="call_4",
name="func_4",
input='{"arg": "value4"}',
),
ToolResultBlock(
id="call_4",
name="func_4",
output=[TextBlock(text="result_4")],
state=ToolResultState.SUCCESS,
),
ThinkingBlock(thinking="thinking_3", signature="sig_3"),
TextBlock(text="text_3"),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": [
{
"type": "thinking",
"thinking": "thinking_1",
"signature": "sig_1",
},
{"type": "text", "text": "text_1"},
{
"type": "tool_use",
"id": "call_1",
"name": "func_1",
"input": {"arg": "value1"},
},
{
"type": "tool_use",
"id": "call_2",
"name": "func_2",
"input": {"arg": "value2"},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_1",
"content": [
{"type": "text", "text": "result_1"},
],
},
{
"type": "tool_result",
"tool_use_id": "call_2",
"content": [
{"type": "text", "text": "result_2"},
],
},
],
},
{
"role": "assistant",
"content": [
{
"type": "thinking",
"thinking": "thinking_2",
"signature": "sig_2",
},
{"type": "text", "text": "text_2"},
{
"type": "tool_use",
"id": "call_3",
"name": "func_3",
"input": {"arg": "value3"},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_3",
"content": [
{"type": "text", "text": "result_3"},
],
},
],
},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_4",
"name": "func_4",
"input": {"arg": "value4"},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_4",
"content": [
{"type": "text", "text": "result_4"},
],
},
],
},
{
"role": "assistant",
"content": [
{
"type": "thinking",
"thinking": "thinking_3",
"signature": "sig_3",
},
{"type": "text", "text": "text_3"},
],
},
],
res,
)
async def test_chat_formatter_hint_block(self) -> None:
"""HintBlock flushes preceding content and becomes a user message."""
fmt = AnthropicChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
TextBlock(text="Let me think about that."),
HintBlock(hint="Remember to be concise."),
TextBlock(text="Here is my answer."),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "assistant",
"content": [
{"type": "text", "text": "Let me think about that."},
],
},
{
"role": "user",
"content": [
{"type": "text", "text": "Remember to be concise."},
],
},
{
"role": "assistant",
"content": [
{"type": "text", "text": "Here is my answer."},
],
},
],
res,
)
async def test_chat_formatter_hint_block_multimodal(self) -> None:
"""Multimodal HintBlock becomes a single user message with text
+ image."""
fmt = AnthropicChatFormatter()
msgs = [
AssistantMsg(
name="assistant",
content=[
HintBlock(
hint=[
TextBlock(text="Inspect this screenshot:"),
DataBlock(
source=Base64Source(
data=self.image_b64,
media_type="image/png",
),
),
],
),
],
),
]
res = await fmt.format(msgs)
self.assertListEqual(
[
{
"role": "user",
"content": [
{
"type": "text",
"text": "Inspect this screenshot:",
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": self.image_b64,
},
},
],
},
],
res,
)
async def test_chat_formatter_parallel_tool_results_merged(
self,
) -> None:
"""Parallel tool_results must be grouped into ONE user message.
Regression test for Issue #1892: AnthropicChatFormatter was flushing
content_blocks on every ToolResultBlock, which split N parallel results
into N separate user messages. Strict endpoints such as DeepSeek's
Anthropic-compatible API reject this with HTTP 400.
"""
fmt = AnthropicChatFormatter()
msgs = [
UserMsg(
name="user",
content="Get weather for Beijing, Shanghai, and Guangzhou.",
),
AssistantMsg(
name="assistant",
content=[
TextBlock(text="Let me check all three cities."),
ToolCallBlock(
id="call_01",
name="get_weather",
input='{"city": "Beijing"}',
),
ToolCallBlock(
id="call_02",
name="get_weather",
input='{"city": "Shanghai"}',
),
ToolCallBlock(
id="call_03",
name="get_weather",
input='{"city": "Guangzhou"}',
),
ToolResultBlock(
id="call_01",
name="get_weather",
output="Sunny, 28\u00b0C",
state=ToolResultState.SUCCESS,
),
ToolResultBlock(
id="call_02",
name="get_weather",
output="Cloudy, 24\u00b0C",
state=ToolResultState.SUCCESS,
),
ToolResultBlock(
id="call_03",
name="get_weather",
output="Rainy, 22\u00b0C",
state=ToolResultState.SUCCESS,
),
],
),
]
res = await fmt.format(msgs)
# Expected: 3 messages total
# [0] user -> original question
# [1] assistant -> text + 3x tool_use
# [2] user -> ALL 3 tool_results in ONE message (the fix)
self.assertListEqual(
[
{
"role": "user",
"content": [
{
"type": "text",
"text": "Get weather for Beijing, Shanghai, "
"and Guangzhou.",
},
],
},
{
"role": "assistant",
"content": [
{
"type": "text",
"text": "Let me check all three cities.",
},
{
"type": "tool_use",
"id": "call_01",
"name": "get_weather",
"input": {"city": "Beijing"},
},
{
"type": "tool_use",
"id": "call_02",
"name": "get_weather",
"input": {"city": "Shanghai"},
},
{
"type": "tool_use",
"id": "call_03",
"name": "get_weather",
"input": {"city": "Guangzhou"},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_01",
"content": [
{"type": "text", "text": "Sunny, 28\u00b0C"},
],
},
{
"type": "tool_result",
"tool_use_id": "call_02",
"content": [
{"type": "text", "text": "Cloudy, 24\u00b0C"},
],
},
{
"type": "tool_result",
"tool_use_id": "call_03",
"content": [
{"type": "text", "text": "Rainy, 22\u00b0C"},
],
},
],
},
],
res,
)