fix: assemble Anthropic streamed tool_use blocks before replay

apps/application/flow/anthropic_tool_content.py

@@ -0,0 +1,135 @@
+# coding=utf-8
+"""Assemble Anthropic streamed tool_use blocks before they are replayed.
+
+Anthropic streams tool arguments as `input_json_delta` events. Those deltas are
+valid on the wire, but they are not legal `messages[].content` block types.
+Replaying them on the next turn yields:
+
+    Input tag 'input_json_delta' found using 'type' does not match any of the expected tags
+
+Empty `text` blocks from the same stream also fail with:
+
+    messages: text content blocks must be non-empty
+"""
+
+from __future__ import annotations
+
+import json
+
+ANTHROPIC_TOOL_STOP_REASONS = ("tool_use", "end_turn")
+_STREAM_DELTA_TYPES = {"input_json_delta"}
+
+
+def is_anthropic_tool_finish(response_metadata, chunk_position=None) -> bool:
+    """True when this chunk closes an Anthropic (or OpenAI-mapped) tool turn."""
+    meta = response_metadata or {}
+    if meta.get("finish_reason") == "tool_calls":
+        return True
+    if meta.get("stop_reason") in ANTHROPIC_TOOL_STOP_REASONS:
+        return True
+    return chunk_position == "last"
+
+
+def collect_input_json_deltas(content) -> dict:
+    """Concatenate `input_json_delta` fragments keyed by content-block index."""
+    collected = {}
+    if not isinstance(content, list):
+        return collected
+    for block in content:
+        if not isinstance(block, dict) or block.get("type") != "input_json_delta":
+            continue
+        index = block.get("index")
+        if index is None:
+            continue
+        piece = block.get("partial_json")
+        if piece is None:
+            piece = block.get("input") or ""
+        if not isinstance(piece, str):
+            piece = json.dumps(piece, ensure_ascii=False) if piece else ""
+        collected[index] = collected.get(index, "") + piece
+    return collected
+
+
+def _parse_tool_input(raw):
+    if raw is None or raw == "":
+        return None
+    if isinstance(raw, (dict, list)):
+        return raw
+    if not isinstance(raw, str):
+        return None
+    try:
+        parsed = json.loads(raw)
+    except (json.JSONDecodeError, TypeError, ValueError):
+        return None
+    return parsed if isinstance(parsed, (dict, list)) else None
+
+
+def _input_from_tool_calls(block, tool_calls):
+    block_id = block.get("id")
+    for tool_call in tool_calls or []:
+        if not isinstance(tool_call, dict):
+            continue
+        if block_id and tool_call.get("id") == block_id:
+            return _parse_tool_input(tool_call.get("args") or tool_call.get("arguments"))
+        if block.get("index") is not None and tool_call.get("index") == block.get("index"):
+            return _parse_tool_input(tool_call.get("args") or tool_call.get("arguments"))
+    return None
+
+
+def _input_from_fragments(block, fragments):
+    if not fragments:
+        return None
+    block_id = block.get("id")
+    block_index = block.get("index")
+    for entry in fragments.values():
+        if not isinstance(entry, dict):
+            continue
+        if block_id and entry.get("id") == block_id:
+            return _parse_tool_input(entry.get("arguments"))
+        if block_index is not None and entry.get("index") == block_index:
+            return _parse_tool_input(entry.get("arguments"))
+    return None
+
+
+def _is_empty_text_block(block) -> bool:
+    if not isinstance(block, dict):
+        return False
+    if block.get("type") not in ("text", "text_delta"):
+        return False
+    return not str(block.get("text") or "").strip()
+
+
+def finalize_anthropic_assistant_content(content, tool_calls=None, fragments=None):
+    """Drop streamed deltas and fill completed `tool_use.input` values.
+
+    Non-list content (plain strings) is returned unchanged.
+    """
+    if not isinstance(content, list):
+        return content
+
+    json_by_index = collect_input_json_deltas(content)
+    finalized = []
+    for block in content:
+        if not isinstance(block, dict):
+            finalized.append(block)
+            continue
+        if block.get("type") in _STREAM_DELTA_TYPES:
+            continue
+        if _is_empty_text_block(block):
+            continue
+        if block.get("type") == "tool_use":
+            new_block = dict(block)
+            filled = (
+                _input_from_fragments(new_block, fragments)
+                or _input_from_tool_calls(new_block, tool_calls)
+                or _parse_tool_input(json_by_index.get(new_block.get("index")))
+            )
+            current = new_block.get("input")
+            if filled is not None and (current in (None, "", {}, []) or not current):
+                new_block["input"] = filled
+            elif current == "":
+                new_block["input"] = {}
+            finalized.append(new_block)
+            continue
+        finalized.append(block)
+    return finalized

apps/application/flow/tests/test_anthropic_tool_content.py

@@ -0,0 +1,64 @@
+import sys
+import unittest
+from pathlib import Path
+
+# Allow `python -m unittest` without Django by putting `apps/` on sys.path.
+sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
+
+from application.flow.anthropic_tool_content import (
+    collect_input_json_deltas,
+    finalize_anthropic_assistant_content,
+    is_anthropic_tool_finish,
+)
+
+
+class AnthropicToolContentTests(unittest.TestCase):
+    def test_finish_detects_anthropic_stop_reason(self):
+        self.assertTrue(is_anthropic_tool_finish({"stop_reason": "tool_use"}))
+        self.assertTrue(is_anthropic_tool_finish({"finish_reason": "tool_calls"}))
+        self.assertTrue(is_anthropic_tool_finish({}, chunk_position="last"))
+        self.assertFalse(is_anthropic_tool_finish({"stop_reason": "pause_turn"}))
+
+    def test_strips_input_json_delta_and_empty_text(self):
+        content = [
+            {"type": "text", "text": ""},
+            {"type": "tool_use", "id": "toolu_1", "name": "live_price", "input": {}, "index": 1},
+            {"type": "input_json_delta", "index": 1, "partial_json": '{"query":'},
+            {"type": "input_json_delta", "index": 1, "partial_json": ' "DJI Neo"}'},
+        ]
+        finalized = finalize_anthropic_assistant_content(content)
+        self.assertEqual(len(finalized), 1)
+        self.assertEqual(finalized[0]["type"], "tool_use")
+        self.assertEqual(finalized[0]["id"], "toolu_1")
+        self.assertEqual(finalized[0]["input"], {"query": "DJI Neo"})
+        self.assertNotIn("input_json_delta", [b.get("type") for b in finalized])
+
+    def test_fills_tool_use_input_from_tool_calls(self):
+        content = [
+            {"type": "tool_use", "id": "toolu_2", "name": "live_price", "input": ""},
+        ]
+        tool_calls = [{"id": "toolu_2", "name": "live_price", "args": {"query": "in stock"}}]
+        finalized = finalize_anthropic_assistant_content(content, tool_calls=tool_calls)
+        self.assertEqual(finalized[0]["input"], {"query": "in stock"})
+
+    def test_fills_tool_use_input_from_fragments(self):
+        content = [
+            {"type": "tool_use", "id": "toolu_3", "name": "live_price", "input": {}},
+        ]
+        fragments = {"1": {"id": "toolu_3", "name": "live_price", "arguments": '{"query": "GEL"}'}}
+        finalized = finalize_anthropic_assistant_content(content, fragments=fragments)
+        self.assertEqual(finalized[0]["input"], {"query": "GEL"})
+
+    def test_collect_input_json_deltas_concatenates_partial_json(self):
+        content = [
+            {"type": "input_json_delta", "index": 0, "partial_json": '{"a":'},
+            {"type": "input_json_delta", "index": 0, "input": " 1}"},
+        ]
+        self.assertEqual(collect_input_json_deltas(content), {0: '{"a": 1}'})
+
+    def test_plain_string_content_is_unchanged(self):
+        self.assertEqual(finalize_anthropic_assistant_content("hello"), "hello")
+
+
+if __name__ == "__main__":
+    unittest.main()

apps/application/flow/tools.py

@@ -46,6 +46,11 @@
 from langchain_core.tools import StructuredTool
 from langchain_core.utils._merge import merge_lists as _original_merge_lists
 from langchain_mcp_adapters.client import MultiServerMCPClient
+from .anthropic_tool_content import (
+    collect_input_json_deltas,
+    finalize_anthropic_assistant_content,
+    is_anthropic_tool_finish,
+)
 from langgraph.checkpoint.memory import MemorySaver
 from maxkb.const import CONFIG
 from pydantic import Field, create_model
@@ -589,11 +594,24 @@ def _upsert_fragment(key, raw_id, func_name, part_args):
                     key = _get_fragment_key(tool_call.get("index"), raw_id)
                     _upsert_fragment(key, raw_id, func_name, part_args)
 
+                # ----------------------------------------------------------------
+                # 2.1 Anthropic streams tool arguments as input_json_delta content
+                #     blocks. Fold those fragments into _tool_fragments so finish
+                #     handling can assemble a legal tool_use.input.
+                # ----------------------------------------------------------------
+                if isinstance(chunk[0].content, list):
+                    for index, partial_json in collect_input_json_deltas(chunk[0].content).items():
+                        key = _get_fragment_key(index, None)
+                        if key is not None:
+                            _upsert_fragment(key, None, None, partial_json)
+
                 # ----------------------------------------------------------------
                 # 3. 检测工具调用结束，更新 tool_calls_info
+                #    Anthropic uses stop_reason=tool_use rather than OpenAI's
+                #    finish_reason=tool_calls.
                 # ----------------------------------------------------------------
-                is_finish_chunk = (
-                    chunk[0].response_metadata.get("finish_reason") == "tool_calls" or chunk[0].chunk_position == "last"
+                is_finish_chunk = is_anthropic_tool_finish(
+                    chunk[0].response_metadata, chunk[0].chunk_position
                 )
 
                 if is_finish_chunk:
@@ -675,6 +693,14 @@ def _upsert_fragment(key, raw_id, func_name, part_args):
                         fixed_tool_calls.append(tc)
                     chunk[0].additional_kwargs["tool_calls"] = fixed_tool_calls
 
+                # Anthropic: never replay input_json_delta / empty text blocks.
+                if isinstance(chunk[0].content, list):
+                    chunk[0].content = finalize_anthropic_assistant_content(
+                        chunk[0].content,
+                        tool_calls=chunk[0].tool_calls,
+                        fragments=_tool_fragments,
+                    )
+
                 yield chunk[0]
 
             if mcp_output_enable and isinstance(chunk[0], ToolMessage):