1use super::*;
2
3pub struct AnthropicProvider {
4 http: reqwest::Client,
5 key_pool: KeyPool,
6 model: String,
7 max_tokens: u32,
8 base_url: String,
9 prompt_cache_enabled: bool,
10 prompt_cache_ttl: String,
11}
12
13impl AnthropicProvider {
14 pub fn new(config: &Config) -> Self {
15 AnthropicProvider {
16 http: reqwest::Client::new(),
17 key_pool: KeyPool::new(&config.api_key, &config.api_keys),
18 model: config.model.clone(),
19 max_tokens: config.max_tokens,
20 base_url: resolve_anthropic_messages_url(config.llm_base_url.as_deref().unwrap_or("")),
21 prompt_cache_enabled: config.anthropic_prompt_cache_enabled,
22 prompt_cache_ttl: config.anthropic_prompt_cache_ttl.clone(),
23 }
24 }
25
26 fn build_request_body(
29 &self,
30 request: &MessagesRequest,
31 ) -> Result<serde_json::Value, MicroClawError> {
32 let mut body = serde_json::to_value(request).map_err(|e| {
33 MicroClawError::LlmApi(format!("failed to serialize Anthropic request: {e}"))
34 })?;
35 if self.prompt_cache_enabled {
36 crate::prompt_cache::apply_anthropic_prompt_cache(&mut body, &self.prompt_cache_ttl);
37 }
38 Ok(body)
39 }
40
41 async fn send_message_stream_single_pass(
42 &self,
43 request: &MessagesRequest,
44 text_tx: Option<&UnboundedSender<String>>,
45 ) -> Result<MessagesResponse, MicroClawError> {
46 let mut streamed_request = request.clone();
47 streamed_request.stream = Some(true);
48
49 debug!(
50 provider = "anthropic",
51 model = %request.model,
52 url = %self.base_url,
53 messages_count = request.messages.len(),
54 "Sending LLM stream request"
55 );
56
57 let body = self.build_request_body(&streamed_request)?;
58 let response = self
59 .http
60 .post(&self.base_url)
61 .header("x-api-key", self.key_pool.current())
62 .header("anthropic-version", "2023-06-01")
63 .header("content-type", "application/json")
64 .json(&body)
65 .send()
66 .await?;
67
68 let status = response.status();
69 if !status.is_success() {
70 let body = response.text().await.unwrap_or_default();
71 if let Ok(api_err) = serde_json::from_str::<AnthropicApiError>(&body) {
72 return Err(MicroClawError::LlmApi(format!(
73 "{}: {}",
74 api_err.error.error_type, api_err.error.message
75 )));
76 }
77 return Err(MicroClawError::LlmApi(format!("HTTP {status}: {body}")));
78 }
79
80 let mut byte_stream = response.bytes_stream();
81 let mut sse = SseEventParser::default();
82 let mut stop_reason: Option<String> = None;
83 let mut usage: Option<Usage> = None;
84 let mut text_blocks: std::collections::HashMap<usize, String> =
85 std::collections::HashMap::new();
86 let mut tool_blocks: std::collections::HashMap<usize, StreamToolUseBlock> =
87 std::collections::HashMap::new();
88 let mut ordered_indexes: Vec<usize> = Vec::new();
89
90 'outer: while let Some(chunk_res) = byte_stream.next().await {
91 let chunk = match chunk_res {
92 Ok(c) => c,
93 Err(_) => break,
94 };
95 for data in sse.push_chunk(chunk.as_ref()) {
96 if data == "[DONE]" {
97 break 'outer;
98 }
99 process_anthropic_stream_event(
100 &data,
101 text_tx,
102 &mut stop_reason,
103 &mut usage,
104 &mut text_blocks,
105 &mut tool_blocks,
106 &mut ordered_indexes,
107 );
108 }
109 }
110 for data in sse.finish() {
111 if data == "[DONE]" {
112 break;
113 }
114 process_anthropic_stream_event(
115 &data,
116 text_tx,
117 &mut stop_reason,
118 &mut usage,
119 &mut text_blocks,
120 &mut tool_blocks,
121 &mut ordered_indexes,
122 );
123 }
124
125 Ok(build_stream_response(
126 ordered_indexes,
127 text_blocks,
128 tool_blocks,
129 stop_reason,
130 usage,
131 ))
132 }
133}
134
135pub(crate) fn resolve_anthropic_messages_url(configured_base: &str) -> String {
136 let trimmed = configured_base.trim().trim_end_matches('/').to_string();
137 if trimmed.is_empty() {
138 return "https://api.anthropic.com/v1/messages".to_string();
139 }
140 if trimmed.ends_with("/v1/messages") {
141 return trimmed;
142 }
143 format!("{trimmed}/v1/messages")
144}
145
146pub(crate) fn process_anthropic_stream_event(
147 data: &str,
148 text_tx: Option<&UnboundedSender<String>>,
149 stop_reason: &mut Option<String>,
150 usage: &mut Option<Usage>,
151 text_blocks: &mut std::collections::HashMap<usize, String>,
152 tool_blocks: &mut std::collections::HashMap<usize, StreamToolUseBlock>,
153 ordered_indexes: &mut Vec<usize>,
154) {
155 let Ok(v) = serde_json::from_str::<serde_json::Value>(data) else {
156 return;
157 };
158
159 let event_type = v.get("type").and_then(|t| t.as_str()).unwrap_or_default();
160 match event_type {
161 "content_block_start" => {
162 if let Some(index) = v
163 .get("index")
164 .and_then(|i| i.as_u64())
165 .and_then(|i| usize::try_from(i).ok())
166 {
167 if !ordered_indexes.contains(&index) {
168 ordered_indexes.push(index);
169 }
170 if let Some(block) = v.get("content_block") {
171 match block.get("type").and_then(|t| t.as_str()) {
172 Some("text") => {
173 let text = block
174 .get("text")
175 .and_then(|t| t.as_str())
176 .unwrap_or_default()
177 .to_string();
178 text_blocks.insert(index, text);
179 }
180 Some("tool_use") => {
181 let id = block
182 .get("id")
183 .and_then(|s| s.as_str())
184 .unwrap_or_default()
185 .to_string();
186 let name = block
187 .get("name")
188 .and_then(|s| s.as_str())
189 .unwrap_or_default()
190 .to_string();
191 let input = block.get("input").cloned().unwrap_or_else(|| json!({}));
192 let input_json = if input.is_object()
193 && input.as_object().is_some_and(|m| m.is_empty())
194 {
195 String::new()
196 } else {
197 serde_json::to_string(&input).unwrap_or_else(|_| "{}".to_string())
198 };
199 tool_blocks.insert(
200 index,
201 StreamToolUseBlock {
202 id,
203 name,
204 input_json,
205 thought_signature: None,
206 },
207 );
208 }
209 _ => {}
210 }
211 }
212 }
213 }
214 "content_block_delta" => {
215 let Some(index) = v
216 .get("index")
217 .and_then(|i| i.as_u64())
218 .and_then(|i| usize::try_from(i).ok())
219 else {
220 return;
221 };
222 let Some(delta) = v.get("delta") else {
223 return;
224 };
225 match delta.get("type").and_then(|t| t.as_str()) {
226 Some("text_delta") => {
227 let piece = delta
228 .get("text")
229 .and_then(|t| t.as_str())
230 .unwrap_or_default();
231 if !piece.is_empty() {
232 text_blocks.entry(index).or_default().push_str(piece);
233 if let Some(tx) = text_tx {
234 let _ = tx.send(piece.to_string());
235 }
236 }
237 }
238 Some("input_json_delta") => {
239 let piece = delta
240 .get("partial_json")
241 .and_then(|t| t.as_str())
242 .unwrap_or_default();
243 if !piece.is_empty() {
244 tool_blocks
245 .entry(index)
246 .or_default()
247 .input_json
248 .push_str(piece);
249 }
250 }
251 _ => {}
252 }
253 }
254 "message_delta" => {
255 if let Some(reason) = v
256 .get("delta")
257 .and_then(|d| d.get("stop_reason"))
258 .and_then(|s| s.as_str())
259 {
260 *stop_reason = Some(reason.to_string());
261 }
262 if let Some(u) = v.get("usage") {
263 *usage = usage_from_json(u);
264 }
265 }
266 "message_start" => {
267 if let Some(u) = v.get("message").and_then(|m| m.get("usage")) {
268 *usage = usage_from_json(u);
269 }
270 }
271 _ => {}
272 }
273}
274
275#[derive(Debug, Deserialize)]
276pub(crate) struct AnthropicApiError {
277 error: AnthropicApiErrorDetail,
278}
279
280#[derive(Debug, Deserialize)]
281pub(crate) struct AnthropicApiErrorDetail {
282 message: String,
283 #[serde(rename = "type")]
284 error_type: String,
285}
286
287#[async_trait]
288impl LlmProvider for AnthropicProvider {
289 async fn send_message(
290 &self,
291 system: &str,
292 messages: Vec<Message>,
293 tools: Option<Vec<ToolDefinition>>,
294 ) -> Result<MessagesResponse, MicroClawError> {
295 self.send_message_with_model(system, messages, tools, None)
296 .await
297 }
298
299 async fn send_message_with_model(
300 &self,
301 system: &str,
302 messages: Vec<Message>,
303 tools: Option<Vec<ToolDefinition>>,
304 model_override: Option<&str>,
305 ) -> Result<MessagesResponse, MicroClawError> {
306 let messages = sanitize_messages(messages);
307 let model = resolve_request_model("anthropic", &self.model, model_override);
308
309 let request = MessagesRequest {
310 model,
311 max_tokens: self.max_tokens,
312 system: system.to_string(),
313 messages,
314 tools,
315 stream: None,
316 };
317
318 let mut retries = 0u32;
319 let max_retries = 3;
320 let body = self.build_request_body(&request)?;
321
322 loop {
323 let send_result = self
324 .http
325 .post(&self.base_url)
326 .header("x-api-key", self.key_pool.current())
327 .header("anthropic-version", "2023-06-01")
328 .header("content-type", "application/json")
329 .json(&body)
330 .send()
331 .await;
332 let response = match send_result {
333 Ok(r) => r,
334 Err(e) if is_retryable_transport_error(&e) && retries < max_retries => {
335 retries += 1;
336 let delay = retry_backoff(retries, None);
337 warn!(
338 "LLM transport error ({e}), retrying in {delay:?} (attempt {retries}/{max_retries})"
339 );
340 tokio::time::sleep(delay).await;
341 continue;
342 }
343 Err(e) => return Err(e.into()),
344 };
345
346 let status = response.status();
347
348 if status.is_success() {
349 let body = response.text().await?;
350 let parsed: MessagesResponse = serde_json::from_str(&body).map_err(|e| {
351 MicroClawError::LlmApi(format!("Failed to parse response: {e}\nBody: {body}"))
352 })?;
353 return Ok(parsed);
354 }
355
356 let rotated = is_key_rotation_status(status.as_u16())
357 && retries < max_retries
358 && self.key_pool.advance();
359 if rotated {
360 warn!(
361 "Auth/rate-limit error (HTTP {status}); rotated to next API key ({} in pool)",
362 self.key_pool.len()
363 );
364 }
365 if (is_retryable_status(status.as_u16()) || rotated) && retries < max_retries {
366 retries += 1;
367 let retry_after = parse_retry_after(response.headers());
368 let delay = if rotated && !is_retryable_status(status.as_u16()) {
369 std::time::Duration::from_millis(200)
370 } else {
371 retry_backoff(retries, retry_after)
372 };
373 warn!(
374 "Transient LLM error (HTTP {status}), retrying in {delay:?} (attempt {retries}/{max_retries})"
375 );
376 tokio::time::sleep(delay).await;
377 continue;
378 }
379
380 let body = response.text().await.unwrap_or_default();
381 if let Ok(api_err) = serde_json::from_str::<AnthropicApiError>(&body) {
382 return Err(MicroClawError::LlmApi(format!(
383 "{}: {}",
384 api_err.error.error_type, api_err.error.message
385 )));
386 }
387 return Err(MicroClawError::LlmApi(format!("HTTP {status}: {body}")));
388 }
389 }
390
391 async fn send_message_stream(
392 &self,
393 system: &str,
394 messages: Vec<Message>,
395 tools: Option<Vec<ToolDefinition>>,
396 text_tx: Option<&UnboundedSender<String>>,
397 ) -> Result<MessagesResponse, MicroClawError> {
398 self.send_message_stream_with_model(system, messages, tools, text_tx, None)
399 .await
400 }
401
402 async fn send_message_stream_with_model(
403 &self,
404 system: &str,
405 messages: Vec<Message>,
406 tools: Option<Vec<ToolDefinition>>,
407 text_tx: Option<&UnboundedSender<String>>,
408 model_override: Option<&str>,
409 ) -> Result<MessagesResponse, MicroClawError> {
410 let messages = sanitize_messages(messages);
411 let model = resolve_request_model("anthropic", &self.model, model_override);
412 let request = MessagesRequest {
413 model,
414 max_tokens: self.max_tokens,
415 system: system.to_string(),
416 messages,
417 tools,
418 stream: Some(true),
419 };
420
421 self.send_message_stream_single_pass(&request, text_tx)
422 .await
423 }
424}
425
426#[cfg(test)]
431mod tests {
432 use super::*;
433 #[allow(unused_imports)]
434 use crate::llm::test_prelude::*;
435
436 #[test]
437 fn test_resolve_anthropic_messages_url_defaults() {
438 let url = resolve_anthropic_messages_url("");
439 assert_eq!(url, "https://api.anthropic.com/v1/messages");
440 }
441
442 #[test]
443 fn test_resolve_anthropic_messages_url_accepts_full_messages_path() {
444 let url = resolve_anthropic_messages_url("http://127.0.0.1:3000/api/v1/messages");
445 assert_eq!(url, "http://127.0.0.1:3000/api/v1/messages");
446 }
447
448 #[test]
449 fn test_resolve_anthropic_messages_url_appends_messages_path_for_prefix_base() {
450 let url = resolve_anthropic_messages_url("http://127.0.0.1:3000/api/");
451 assert_eq!(url, "http://127.0.0.1:3000/api/v1/messages");
452 }
453}