1use serde::{Deserialize, Serialize};
2use tokio::sync::broadcast;
3use tokio_util::sync::CancellationToken;
4
5use crate::error::RuntimeError;
6use crate::event::{NodeEvent, Observable};
7use crate::message::{ImageData, Message, MessageOrigin, MessagePart, MessageRole};
8use crate::provider::{
9 AssistantMessage, CallTiming, DEFAULT_STREAM_BUFFER, LlmRequest, Provider, StopReason,
10 TokenUsage, estimate_tokens,
11};
12use crate::providers::classify_attachment_error;
13use crate::tool::BoxFut;
14
15pub struct AnthropicProvider {
16 name: String,
17 api_key: String,
18 base_url: String,
19 client: reqwest::Client,
20 max_tokens: u32,
21 anthropic_version: String,
22}
23
24impl AnthropicProvider {
25 pub fn new(name: impl Into<String>, api_key: impl Into<String>) -> Self {
26 Self {
27 name: name.into(),
28 api_key: api_key.into(),
29 base_url: "https://api.anthropic.com".into(),
30 client: reqwest::Client::new(),
31 max_tokens: 16384,
32 anthropic_version: "2023-06-01".into(),
33 }
34 }
35
36 pub fn with_base_url(mut self, url: impl Into<String>) -> Self {
37 self.base_url = url.into();
38 self
39 }
40
41 pub fn with_max_tokens(mut self, n: u32) -> Self {
42 self.max_tokens = n;
43 self
44 }
45
46 pub fn with_anthropic_version(mut self, v: impl Into<String>) -> Self {
47 self.anthropic_version = v.into();
48 self
49 }
50
51 fn build_body(&self, req: &LlmRequest, stream: bool) -> MessagesRequest {
52 let raw_wire: Vec<WireMessage> = req
53 .messages
54 .iter()
55 .map(|m| build_wire_message(m, false))
56 .collect();
57 let wire_messages = merge_consecutive_same_role(raw_wire);
58 let tools: Vec<WireTool> = req
59 .tools
60 .iter()
61 .map(|t| WireTool {
62 name: crate::tool_naming::to_wire(&t.name),
63 description: t.description.clone(),
64 input_schema: t.input_schema.clone(),
65 })
66 .collect();
67 MessagesRequest {
68 model: req.model.clone(),
69 max_tokens: self.max_tokens,
70 stream,
71 system: req.system.clone(),
72 messages: wire_messages,
73 tools,
74 thinking: if req.thinking_enabled {
75 Some(ThinkingConfig {
76 kind: "enabled",
77 budget_tokens: Some(self.max_tokens.saturating_sub(4096).max(1024)),
78 })
79 } else {
80 Some(ThinkingConfig {
81 kind: "disabled",
82 budget_tokens: None,
83 })
84 },
85 cache_control: if req.cache_prompt {
86 Some(CacheControl { kind: "ephemeral" })
87 } else {
88 None
89 },
90 }
91 }
92
93 fn build_request(&self, req: &LlmRequest, stream: bool) -> reqwest::RequestBuilder {
94 let body = self.build_body(req, stream);
95 self.client
96 .post(format!("{}/v1/messages", self.base_url))
97 .header("x-api-key", &self.api_key)
98 .header("anthropic-version", &self.anthropic_version)
99 .json(&body)
100 }
101
102 #[doc(hidden)]
103 pub fn wire_body_bytes(&self, req: &LlmRequest, stream: bool) -> Vec<u8> {
104 serde_json::to_vec(&self.build_body(req, stream)).expect("serialize wire body")
105 }
106}
107
108fn build_wire_message(m: &Message, apply_cache_control: bool) -> WireMessage {
111 let role = match m.role {
112 MessageRole::User => "user",
113 MessageRole::Assistant => "assistant",
114 MessageRole::System => "user",
115 MessageRole::Tool => "user",
116 };
117 let mut blocks: Vec<ContentPart> = Vec::with_capacity(m.parts.len());
118 let last_idx = m.parts.len().saturating_sub(1);
119 for (i, part) in m.parts.iter().enumerate() {
120 blocks.push(match part {
121 MessagePart::CompactSummary { summary, .. } => ContentPart::Text {
122 text: summary.clone(),
123 cache_control: if apply_cache_control && i == last_idx {
124 Some(CacheControl { kind: "ephemeral" })
125 } else {
126 None
127 },
128 },
129 MessagePart::Text { text } => ContentPart::Text {
130 text: text.clone(),
131 cache_control: if apply_cache_control && i == last_idx {
132 Some(CacheControl { kind: "ephemeral" })
133 } else {
134 None
135 },
136 },
137 MessagePart::Image { source } => {
138 let data = match &source.data {
139 ImageData::Base64 { data } => data.clone(),
140 ImageData::Path { path } => {
141 let bytes = std::fs::read(path).unwrap_or_default();
142 use base64::Engine;
143 base64::engine::general_purpose::STANDARD.encode(&bytes)
144 }
145 };
146 ContentPart::Image {
147 source: ImageSourceWire {
148 kind: "base64",
149 media_type: source.media_type.clone(),
150 data,
151 },
152 }
153 }
154 MessagePart::ToolUse { id, name, input } => ContentPart::ToolUse {
155 id: id.clone(),
156 name: crate::tool_naming::to_wire(name),
157 input: input.clone(),
158 },
159 MessagePart::Thinking {
160 thinking,
161 signature,
162 } => {
163 if signature.is_none() {
164 continue;
165 }
166 ContentPart::Thinking {
167 thinking: thinking.clone(),
168 signature: signature.clone(),
169 }
170 }
171 MessagePart::ToolResult {
172 tool_use_id,
173 content,
174 is_error,
175 } => ContentPart::ToolResult {
176 tool_use_id: tool_use_id.clone(),
177 content: content.clone(),
178 is_error: *is_error,
179 },
180 });
181 }
182 WireMessage {
183 role,
184 content: MessageContent::Blocks(blocks),
185 }
186}
187
188fn merge_consecutive_same_role(wire: Vec<WireMessage>) -> Vec<WireMessage> {
189 let mut out: Vec<WireMessage> = Vec::with_capacity(wire.len());
190 for msg in wire {
191 let WireMessage { role, content } = msg;
192 let mut content = Some(content);
193 if let Some(last) = out.last_mut()
194 && last.role == role
195 && let Some(msg_content) = content.take()
196 {
197 let MessageContent::Blocks(last_blocks) = &mut last.content;
198 let MessageContent::Blocks(msg_blocks) = msg_content;
199 last_blocks.extend(msg_blocks);
200 }
201 if let Some(content) = content {
202 out.push(WireMessage { role, content });
203 }
204 }
205 out
206}
207
208impl Provider for AnthropicProvider {
209 fn name(&self) -> &str {
210 &self.name
211 }
212
213 fn call<'a>(&'a self, req: LlmRequest) -> BoxFut<'a, Result<AssistantMessage, RuntimeError>> {
214 let request = self.build_request(&req, false);
215 Box::pin(async move {
216 let resp = request.send().await.map_err(net_err)?;
217 let status = resp.status();
218 let body: MessagesResponse = if status.is_success() {
219 resp.json().await.map_err(net_err)?
220 } else {
221 let body_text = resp.text().await.unwrap_or_default();
222 if let Some(reason) = classify_attachment_error(status.as_u16(), &body_text) {
223 return Err(RuntimeError::AttachmentError { reason });
224 }
225 return Err(RuntimeError::ToolFailed(format!(
226 "anthropic http {status}: {body_text}"
227 )));
228 };
229 Ok(response_to_assistant(
230 body,
231 next_turn_id_from_req(&req),
232 &req.tools,
233 ))
234 })
235 }
236
237 fn call_streaming(&self, req: LlmRequest) -> Observable<AssistantMessage> {
238 let request = self.build_request(&req, true);
239 let turn_id = next_turn_id_from_req(&req);
240 let tools: Vec<crate::tool::ToolSpec> = req.tools.clone();
241 let (tx, events) = broadcast::channel(DEFAULT_STREAM_BUFFER);
242 let cancel = CancellationToken::new();
243 let cancel_for_task = cancel.clone();
244 let output: BoxFut<'static, Result<AssistantMessage, RuntimeError>> = Box::pin(
245 async move {
246 use eventsource_stream::Eventsource;
247 use futures::StreamExt;
248
249 let resp = tokio::select! {
250 biased;
251 _ = cancel_for_task.cancelled() => return Err(RuntimeError::Cancelled("anthropic cancelled before send".into())),
252 r = request.send() => r.map_err(net_err)?,
253 };
254 let status = resp.status();
255 if !status.is_success() {
256 let body = resp.text().await.unwrap_or_default();
257 if let Some(reason) = classify_attachment_error(status.as_u16(), &body) {
258 return Err(RuntimeError::AttachmentError { reason });
259 }
260 return Err(RuntimeError::ToolFailed(format!(
261 "anthropic http {status}: {body}"
262 )));
263 }
264
265 let mut stream = resp.bytes_stream().eventsource();
266 let mut acc_text = String::new();
267 let mut acc_thinking = String::new();
268 let mut acc_signature: Option<String> = None;
269 let mut cumulative = 0u64;
270 let mut input_tokens: u64 = 0;
271 let mut cache_read_tokens: u64 = 0;
272 let mut cache_write_tokens: u64 = 0;
273 let mut tool_use_partial: Vec<PartialToolUse> = Vec::new();
274 let mut stop_reason = StopReason::End;
275 while let Some(event) = tokio::select! {
276 biased;
277 _ = cancel_for_task.cancelled() => None,
278 next = stream.next() => next,
279 } {
280 let event = event.map_err(|e| RuntimeError::ToolFailed(format!("sse: {e}")))?;
281 if event.data.is_empty() {
282 continue;
283 }
284 let parsed: serde_json::Value = match serde_json::from_str(&event.data) {
285 Ok(v) => v,
286 Err(_) => continue,
287 };
288 let ty = parsed.get("type").and_then(|v| v.as_str()).unwrap_or("");
289 match ty {
290 "message_start" => {
291 if let Some(usage) = parsed.pointer("/message/usage") {
292 input_tokens = usage
293 .get("input_tokens")
294 .and_then(|v| v.as_u64())
295 .unwrap_or(0);
296 cache_read_tokens = usage
297 .get("cache_read_input_tokens")
298 .and_then(|v| v.as_u64())
299 .unwrap_or(0);
300 cache_write_tokens = usage
301 .get("cache_creation_input_tokens")
302 .and_then(|v| v.as_u64())
303 .unwrap_or(0);
304 }
305 }
306 "content_block_start" => {
307 if let Some(block) = parsed.get("content_block") {
308 match block.get("type").and_then(|v| v.as_str()) {
309 Some("tool_use") => {
310 if let (Some(id), Some(name)) = (
311 block.get("id").and_then(|v| v.as_str()),
312 block.get("name").and_then(|v| v.as_str()),
313 ) {
314 tool_use_partial.push(PartialToolUse {
315 id: id.to_string(),
316 name: name.to_string(),
317 input_json: String::new(),
318 });
319 }
320 }
321 Some("thinking") => {
322 if let Some(sig) =
323 block.get("signature").and_then(|v| v.as_str())
324 {
325 acc_signature = Some(sig.to_string());
326 }
327 }
328 _ => {}
329 }
330 }
331 }
332 "content_block_delta" => {
333 if let Some(delta) = parsed.get("delta") {
334 let delta_ty =
335 delta.get("type").and_then(|v| v.as_str()).unwrap_or("");
336 if delta_ty == "text_delta" {
337 if let Some(text) = delta.get("text").and_then(|v| v.as_str()) {
338 acc_text.push_str(text);
339 cumulative += estimate_tokens(text);
340 let _ = tx.send(NodeEvent::LlmChunk {
341 text: text.to_string(),
342 cumulative_tokens: cumulative,
343 });
344 }
345 } else if delta_ty == "thinking_delta" {
346 if let Some(text) =
347 delta.get("thinking").and_then(|v| v.as_str())
348 {
349 acc_thinking.push_str(text);
350 let _ = tx.send(NodeEvent::ThinkingChunk {
351 text: text.to_string(),
352 });
353 }
354 } else if let Some(text) =
355 delta.get("reasoning_content").and_then(|v| v.as_str())
356 {
357 acc_thinking.push_str(text);
358 let _ = tx.send(NodeEvent::ThinkingChunk {
359 text: text.to_string(),
360 });
361 } else if delta_ty == "signature_delta" {
362 if let Some(sig) =
363 delta.get("signature").and_then(|v| v.as_str())
364 {
365 acc_signature = Some(sig.to_string());
366 }
367 } else if delta_ty == "input_json_delta"
368 && let Some(partial) =
369 delta.get("partial_json").and_then(|v| v.as_str())
370 && let Some(last) = tool_use_partial.last_mut()
371 {
372 last.input_json.push_str(partial);
373 }
374 }
375 }
376 "message_delta" => {
377 if let Some(out) = parsed
378 .pointer("/usage/output_tokens")
379 .and_then(|v| v.as_u64())
380 {
381 cumulative = out;
382 }
383 if let Some(inp) = parsed
384 .pointer("/usage/input_tokens")
385 .and_then(|v| v.as_u64())
386 {
387 input_tokens = inp;
388 }
389 if let Some(cr) = parsed
390 .pointer("/usage/cache_read_input_tokens")
391 .and_then(|v| v.as_u64())
392 {
393 cache_read_tokens = cr;
394 }
395 if let Some(cw) = parsed
396 .pointer("/usage/cache_creation_input_tokens")
397 .and_then(|v| v.as_u64())
398 {
399 cache_write_tokens = cw;
400 }
401 if let Some(reason) = parsed
402 .pointer("/delta/stop_reason")
403 .and_then(|v| v.as_str())
404 {
405 stop_reason = parse_stop_reason(reason);
406 }
407 }
408 "message_stop" => break,
409 _ => {}
410 }
411 }
412 if cancel_for_task.is_cancelled() {
413 let _ = tx.send(NodeEvent::LlmDone {
414 total_tokens: cumulative,
415 });
416 return Err(RuntimeError::Cancelled(
417 "anthropic cancelled mid-stream".into(),
418 ));
419 }
420 let _ = tx.send(NodeEvent::LlmDone {
421 total_tokens: cumulative,
422 });
423
424 let mut parts: Vec<MessagePart> = Vec::new();
425 if !acc_thinking.is_empty() {
426 if req.thinking_enabled && acc_signature.is_none() {
427 return Err(RuntimeError::ThinkingSignatureMissing);
428 }
429 parts.push(MessagePart::Thinking {
430 thinking: acc_thinking,
431 signature: acc_signature,
432 });
433 }
434 if !acc_text.is_empty() {
435 parts.push(MessagePart::Text { text: acc_text });
436 }
437 for pu in tool_use_partial {
438 let input: serde_json::Value = if pu.input_json.is_empty() {
439 serde_json::Value::Object(Default::default())
440 } else {
441 serde_json::from_str(&pu.input_json).unwrap_or(serde_json::Value::Null)
442 };
443 parts.push(MessagePart::ToolUse {
444 id: pu.id,
445 name: crate::tool_naming::from_wire(&pu.name, &tools),
446 input,
447 });
448 }
449 Ok(AssistantMessage {
450 message: Message {
451 role: MessageRole::Assistant,
452 parts,
453 turn_id,
454 origin: MessageOrigin::User,
455 },
456 stop_reason,
457 token_usage: TokenUsage {
458 input: input_tokens,
459 cached_input: cache_read_tokens,
460 output: cumulative,
461 cache_write: cache_write_tokens,
462 ..Default::default()
463 },
464 timing: CallTiming::default(),
465 model: String::new(),
466 response_id: None,
467 })
468 },
469 );
470 Observable {
471 output,
472 events,
473 cancel,
474 }
475 }
476
477 fn test_connection(&self) -> BoxFut<'_, Result<String, String>> {
478 let base_url = self.base_url.clone();
479 let api_key = self.api_key.clone();
480 let name = self.name.clone();
481 Box::pin(async move {
482 let client = reqwest::Client::builder()
483 .timeout(std::time::Duration::from_secs(15))
484 .build()
485 .map_err(|e| e.to_string())?;
486 let resp = client
487 .get(format!("{}/v1/models", base_url.trim_end_matches('/')))
488 .header("x-api-key", &api_key)
489 .header("anthropic-version", "2023-06-01")
490 .send()
491 .await
492 .map_err(|e| format!("connection failed — {e}"))?;
493 let status = resp.status();
494 if status.is_success() {
495 Ok(format!("\"{name}\" responded OK"))
496 } else {
497 let body = resp.text().await.unwrap_or_default();
498 Err(format!(
499 "returned {status} — {}",
500 &body[..body.len().min(200)]
501 ))
502 }
503 })
504 }
505}
506
507struct PartialToolUse {
508 id: String,
509 name: String,
510 input_json: String,
511}
512
513fn response_to_assistant(
514 body: MessagesResponse,
515 turn_id: crate::event::TurnId,
516 tools: &[crate::tool::ToolSpec],
517) -> AssistantMessage {
518 let mut parts: Vec<MessagePart> = Vec::new();
519 for block in body.content {
520 match block {
521 ContentBlock::Text { text } => parts.push(MessagePart::Text { text }),
522 ContentBlock::Thinking {
523 thinking,
524 signature,
525 } => parts.push(MessagePart::Thinking {
526 thinking,
527 signature,
528 }),
529 ContentBlock::ToolUse { id, name, input } => parts.push(MessagePart::ToolUse {
530 id,
531 name: crate::tool_naming::from_wire(&name, tools),
532 input,
533 }),
534 ContentBlock::Other => {}
535 }
536 }
537 let stop_reason = body
538 .stop_reason
539 .as_deref()
540 .map(parse_stop_reason)
541 .unwrap_or(StopReason::End);
542 let usage = body
543 .usage
544 .map(|u| TokenUsage {
545 input: u.input_tokens.unwrap_or(0),
546 cached_input: u.cache_read_input_tokens.unwrap_or(0),
547 output: u.output_tokens.unwrap_or(0),
548 cache_write: u.cache_creation_input_tokens.unwrap_or(0),
549 ..Default::default()
550 })
551 .unwrap_or_default();
552 AssistantMessage {
553 message: Message {
554 role: MessageRole::Assistant,
555 parts,
556 turn_id,
557 origin: MessageOrigin::User,
558 },
559 stop_reason,
560 token_usage: usage,
561 timing: CallTiming::default(),
562 model: body.model.unwrap_or_default(),
563 response_id: body.id,
564 }
565}
566
567fn parse_stop_reason(s: &str) -> StopReason {
568 match s {
569 "tool_use" => StopReason::ToolUse,
570 "max_tokens" => StopReason::Length,
571 _ => StopReason::End,
572 }
573}
574
575fn next_turn_id_from_req(req: &LlmRequest) -> crate::event::TurnId {
576 req.messages
577 .first()
578 .map(|m| m.turn_id.clone())
579 .unwrap_or_else(crate::event::TurnId::now)
580}
581
582fn net_err(e: reqwest::Error) -> RuntimeError {
583 RuntimeError::ToolFailed(format!("anthropic net: {e}"))
584}
585
586#[derive(Serialize, Clone)]
587struct MessagesRequest {
588 model: String,
589 max_tokens: u32,
590 stream: bool,
591 #[serde(skip_serializing_if = "Option::is_none")]
592 system: Option<String>,
593 messages: Vec<WireMessage>,
594 #[serde(skip_serializing_if = "Vec::is_empty")]
595 tools: Vec<WireTool>,
596 #[serde(skip_serializing_if = "Option::is_none")]
597 thinking: Option<ThinkingConfig>,
598 #[serde(skip_serializing_if = "Option::is_none")]
599 cache_control: Option<CacheControl>,
600}
601
602#[derive(Serialize, Clone)]
603struct ThinkingConfig {
604 #[serde(rename = "type")]
605 kind: &'static str,
606 #[serde(skip_serializing_if = "Option::is_none")]
607 budget_tokens: Option<u32>,
608}
609
610#[derive(Serialize, Clone)]
611struct WireTool {
612 name: String,
613 #[serde(skip_serializing_if = "Option::is_none")]
614 description: Option<String>,
615 input_schema: serde_json::Value,
616}
617
618#[derive(Serialize, Clone)]
619struct WireMessage {
620 role: &'static str,
621 content: MessageContent,
622}
623
624#[derive(Serialize, Clone)]
625#[serde(untagged)]
626enum MessageContent {
627 Blocks(Vec<ContentPart>),
628}
629
630#[derive(Serialize, Clone)]
631#[serde(tag = "type", rename_all = "snake_case")]
632enum ContentPart {
633 Text {
634 text: String,
635 #[serde(skip_serializing_if = "Option::is_none")]
636 cache_control: Option<CacheControl>,
637 },
638 Thinking {
639 thinking: String,
640 #[serde(skip_serializing_if = "Option::is_none")]
641 signature: Option<String>,
642 },
643 Image {
644 source: ImageSourceWire,
645 },
646 ToolUse {
647 id: String,
648 name: String,
649 input: serde_json::Value,
650 },
651 ToolResult {
652 tool_use_id: String,
653 content: String,
654 #[serde(skip_serializing_if = "core::ops::Not::not")]
655 is_error: bool,
656 },
657}
658
659#[derive(Serialize, Clone)]
660struct ImageSourceWire {
661 #[serde(rename = "type")]
662 kind: &'static str,
663 media_type: String,
664 data: String,
665}
666
667#[derive(Serialize, Clone)]
668struct CacheControl {
669 #[serde(rename = "type")]
670 kind: &'static str,
671}
672
673#[derive(Deserialize)]
674struct MessagesResponse {
675 content: Vec<ContentBlock>,
676 #[serde(default)]
677 stop_reason: Option<String>,
678 #[serde(default)]
679 usage: Option<AnthropicUsage>,
680 #[serde(default)]
681 model: Option<String>,
682 #[serde(default)]
683 id: Option<String>,
684}
685
686#[derive(Deserialize, Default)]
687struct AnthropicUsage {
688 #[serde(default)]
689 input_tokens: Option<u64>,
690 #[serde(default)]
691 output_tokens: Option<u64>,
692 #[serde(default)]
693 cache_read_input_tokens: Option<u64>,
694 #[serde(default)]
695 cache_creation_input_tokens: Option<u64>,
696}
697
698#[derive(Deserialize)]
699#[serde(tag = "type", rename_all = "snake_case")]
700enum ContentBlock {
701 Text {
702 text: String,
703 },
704 Thinking {
705 thinking: String,
706 #[serde(default)]
707 signature: Option<String>,
708 },
709 ToolUse {
710 id: String,
711 name: String,
712 input: serde_json::Value,
713 },
714 #[serde(other)]
715 Other,
716}