1use async_trait::async_trait;
2use eventsource_stream::Eventsource;
3use futures_core::Stream;
4use futures_util::StreamExt;
5use reqwest::Client;
6use serde_json::{Value, json};
7use std::pin::Pin;
8
9use super::{LlmCapabilities, LlmClient, ReasoningConfig, StreamChunk, UsageInfo};
10use crate::types::{AgentError, AgentResult, ChatMessage, ImageAttachment, ResponseFormat};
11
12pub struct AnthropicClient {
13 api_key: String,
14 model: String,
15 base_url: String,
16 client: Client,
17}
18
19impl AnthropicClient {
20 pub fn new(api_key: String, model: String, base_url: Option<String>) -> Self {
21 Self::new_with_config(
22 api_key,
23 model,
24 base_url,
25 crate::llm::LlmClientConfig::default(),
26 )
27 }
28
29 pub fn new_with_config(
30 api_key: String,
31 model: String,
32 base_url: Option<String>,
33 config: crate::llm::LlmClientConfig,
34 ) -> Self {
35 let client = Client::builder()
36 .connect_timeout(config.connect_timeout)
37 .timeout(config.request_timeout)
38 .pool_max_idle_per_host(config.pool_max_idle_per_host)
39 .pool_idle_timeout(config.pool_idle_timeout)
40 .build()
41 .unwrap_or_else(|e| {
42 tracing::warn!(error = %e, "Failed to build reqwest client with custom config, falling back to default");
43 Client::new()
44 });
45 Self {
46 api_key,
47 model,
48 base_url: base_url.unwrap_or_else(|| "https://api.anthropic.com".to_string()),
49 client,
50 }
51 }
52
53 fn convert_messages(messages: &[ChatMessage]) -> (Option<String>, Vec<Value>) {
54 let mut system_prompt: Option<String> = None;
55 let mut result: Vec<Value> = Vec::new();
56
57 for msg in messages {
58 match msg {
59 ChatMessage::System { content, .. } => {
60 system_prompt = Some(content.clone());
61 }
62 ChatMessage::User {
63 content, images, ..
64 } => {
65 let mut content_parts: Vec<Value> = Vec::new();
66 content_parts.push(json!({"type": "text", "text": content}));
67 for img in images {
68 match img {
69 ImageAttachment::Url { url, detail: _ } => {
70 content_parts.push(json!({
71 "type": "image",
72 "source": {
73 "type": "url",
74 "url": url,
75 }
76 }));
77 }
78 ImageAttachment::Base64 {
79 data,
80 media_type,
81 detail: _,
82 } => {
83 let mime = media_type.as_deref().unwrap_or("image/jpeg");
84 content_parts.push(json!({
85 "type": "image",
86 "source": {
87 "type": "base64",
88 "media_type": mime,
89 "data": data,
90 }
91 }));
92 }
93 }
94 }
95 result.push(json!({
96 "role": "user",
97 "content": content_parts,
98 }));
99 }
100 ChatMessage::Assistant {
101 content,
102 reasoning_content: _,
103 tool_calls,
104 } => {
105 let mut parts: Vec<Value> = Vec::new();
106 if let Some(text) = content {
107 if !text.is_empty() {
108 parts.push(json!({"type": "text", "text": text}));
109 }
110 }
111 if let Some(tc) = tool_calls {
112 for t in tc {
113 let input: Value =
114 serde_json::from_str(&t.arguments).unwrap_or(Value::Null);
115 parts.push(json!({
116 "type": "tool_use",
117 "id": t.id,
118 "name": t.name,
119 "input": input,
120 }));
121 }
122 }
123 if !parts.is_empty() {
124 result.push(json!({"role": "assistant", "content": parts}));
125 }
126 }
127 ChatMessage::Tool {
128 tool_call_id,
129 content,
130 } => {
131 result.push(json!({
132 "role": "user",
133 "content": [{
134 "type": "tool_result",
135 "tool_use_id": tool_call_id,
136 "content": content,
137 }]
138 }));
139 }
140 }
141 }
142
143 (system_prompt, result)
144 }
145
146 fn convert_tools(tools: &[Value]) -> Vec<Value> {
147 tools
148 .iter()
149 .filter_map(|tool| {
150 let func = tool.get("function")?;
151 let name = func.get("name")?.as_str()?;
152 let description = func
153 .get("description")
154 .and_then(Value::as_str)
155 .unwrap_or("");
156 let input_schema = func
157 .get("parameters")
158 .cloned()
159 .unwrap_or_else(|| json!({"type": "object"}));
160 Some(json!({
161 "name": name,
162 "description": description,
163 "input_schema": input_schema,
164 }))
165 })
166 .collect()
167 }
168
169 fn build_body(
170 messages: &[ChatMessage],
171 tools: &[Value],
172 model: &str,
173 reasoning: Option<&ReasoningConfig>,
174 ) -> Value {
175 let (system_prompt, anthropic_messages) = Self::convert_messages(messages);
176 let anthropic_tools = Self::convert_tools(tools);
177
178 let mut body = json!({
179 "model": model,
180 "max_tokens": 8192,
181 "messages": anthropic_messages,
182 });
183
184 if !anthropic_tools.is_empty() {
185 if let Some(obj) = body.as_object_mut() {
186 obj.insert("tools".to_string(), json!(anthropic_tools));
187 }
188 }
189
190 if let Some(system) = system_prompt {
191 if let Some(obj) = body.as_object_mut() {
192 obj.insert("system".to_string(), json!(system));
193 }
194 }
195
196 if let Some(config) = reasoning {
197 if config.enabled == Some(true) || config.budget_tokens.is_some() {
198 let mut thinking = serde_json::Map::new();
199 thinking.insert("type".to_string(), json!("enabled"));
200 if let Some(budget) = config.budget_tokens {
201 thinking.insert("budget_tokens".to_string(), json!(budget));
202 }
203 if let Some(obj) = body.as_object_mut() {
204 obj.insert("thinking".to_string(), Value::Object(thinking));
205 }
206 } else if config.enabled == Some(false) {
207 let mut thinking = serde_json::Map::new();
208 thinking.insert("type".to_string(), json!("disabled"));
209 if let Some(obj) = body.as_object_mut() {
210 obj.insert("thinking".to_string(), Value::Object(thinking));
211 }
212 }
213 }
214
215 body
216 }
217
218 fn parse_sse(data_str: &str, event_type: &str) -> AgentResult<StreamChunk> {
219 if data_str.is_empty() {
220 return Ok(StreamChunk::Text(String::new()));
221 }
222
223 let data: Value = serde_json::from_str(data_str)
224 .map_err(|e| AgentError::json(format!("Anthropic SSE JSON: {e}")))?;
225
226 match event_type {
227 "message_start" => {
228 let input_tokens = data
229 .get("message")
230 .and_then(|m| m.get("usage"))
231 .and_then(|u| u.get("input_tokens"))
232 .and_then(Value::as_u64)
233 .map(|v| v as u32);
234 let output_tokens = data
235 .get("message")
236 .and_then(|m| m.get("usage"))
237 .and_then(|u| u.get("output_tokens"))
238 .and_then(Value::as_u64)
239 .map(|v| v as u32);
240 Ok(StreamChunk::Usage(UsageInfo {
241 prompt_tokens: input_tokens,
242 completion_tokens: output_tokens,
243 total_tokens: None,
244 }))
245 }
246 "content_block_start" => {
247 let cb = data.get("content_block");
248 let idx = data.get("index").and_then(Value::as_u64).unwrap_or(0);
249 if let Some(cb) = cb {
250 if cb.get("type").and_then(Value::as_str) == Some("tool_use") {
251 let id = cb
252 .get("id")
253 .and_then(Value::as_str)
254 .unwrap_or("")
255 .to_string();
256 let name = cb
257 .get("name")
258 .and_then(Value::as_str)
259 .unwrap_or("")
260 .to_string();
261 return Ok(StreamChunk::ToolCall(json!({
262 "delta": {
263 "tool_calls": [{
264 "index": idx,
265 "id": if id.is_empty() { Value::Null } else { json!(id) },
266 "function": {
267 "name": name,
268 "arguments": "",
269 }
270 }]
271 }
272 })));
273 }
274 }
275 Ok(StreamChunk::Text(String::new()))
276 }
277 "content_block_delta" => {
278 let delta = data.get("delta");
279 let idx = data.get("index").and_then(Value::as_u64).unwrap_or(0);
280 if let Some(d) = delta {
281 match d.get("type").and_then(Value::as_str) {
282 Some("text_delta") => {
283 let text = d
284 .get("text")
285 .and_then(Value::as_str)
286 .unwrap_or("")
287 .to_string();
288 Ok(StreamChunk::Text(text))
289 }
290 Some("input_json_delta") => {
291 let partial = d
292 .get("partial_json")
293 .and_then(Value::as_str)
294 .unwrap_or("")
295 .to_string();
296 Ok(StreamChunk::ToolCall(json!({
297 "delta": {
298 "tool_calls": [{
299 "index": idx,
300 "function": {
301 "arguments": partial,
302 }
303 }]
304 }
305 })))
306 }
307 Some("thinking_delta") => {
308 let thinking = d
309 .get("thinking")
310 .and_then(Value::as_str)
311 .unwrap_or("")
312 .to_string();
313 Ok(StreamChunk::Thought(thinking))
314 }
315 _ => Ok(StreamChunk::Text(String::new())),
316 }
317 } else {
318 Ok(StreamChunk::Text(String::new()))
319 }
320 }
321 "content_block_stop" => Ok(StreamChunk::Text(String::new())),
322 "message_delta" => {
323 let output_tokens = data
324 .get("usage")
325 .and_then(|u| u.get("output_tokens"))
326 .and_then(Value::as_u64)
327 .map(|v| v as u32);
328 Ok(StreamChunk::Usage(UsageInfo {
329 prompt_tokens: None,
330 completion_tokens: output_tokens,
331 total_tokens: None,
332 }))
333 }
334 "message_stop" => Ok(StreamChunk::Stop),
335 "ping" => Ok(StreamChunk::Text(String::new())),
336 _ => Ok(StreamChunk::Text(String::new())),
337 }
338 }
339}
340
341#[async_trait]
342impl LlmClient for AnthropicClient {
343 async fn chat(
344 &self,
345 messages: &[ChatMessage],
346 tools: &[Value],
347 reasoning: Option<&ReasoningConfig>,
348 _response_format: Option<&ResponseFormat>,
349 ) -> AgentResult<Value> {
350 let url = format!("{}/v1/messages", self.base_url);
351 let body = Self::build_body(messages, tools, &self.model, reasoning);
352 tracing::debug!(model = %self.model, url = %url, body = %serde_json::to_string_pretty(&body).unwrap_or_default(), "Anthropic chat request");
353
354 let response = self
355 .client
356 .post(&url)
357 .header("x-api-key", &self.api_key)
358 .header("anthropic-version", "2023-06-01")
359 .header("Content-Type", "application/json")
360 .json(&body)
361 .send()
362 .await
363 .map_err(|e| AgentError::llm(format!("HTTP request failed: {e}")))?;
364
365 let status = response.status();
366 let res_json: Value = response
367 .json()
368 .await
369 .map_err(|e| AgentError::json(format!("Response JSON parse failed: {e}")))?;
370
371 if !status.is_success() {
372 let err_msg = res_json
373 .get("error")
374 .and_then(|e| e.get("message"))
375 .and_then(Value::as_str)
376 .unwrap_or("unknown error");
377 tracing::warn!(status = %status, error = %err_msg, "Anthropic API non-success");
378 return Err(AgentError::LlmApi {
379 message: err_msg.to_string(),
380 });
381 }
382
383 tracing::debug!(status = %status, "Anthropic chat response received");
384 Ok(res_json)
385 }
386
387 async fn chat_stream(
388 &self,
389 messages: &[ChatMessage],
390 tools: &[Value],
391 reasoning: Option<&ReasoningConfig>,
392 _response_format: Option<&ResponseFormat>,
393 ) -> AgentResult<Pin<Box<dyn Stream<Item = AgentResult<StreamChunk>> + Send>>> {
394 let url = format!("{}/v1/messages", self.base_url);
395 let mut body = Self::build_body(messages, tools, &self.model, reasoning);
396
397 if let Some(obj) = body.as_object_mut() {
398 obj.insert("stream".to_string(), json!(true));
399 }
400 tracing::debug!(model = %self.model, url = %url, body = %serde_json::to_string_pretty(&body).unwrap_or_default(), "Anthropic chat_stream request");
401
402 let response = self
403 .client
404 .post(&url)
405 .header("x-api-key", &self.api_key)
406 .header("anthropic-version", "2023-06-01")
407 .header("Content-Type", "application/json")
408 .json(&body)
409 .send()
410 .await
411 .map_err(|e| AgentError::llm(format!("HTTP request failed: {e}")))?;
412
413 if !response.status().is_success() {
414 let status = response.status();
415 let err_text = response
416 .text()
417 .await
418 .map_err(|e| AgentError::llm(format!("Failed to read error response: {e}")))?;
419 tracing::warn!(%status, error = %err_text, "Anthropic API stream non-success");
420 return Err(AgentError::LlmApi { message: err_text });
421 }
422
423 let stream = response
424 .bytes_stream()
425 .eventsource()
426 .filter_map(|event| async move {
427 match event {
428 Ok(ref ev) if ev.event == "error" => {
429 let err_msg = ev.data.clone();
430 Some(Err(AgentError::LlmApi { message: err_msg }))
431 }
432 Ok(ev) => {
433 let event_type = if ev.event.is_empty() {
434 "message_stop"
435 } else {
436 ev.event.as_str()
437 };
438 match Self::parse_sse(&ev.data, event_type) {
439 Ok(chunk) => Some(Ok(chunk)),
440 Err(e) => Some(Err(e)),
441 }
442 }
443 Err(e) => Some(Err(AgentError::LlmStream(format!("SSE Stream error: {e}")))),
444 }
445 });
446
447 Ok(Box::pin(stream))
448 }
449
450 fn capabilities(&self) -> LlmCapabilities {
451 LlmCapabilities {
452 supports_streaming: true,
453 supports_tools: true,
454 supports_vision: true,
455 supports_thinking: true,
456 max_context_tokens: Some(200_000),
457 max_output_tokens: Some(8_192),
458 }
459 }
460
461 fn model_name(&self) -> &str {
462 &self.model
463 }
464}