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 && !text.is_empty()
108 {
109 parts.push(json!({"type": "text", "text": text}));
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 ChatMessage::Custom { role: _, data } => {
141 result.push(json!({
143 "role": "user",
144 "content": [{
145 "type": "text",
146 "text": data.to_string(),
147 }]
148 }));
149 }
150 }
151 }
152
153 (system_prompt, result)
154 }
155
156 fn convert_tools(tools: &[Value]) -> Vec<Value> {
157 tools
158 .iter()
159 .filter_map(|tool| {
160 let func = tool.get("function")?;
161 let name = func.get("name")?.as_str()?;
162 let description = func
163 .get("description")
164 .and_then(Value::as_str)
165 .unwrap_or("");
166 let input_schema = func
167 .get("parameters")
168 .cloned()
169 .unwrap_or_else(|| json!({"type": "object"}));
170 Some(json!({
171 "name": name,
172 "description": description,
173 "input_schema": input_schema,
174 }))
175 })
176 .collect()
177 }
178
179 fn build_body(
180 messages: &[ChatMessage],
181 tools: &[Value],
182 model: &str,
183 reasoning: Option<&ReasoningConfig>,
184 ) -> Value {
185 let (system_prompt, anthropic_messages) = Self::convert_messages(messages);
186 let anthropic_tools = Self::convert_tools(tools);
187
188 let mut body = json!({
189 "model": model,
190 "max_tokens": 8192,
191 "messages": anthropic_messages,
192 });
193
194 if !anthropic_tools.is_empty()
195 && let Some(obj) = body.as_object_mut()
196 {
197 obj.insert("tools".to_string(), json!(anthropic_tools));
198 }
199
200 if let Some(system) = system_prompt
201 && let Some(obj) = body.as_object_mut()
202 {
203 obj.insert("system".to_string(), json!(system));
204 }
205
206 if let Some(config) = reasoning {
207 if config.enabled == Some(true) || config.budget_tokens.is_some() {
208 let mut thinking = serde_json::Map::new();
209 thinking.insert("type".to_string(), json!("enabled"));
210 if let Some(budget) = config.budget_tokens {
211 thinking.insert("budget_tokens".to_string(), json!(budget));
212 }
213 if let Some(obj) = body.as_object_mut() {
214 obj.insert("thinking".to_string(), Value::Object(thinking));
215 }
216 } else if config.enabled == Some(false) {
217 let mut thinking = serde_json::Map::new();
218 thinking.insert("type".to_string(), json!("disabled"));
219 if let Some(obj) = body.as_object_mut() {
220 obj.insert("thinking".to_string(), Value::Object(thinking));
221 }
222 }
223 }
224
225 body
226 }
227
228 fn parse_sse(data_str: &str, event_type: &str) -> AgentResult<StreamChunk> {
229 if data_str.is_empty() {
230 return Ok(StreamChunk::Text(String::new()));
231 }
232
233 let data: Value = serde_json::from_str(data_str)
234 .map_err(|e| AgentError::json(format!("Anthropic SSE JSON: {e}")))?;
235
236 match event_type {
237 "message_start" => {
238 let input_tokens = data
239 .get("message")
240 .and_then(|m| m.get("usage"))
241 .and_then(|u| u.get("input_tokens"))
242 .and_then(Value::as_u64)
243 .map(|v| v as u32);
244 let output_tokens = data
245 .get("message")
246 .and_then(|m| m.get("usage"))
247 .and_then(|u| u.get("output_tokens"))
248 .and_then(Value::as_u64)
249 .map(|v| v as u32);
250 Ok(StreamChunk::Usage(UsageInfo {
251 prompt_tokens: input_tokens,
252 completion_tokens: output_tokens,
253 total_tokens: None,
254 }))
255 }
256 "content_block_start" => {
257 let cb = data.get("content_block");
258 let idx = data.get("index").and_then(Value::as_u64).unwrap_or(0);
259 if let Some(cb) = cb
260 && cb.get("type").and_then(Value::as_str) == Some("tool_use")
261 {
262 let id = cb
263 .get("id")
264 .and_then(Value::as_str)
265 .unwrap_or("")
266 .to_string();
267 let name = cb
268 .get("name")
269 .and_then(Value::as_str)
270 .unwrap_or("")
271 .to_string();
272 return Ok(StreamChunk::ToolCall(json!({
273 "delta": {
274 "tool_calls": [{
275 "index": idx,
276 "id": if id.is_empty() { Value::Null } else { json!(id) },
277 "function": {
278 "name": name,
279 "arguments": "",
280 }
281 }]
282 }
283 })));
284 }
285 Ok(StreamChunk::Text(String::new()))
286 }
287 "content_block_delta" => {
288 let delta = data.get("delta");
289 let idx = data.get("index").and_then(Value::as_u64).unwrap_or(0);
290 if let Some(d) = delta {
291 match d.get("type").and_then(Value::as_str) {
292 Some("text_delta") => {
293 let text = d
294 .get("text")
295 .and_then(Value::as_str)
296 .unwrap_or("")
297 .to_string();
298 Ok(StreamChunk::Text(text))
299 }
300 Some("input_json_delta") => {
301 let partial = d
302 .get("partial_json")
303 .and_then(Value::as_str)
304 .unwrap_or("")
305 .to_string();
306 Ok(StreamChunk::ToolCall(json!({
307 "delta": {
308 "tool_calls": [{
309 "index": idx,
310 "function": {
311 "arguments": partial,
312 }
313 }]
314 }
315 })))
316 }
317 Some("thinking_delta") => {
318 let thinking = d
319 .get("thinking")
320 .and_then(Value::as_str)
321 .unwrap_or("")
322 .to_string();
323 Ok(StreamChunk::Thought(thinking))
324 }
325 _ => Ok(StreamChunk::Text(String::new())),
326 }
327 } else {
328 Ok(StreamChunk::Text(String::new()))
329 }
330 }
331 "content_block_stop" => Ok(StreamChunk::Text(String::new())),
332 "message_delta" => {
333 let output_tokens = data
334 .get("usage")
335 .and_then(|u| u.get("output_tokens"))
336 .and_then(Value::as_u64)
337 .map(|v| v as u32);
338 Ok(StreamChunk::Usage(UsageInfo {
339 prompt_tokens: None,
340 completion_tokens: output_tokens,
341 total_tokens: None,
342 }))
343 }
344 "message_stop" => {
345 let finish_reason = data
346 .get("message")
347 .and_then(|m| m.get("stop_reason"))
348 .and_then(Value::as_str)
349 .map(String::from);
350 Ok(StreamChunk::Stop { finish_reason })
351 }
352 "ping" => Ok(StreamChunk::Text(String::new())),
353 _ => Ok(StreamChunk::Text(String::new())),
354 }
355 }
356}
357
358#[async_trait]
359impl LlmClient for AnthropicClient {
360 async fn chat(
361 &self,
362 messages: &[ChatMessage],
363 tools: &[Value],
364 reasoning: Option<&ReasoningConfig>,
365 _response_format: Option<&ResponseFormat>,
366 ) -> AgentResult<Value> {
367 let url = format!("{}/v1/messages", self.base_url);
368 let body = Self::build_body(messages, tools, &self.model, reasoning);
369 tracing::debug!(model = %self.model, url = %url, body = %serde_json::to_string_pretty(&body).unwrap_or_default(), "Anthropic chat request");
370
371 let response = self
372 .client
373 .post(&url)
374 .header("x-api-key", &self.api_key)
375 .header("anthropic-version", "2023-06-01")
376 .header("Content-Type", "application/json")
377 .json(&body)
378 .send()
379 .await
380 .map_err(|e| AgentError::llm(format!("HTTP request failed: {e}")))?;
381
382 let status = response.status();
383 let res_json: Value = response
384 .json()
385 .await
386 .map_err(|e| AgentError::json(format!("Response JSON parse failed: {e}")))?;
387
388 if !status.is_success() {
389 let err_msg = res_json
390 .get("error")
391 .and_then(|e| e.get("message"))
392 .and_then(Value::as_str)
393 .unwrap_or("unknown error");
394 tracing::warn!(status = %status, error = %err_msg, "Anthropic API non-success");
395 return Err(AgentError::LlmApi {
396 message: err_msg.to_string(),
397 });
398 }
399
400 tracing::debug!(status = %status, "Anthropic chat response received");
401 Ok(res_json)
402 }
403
404 async fn chat_stream(
405 &self,
406 messages: &[ChatMessage],
407 tools: &[Value],
408 reasoning: Option<&ReasoningConfig>,
409 _response_format: Option<&ResponseFormat>,
410 ) -> AgentResult<Pin<Box<dyn Stream<Item = AgentResult<StreamChunk>> + Send>>> {
411 let url = format!("{}/v1/messages", self.base_url);
412 let mut body = Self::build_body(messages, tools, &self.model, reasoning);
413
414 if let Some(obj) = body.as_object_mut() {
415 obj.insert("stream".to_string(), json!(true));
416 }
417 tracing::debug!(model = %self.model, url = %url, body = %serde_json::to_string_pretty(&body).unwrap_or_default(), "Anthropic chat_stream request");
418
419 let response = self
420 .client
421 .post(&url)
422 .header("x-api-key", &self.api_key)
423 .header("anthropic-version", "2023-06-01")
424 .header("Content-Type", "application/json")
425 .json(&body)
426 .send()
427 .await
428 .map_err(|e| AgentError::llm(format!("HTTP request failed: {e}")))?;
429
430 if !response.status().is_success() {
431 let status = response.status();
432 let err_text = response
433 .text()
434 .await
435 .map_err(|e| AgentError::llm(format!("Failed to read error response: {e}")))?;
436 tracing::warn!(%status, error = %err_text, "Anthropic API stream non-success");
437 return Err(AgentError::LlmApi { message: err_text });
438 }
439
440 let stream = response
441 .bytes_stream()
442 .eventsource()
443 .filter_map(|event| async move {
444 match event {
445 Ok(ref ev) if ev.event == "error" => {
446 let err_msg = ev.data.clone();
447 Some(Err(AgentError::LlmApi { message: err_msg }))
448 }
449 Ok(ev) => {
450 let event_type = if ev.event.is_empty() {
451 "message_stop"
452 } else {
453 ev.event.as_str()
454 };
455 match Self::parse_sse(&ev.data, event_type) {
456 Ok(chunk) => Some(Ok(chunk)),
457 Err(e) => Some(Err(e)),
458 }
459 }
460 Err(e) => Some(Err(AgentError::LlmStream(format!("SSE Stream error: {e}")))),
461 }
462 });
463
464 Ok(Box::pin(stream))
465 }
466
467 fn capabilities(&self) -> LlmCapabilities {
468 LlmCapabilities {
469 supports_streaming: true,
470 supports_tools: true,
471 supports_vision: true,
472 supports_thinking: true,
473 max_context_tokens: Some(200_000),
474 max_output_tokens: Some(8_192),
475 }
476 }
477
478 fn model_name(&self) -> &str {
479 &self.model
480 }
481}
482
483#[async_trait]
484impl super::StreamClient for AnthropicClient {
485 async fn stream(
486 &self,
487 messages: &[crate::types::ChatMessage],
488 tools: &[serde_json::Value],
489 reasoning: Option<&super::ReasoningConfig>,
490 response_format: Option<&crate::types::ResponseFormat>,
491 ) -> crate::types::AgentResult<
492 std::pin::Pin<
493 Box<
494 dyn futures_core::Stream<Item = crate::types::AgentResult<super::StreamChunk>>
495 + Send,
496 >,
497 >,
498 > {
499 <Self as super::LlmClient>::chat_stream(self, messages, tools, reasoning, response_format)
500 .await
501 }
502
503 fn capabilities(&self) -> super::LlmCapabilities {
504 <Self as super::LlmClient>::capabilities(self)
505 }
506
507 fn model_name(&self) -> &str {
508 <Self as super::LlmClient>::model_name(self)
509 }
510}