11"""Anthropic text client (modality)."""
22
33import base64
4+ import contextlib
45from typing import Any , Unpack
56
67from celeste .artifacts import ImageArtifact
1011from celeste .providers .anthropic .messages .streaming import (
1112 AnthropicMessagesStream as _AnthropicMessagesStream ,
1213)
14+ from celeste .tools import ToolCall , ToolResult
1315from celeste .types import ImageContent , Message , TextContent , VideoContent
1416from celeste .utils import detect_mime_type
1517
@@ -30,15 +32,31 @@ class AnthropicTextStream(_AnthropicMessagesStream, TextStream):
3032 def __init__ (self , * args : Any , ** kwargs : Any ) -> None :
3133 super ().__init__ (* args , ** kwargs )
3234 self ._message_start : dict [str , Any ] | None = None
35+ self ._tool_calls : dict [int , dict [str , Any ]] = {}
3336
3437 def _parse_chunk (self , event_data : dict [str , Any ]) -> TextChunk | None :
35- """Parse one SSE event into a typed chunk (captures message_start)."""
38+ """Parse one SSE event into a typed chunk (captures message_start and tool_use )."""
3639 event_type = event_data .get ("type" )
3740 if event_type == "message_start" :
3841 message = event_data .get ("message" )
3942 if isinstance (message , dict ):
4043 self ._message_start = message
4144 return None
45+ if event_type == "content_block_start" :
46+ block = event_data .get ("content_block" , {})
47+ if block .get ("type" ) == "tool_use" :
48+ idx = event_data .get ("index" , len (self ._tool_calls ))
49+ self ._tool_calls [idx ] = {
50+ "id" : block .get ("id" , "" ),
51+ "name" : block .get ("name" , "" ),
52+ "input_json" : "" ,
53+ }
54+ elif event_type == "content_block_delta" :
55+ delta = event_data .get ("delta" , {})
56+ if delta .get ("type" ) == "input_json_delta" :
57+ idx = event_data .get ("index" , - 1 )
58+ if idx in self ._tool_calls :
59+ self ._tool_calls [idx ]["input_json" ] += delta .get ("partial_json" , "" )
4260 return super ()._parse_chunk (event_data )
4361
4462 def _aggregate_event_data (self , chunks : list [TextChunk ]) -> list [dict [str , Any ]]:
@@ -49,6 +67,21 @@ def _aggregate_event_data(self, chunks: list[TextChunk]) -> list[dict[str, Any]]
4967 events .extend (super ()._aggregate_event_data (chunks ))
5068 return events
5169
70+ def _aggregate_tool_calls (
71+ self , chunks : list [TextChunk ], raw_events : list [dict [str , Any ]]
72+ ) -> list [ToolCall ]:
73+ """Reconstruct tool calls from accumulated content_block events."""
74+ import json as _json
75+
76+ result : list [ToolCall ] = []
77+ for tc in self ._tool_calls .values ():
78+ arguments = {}
79+ if tc ["input_json" ]:
80+ with contextlib .suppress (ValueError , TypeError ):
81+ arguments = _json .loads (tc ["input_json" ])
82+ result .append (ToolCall (id = tc ["id" ], name = tc ["name" ], arguments = arguments ))
83+ return result
84+
5285
5386class AnthropicTextClient (AnthropicMessagesClient , TextClient ):
5487 """Anthropic text client."""
@@ -86,10 +119,12 @@ def _init_request(self, inputs: TextInput) -> dict[str, Any]:
86119 if inputs .messages is not None :
87120 system_blocks : list [dict [str , Any ]] = []
88121 messages : list [dict [str , Any ]] = []
122+ pending_tool_results : list [dict [str , Any ]] = []
89123
90124 for message in inputs .messages :
91125 role = message .role
92126 content = message .content
127+
93128 if role in {"system" , "developer" }:
94129 if isinstance (content , list ):
95130 for block in content :
@@ -105,8 +140,27 @@ def _init_request(self, inputs: TextInput) -> dict[str, Any]:
105140 system_blocks .append ({"type" : "text" , "text" : str (content )})
106141 continue
107142
143+ if isinstance (message , ToolResult ):
144+ pending_tool_results .append (
145+ {
146+ "type" : "tool_result" ,
147+ "tool_use_id" : message .tool_call_id ,
148+ "content" : str (content ),
149+ }
150+ )
151+ continue
152+
153+ # Flush pending tool results as a single user message
154+ if pending_tool_results :
155+ messages .append ({"role" : "user" , "content" : pending_tool_results })
156+ pending_tool_results = []
157+
108158 messages .append ({"role" : role , "content" : content })
109159
160+ # Flush remaining tool results
161+ if pending_tool_results :
162+ messages .append ({"role" : "user" , "content" : pending_tool_results })
163+
110164 request : dict [str , Any ] = {"messages" : messages }
111165 if system_blocks :
112166 request ["system" ] = system_blocks
@@ -166,6 +220,16 @@ def _parse_content(
166220
167221 return text_content
168222
223+ def _parse_tool_calls (self , response_data : dict [str , Any ]) -> list [ToolCall ]:
224+ """Parse tool calls from Anthropic response."""
225+ return [
226+ ToolCall (
227+ id = block ["id" ], name = block ["name" ], arguments = block .get ("input" , {})
228+ )
229+ for block in response_data .get ("content" , [])
230+ if block .get ("type" ) == "tool_use"
231+ ]
232+
169233 def _stream_class (self ) -> type [TextStream ]:
170234 """Return the Stream class for this provider."""
171235 return AnthropicTextStream
0 commit comments