Class ToolCallAccumulator

src/mcp_cli/chat/streaming_handler.py:86–219

Manages accumulation of tool calls across streaming chunks. Uses Pydantic models for type-safe state management. Handles fragmented JSON and validates tool call structure.

class ToolCallAccumulator:
    """Manages accumulation of tool calls across streaming chunks.

    Uses Pydantic models for type-safe state management.
    Handles fragmented JSON and validates tool call structure.
    """

    def __init__(self) -> None:
        self._accumulated: list[ToolCallData] = []

    def process_chunk_tool_calls(self, chunk_tool_calls: list[dict[str, Any]]) -> None:
        """Process tool calls from a chunk and accumulate them.

        Args:
            chunk_tool_calls: Tool calls extracted from current chunk
        """
        for tc_dict in chunk_tool_calls:
            # Convert dict to Pydantic model
            tc = ToolCallData.from_dict(tc_dict)

            # Find or create accumulator for this tool call
            existing = self._find_accumulated_call(tc.id, tc.index)

            if existing:
                # Merge chunk data using intelligent JSON merging
                self._merge_tool_call(existing, tc)
            else:
                # New tool call - add to accumulated list
                self._accumulated.append(tc)

    def _find_accumulated_call(self, tc_id: str, tc_index: int) -> ToolCallData | None:
        """Find existing accumulated tool call by ID or index."""
        for tc in self._accumulated:
            if tc.id == tc_id or tc.index == tc_index:
                return tc
        return None

    def _merge_tool_call(self, existing: ToolCallData, new: ToolCallData) -> None:
        """Merge new tool call data into existing accumulator."""
        # Update function name if provided
        if new.function.name:
            existing.function.name = new.function.name

        # Accumulate arguments using intelligent JSON merging
        if new.function.arguments:
            merged_args = self._merge_json_strings(
                existing.function.arguments, new.function.arguments
            )
            existing.function.arguments = merged_args

    def _merge_json_strings(self, current: str, new: str) -> str:
        """Merge two JSON strings intelligently.

        Tries multiple strategies:
        1. Parse both and merge dicts
        2. Concatenate and validate
        3. Fix common issues
        """