1use super::{
2 FinishReason, LLMRequest, LlmProvider, ProviderError, ProviderStreamChunk, TokenUsage,
3};
4use crate::shared::{ContentBlock, Message, Role, ToolDefinition};
5use async_trait::async_trait;
6use futures_util::stream::BoxStream;
7use reqwest::{self, StatusCode, header::HeaderMap};
8use serde::{Deserialize, Serialize};
9use serde_json::{Map, Value};
10use std::{collections::BTreeMap, time::Duration};
11
12const DEFAULT_BASE_URL: &str = "https://api.anthropic.com/v1";
13const ANTHROPIC_VERSION: &str = "2023-06-01";
14const DEFAULT_MAX_TOKENS: u32 = 4096;
15
16pub struct AnthropicProvider {
17 pub api_key: String,
18 pub base_url: String,
19 client: reqwest::Client,
20}
21
22impl AnthropicProvider {
23 pub fn new(api_key: impl Into<String>) -> Self {
24 Self::with_base_url(api_key, DEFAULT_BASE_URL)
25 }
26
27 pub fn with_base_url(api_key: impl Into<String>, base_url: impl Into<String>) -> Self {
28 Self::with_timeout(api_key, base_url, Duration::from_secs(120))
29 }
30
31 pub fn with_timeout(
32 api_key: impl Into<String>,
33 base_url: impl Into<String>,
34 timeout: Duration,
35 ) -> Self {
36 let client = reqwest::Client::builder()
37 .connect_timeout(Duration::from_secs(10))
38 .timeout(timeout)
39 .build()
40 .expect("reqwest client configuration should be valid");
41 Self {
42 api_key: api_key.into(),
43 base_url: base_url.into(),
44 client,
45 }
46 }
47}
48
49#[derive(Debug, Serialize)]
50struct AnthropicReq {
51 model: String,
52 max_tokens: u32,
53 #[serde(skip_serializing_if = "Option::is_none")]
54 system: Option<String>,
55 messages: Vec<AnthropicMessage>,
56 #[serde(skip_serializing_if = "Vec::is_empty")]
57 tools: Vec<AnthropicTool>,
58 #[serde(skip_serializing_if = "Option::is_none")]
59 temperature: Option<f32>,
60 stream: bool,
61}
62
63#[derive(Debug, Serialize, Deserialize, Clone)]
64struct AnthropicMessage {
65 role: String,
66 content: Vec<AnthropicContentBlock>,
67}
68
69#[derive(Debug, Serialize, Deserialize, Clone)]
70#[serde(tag = "type", rename_all = "snake_case")]
71enum AnthropicContentBlock {
72 Text {
73 text: String,
74 },
75 ToolUse {
76 id: String,
77 name: String,
78 input: Value,
79 },
80 ToolResult {
81 tool_use_id: String,
82 content: String,
83 #[serde(skip_serializing_if = "std::ops::Not::not")]
84 is_error: bool,
85 },
86}
87
88#[derive(Debug, Serialize)]
89struct AnthropicTool {
90 name: String,
91 description: String,
92 input_schema: Value,
93}
94
95#[derive(Debug, Deserialize)]
96#[serde(tag = "type", rename_all = "snake_case")]
97enum StreamEvent {
98 MessageStart {
99 message: StreamMessageStart,
100 },
101 ContentBlockStart {
102 index: usize,
103 content_block: Value,
104 },
105 ContentBlockDelta {
106 index: usize,
107 delta: StreamDelta,
108 },
109 ContentBlockStop {
110 index: usize,
111 },
112 MessageDelta {
113 delta: StreamMessageDelta,
114 usage: Option<AnthropicUsage>,
115 },
116 MessageStop,
117 Ping,
118 Error {
119 error: AnthropicErrorBody,
120 },
121}
122
123#[derive(Debug, Deserialize)]
124struct StreamMessageStart {
125 usage: Option<AnthropicUsage>,
126}
127
128#[derive(Debug, Deserialize)]
129struct StreamMessageDelta {
130 stop_reason: Option<String>,
131}
132
133#[derive(Debug, Deserialize)]
134#[serde(tag = "type", rename_all = "snake_case")]
135enum StreamDelta {
136 TextDelta { text: String },
137 InputJsonDelta { partial_json: String },
138 ThinkingDelta { thinking: String },
139 SignatureDelta { signature: String },
140}
141
142#[derive(Debug, Deserialize, Clone)]
143struct AnthropicUsage {
144 input_tokens: Option<u32>,
145 output_tokens: Option<u32>,
146}
147
148impl From<AnthropicUsage> for TokenUsage {
149 fn from(value: AnthropicUsage) -> Self {
150 Self {
151 input_tokens: value.input_tokens,
152 output_tokens: value.output_tokens,
153 total_tokens: match (value.input_tokens, value.output_tokens) {
154 (Some(input), Some(output)) => Some(input + output),
155 _ => None,
156 },
157 }
158 }
159}
160
161#[derive(Debug, Deserialize)]
162struct AnthropicErrorBody {
163 #[serde(rename = "type")]
164 kind: String,
165 message: String,
166}
167
168#[derive(Default)]
169struct PendingToolBlock {
170 id: Option<String>,
171 name: Option<String>,
172 arguments: String,
173}
174
175#[async_trait]
176impl LlmProvider for AnthropicProvider {
177 fn name(&self) -> &str {
178 "anthropic"
179 }
180
181 fn request_snapshot(&self, req: &LLMRequest) -> Value {
182 serde_json::to_value(build_anthropic_payload(req.clone())).unwrap_or(Value::Null)
183 }
184
185 async fn chat_completion(
186 &self,
187 req: LLMRequest,
188 ) -> Result<BoxStream<'static, Result<ProviderStreamChunk, ProviderError>>, ProviderError> {
189 let payload = build_anthropic_payload(req);
190 let res = self
191 .client
192 .post(format!("{}/messages", self.base_url.trim_end_matches('/')))
193 .header("x-api-key", &self.api_key)
194 .header("anthropic-version", ANTHROPIC_VERSION)
195 .json(&payload)
196 .send()
197 .await
198 .map_err(map_transport_error)?;
199
200 let status = res.status();
201 if !status.is_success() {
202 let headers = res.headers().clone();
203 let err_text = res.text().await.unwrap_or_default();
204 return Err(map_status_error(status, &headers, err_text));
205 }
206
207 let mut res = res;
208 let output_stream = async_stream::stream! {
209 let mut byte_buffer = Vec::<u8>::new();
210 let mut sse_event = SseEvent::default();
211 let mut pending_tools = BTreeMap::<usize, PendingToolBlock>::new();
212 let mut stop_reason: Option<String> = None;
213 let mut usage: Option<TokenUsage> = None;
214
215 loop {
216 let chunk = match res.chunk().await {
217 Ok(Some(chunk)) => chunk,
218 Ok(None) => break,
219 Err(e) => {
220 yield Err(map_transport_error(e));
221 return;
222 }
223 };
224
225 let lines = match take_sse_lines(&mut byte_buffer, &chunk) {
226 Ok(lines) => lines,
227 Err(e) => {
228 yield Err(e);
229 return;
230 }
231 };
232
233 for line in lines {
234 if line.is_empty() {
235 if let Some(data) = sse_event.data.take() {
236 match serde_json::from_str::<StreamEvent>(&data) {
237 Ok(event) => {
238 match handle_stream_event(
239 event,
240 &mut pending_tools,
241 &mut stop_reason,
242 &mut usage,
243 ) {
244 Ok(output) => {
245 for chunk in output.chunks {
246 yield Ok(chunk);
247 }
248 if output.done {
249 return;
250 }
251 }
252 Err(error) => {
253 yield Err(error);
254 return;
255 }
256 }
257 }
258 Err(e) => {
259 yield Err(ProviderError::transient(format!("failed to parse Anthropic SSE event: {e}; data={data}")));
260 return;
261 }
262 }
263 }
264 sse_event.event = None;
265 continue;
266 }
267
268 if let Some(event) = line.strip_prefix("event: ") {
269 sse_event.event = Some(event.to_string());
270 } else if let Some(data) = line.strip_prefix("data: ") {
271 match &mut sse_event.data {
272 Some(existing) => {
273 existing.push('\n');
274 existing.push_str(data);
275 }
276 None => sse_event.data = Some(data.to_string()),
277 }
278 }
279 }
280 }
281 yield Err(ProviderError::transient("Anthropic stream ended before message_stop"));
282 };
283
284 Ok(Box::pin(output_stream))
285 }
286}
287
288#[derive(Default)]
289struct SseEvent {
290 event: Option<String>,
291 data: Option<String>,
292}
293
294#[derive(Default)]
295struct StreamEventOutput {
296 chunks: Vec<ProviderStreamChunk>,
297 done: bool,
298}
299
300fn handle_stream_event(
301 event: StreamEvent,
302 pending_tools: &mut BTreeMap<usize, PendingToolBlock>,
303 stop_reason: &mut Option<String>,
304 usage: &mut Option<TokenUsage>,
305) -> Result<StreamEventOutput, ProviderError> {
306 let mut output = StreamEventOutput::default();
307 match event {
308 StreamEvent::MessageStart { message } => {
309 if let Some(event_usage) = message.usage {
310 merge_usage(usage, event_usage);
311 }
312 }
313 StreamEvent::ContentBlockStart {
314 index,
315 content_block,
316 } => {
317 if content_block.get("type").and_then(Value::as_str) == Some("tool_use") {
318 let block = pending_tools.entry(index).or_default();
319 block.id = content_block
320 .get("id")
321 .and_then(Value::as_str)
322 .map(str::to_string);
323 block.name = content_block
324 .get("name")
325 .and_then(Value::as_str)
326 .map(str::to_string);
327 if let Some(input) = content_block.get("input") {
328 if input.as_object().is_some_and(|object| !object.is_empty()) {
329 block.arguments = input.to_string();
330 }
331 }
332 }
333 }
334 StreamEvent::ContentBlockDelta { index, delta } => match delta {
335 StreamDelta::TextDelta { text } => {
336 if !text.is_empty() {
337 output.chunks.push(ProviderStreamChunk::ContentDelta(text));
338 }
339 }
340 StreamDelta::ThinkingDelta { thinking } => {
341 if !thinking.is_empty() {
342 output
343 .chunks
344 .push(ProviderStreamChunk::ThinkingDelta(thinking));
345 }
346 }
347 StreamDelta::InputJsonDelta { partial_json } => {
348 pending_tools
349 .entry(index)
350 .or_default()
351 .arguments
352 .push_str(&partial_json);
353 }
354 StreamDelta::SignatureDelta { signature } => {
355 let _ = signature;
356 }
357 },
358 StreamEvent::ContentBlockStop { index } => {
359 if let Some(block) = pending_tools.remove(&index) {
360 output.chunks.push(ProviderStreamChunk::ToolCallDelta {
361 index,
362 id: block.id,
363 name: block.name,
364 arguments_delta: if block.arguments.is_empty() {
365 "{}".to_string()
366 } else {
367 block.arguments
368 },
369 });
370 }
371 }
372 StreamEvent::MessageDelta {
373 delta,
374 usage: event_usage,
375 } => {
376 *stop_reason = delta.stop_reason;
377 if let Some(event_usage) = event_usage {
378 merge_usage(usage, event_usage);
379 }
380 }
381 StreamEvent::MessageStop => {
382 output.chunks.push(ProviderStreamChunk::Done {
383 finish_reason: map_stop_reason(stop_reason.as_deref()),
384 usage: usage.take(),
385 });
386 output.done = true;
387 }
388 StreamEvent::Ping => {}
389 StreamEvent::Error { error } => {
390 return Err(map_stream_error(error));
391 }
392 }
393 Ok(output)
394}
395
396fn merge_usage(usage: &mut Option<TokenUsage>, incoming: AnthropicUsage) {
397 let incoming = TokenUsage::from(incoming);
398 match usage {
399 Some(existing) => {
400 if incoming.input_tokens.is_some() {
401 existing.input_tokens = incoming.input_tokens;
402 }
403 if incoming.output_tokens.is_some() {
404 existing.output_tokens = incoming.output_tokens;
405 }
406 existing.total_tokens = match (existing.input_tokens, existing.output_tokens) {
407 (Some(input), Some(output)) => Some(input + output),
408 _ => None,
409 };
410 }
411 None => *usage = Some(incoming),
412 }
413}
414
415fn build_anthropic_payload(req: LLMRequest) -> AnthropicReq {
416 let (messages, extra_system) = transform_messages(req.messages);
417 let system = join_system(req.system, extra_system);
418 let tools = req.tools.iter().map(build_tool_schema).collect();
419
420 AnthropicReq {
421 model: req.model,
422 max_tokens: req.max_tokens.unwrap_or(DEFAULT_MAX_TOKENS),
423 system,
424 messages,
425 tools,
426 temperature: req.temperature,
427 stream: true,
428 }
429}
430
431fn join_system(primary: Option<String>, extra: Option<String>) -> Option<String> {
432 match (primary, extra) {
433 (Some(primary), Some(extra)) if !extra.is_empty() => Some(format!("{primary}\n\n{extra}")),
434 (Some(primary), _) => Some(primary),
435 (None, Some(extra)) if !extra.is_empty() => Some(extra),
436 _ => None,
437 }
438}
439
440fn transform_messages(messages: Vec<Message>) -> (Vec<AnthropicMessage>, Option<String>) {
441 let mut out = Vec::<AnthropicMessage>::new();
442 let mut tool_results = Vec::<AnthropicContentBlock>::new();
443 let mut system_parts = Vec::<String>::new();
444
445 for message in messages {
446 match message.role {
447 Role::System => system_parts.push(flatten_blocks(message.blocks)),
448 Role::User => {
449 flush_tool_results(&mut out, &mut tool_results);
450 push_message(
451 &mut out,
452 "user",
453 message
454 .blocks
455 .into_iter()
456 .filter_map(user_content_block)
457 .collect(),
458 );
459 }
460 Role::Assistant => {
461 flush_tool_results(&mut out, &mut tool_results);
462 push_message(
463 &mut out,
464 "assistant",
465 message
466 .blocks
467 .into_iter()
468 .filter_map(assistant_content_block)
469 .collect(),
470 );
471 }
472 Role::Tool => {
473 tool_results.extend(message.blocks.into_iter().filter_map(tool_result_block));
474 }
475 }
476 }
477 flush_tool_results(&mut out, &mut tool_results);
478
479 let system = if system_parts.is_empty() {
480 None
481 } else {
482 Some(system_parts.join("\n\n"))
483 };
484 (out, system)
485}
486
487fn flush_tool_results(
488 out: &mut Vec<AnthropicMessage>,
489 tool_results: &mut Vec<AnthropicContentBlock>,
490) {
491 if tool_results.is_empty() {
492 return;
493 }
494 let content = std::mem::take(tool_results);
495 push_message(out, "user", content);
496}
497
498fn push_message(out: &mut Vec<AnthropicMessage>, role: &str, content: Vec<AnthropicContentBlock>) {
499 if content.is_empty() {
500 return;
501 }
502 if let Some(last) = out.last_mut() {
503 if last.role == role {
504 last.content.extend(content);
505 return;
506 }
507 }
508 out.push(AnthropicMessage {
509 role: role.to_string(),
510 content,
511 });
512}
513
514fn user_content_block(block: ContentBlock) -> Option<AnthropicContentBlock> {
515 match block {
516 ContentBlock::Text { text } => Some(AnthropicContentBlock::Text { text }),
517 ContentBlock::ToolResult { .. } => tool_result_block(block),
518 ContentBlock::Thinking { .. } | ContentBlock::ToolUse { .. } => None,
519 }
520}
521
522fn assistant_content_block(block: ContentBlock) -> Option<AnthropicContentBlock> {
523 match block {
524 ContentBlock::Text { text } => Some(AnthropicContentBlock::Text { text }),
525 ContentBlock::ToolUse { id, name, input } => {
526 Some(AnthropicContentBlock::ToolUse { id, name, input })
527 }
528 ContentBlock::Thinking { .. } | ContentBlock::ToolResult { .. } => None,
529 }
530}
531
532fn tool_result_block(block: ContentBlock) -> Option<AnthropicContentBlock> {
533 let ContentBlock::ToolResult {
534 tool_use_id,
535 content,
536 is_error,
537 } = block
538 else {
539 return None;
540 };
541 Some(AnthropicContentBlock::ToolResult {
542 tool_use_id,
543 content: flatten_blocks(content),
544 is_error,
545 })
546}
547
548fn flatten_blocks(blocks: Vec<ContentBlock>) -> String {
549 let mut out = String::new();
550 for block in blocks {
551 match block {
552 ContentBlock::Text { text } => out.push_str(&text),
553 ContentBlock::Thinking { .. } => {}
554 ContentBlock::ToolUse { id, name, input } => {
555 out.push_str(&format!("[tool_use id={id} name={name} input={}]", input));
556 }
557 ContentBlock::ToolResult {
558 tool_use_id,
559 content,
560 is_error,
561 } => {
562 if is_error {
563 out.push_str("[tool_result error ");
564 } else {
565 out.push_str("[tool_result ");
566 }
567 out.push_str(&format!(
568 "tool_use_id={tool_use_id}]{}",
569 flatten_blocks(content)
570 ));
571 }
572 }
573 }
574 out
575}
576
577fn build_tool_schema(tool: &ToolDefinition) -> AnthropicTool {
578 AnthropicTool {
579 name: tool.name.clone(),
580 description: tool.desc.clone(),
581 input_schema: normalize_anthropic_tool_schema(tool.arguments.clone()),
582 }
583}
584
585fn normalize_anthropic_tool_schema(mut schema: Value) -> Value {
586 let definitions = schema
587 .get("definitions")
588 .or_else(|| schema.get("$defs"))
589 .cloned()
590 .unwrap_or_else(|| Value::Object(Map::new()));
591 inline_local_refs(&mut schema, &definitions);
592 simplify_tool_schema(&mut schema);
593 schema
594}
595
596fn inline_local_refs(value: &mut Value, definitions: &Value) {
597 match value {
598 Value::Object(map) => {
599 if let Some(ref_value) = map.get("$ref").and_then(Value::as_str) {
600 if let Some(mut resolved) = resolve_local_ref(ref_value, definitions) {
601 inline_local_refs(&mut resolved, definitions);
602 if let Value::Object(resolved_map) = &mut resolved {
603 for (key, value) in map.iter() {
604 if key != "$ref" {
605 resolved_map
606 .entry(key.clone())
607 .or_insert_with(|| value.clone());
608 }
609 }
610 }
611 *value = resolved;
612 return;
613 }
614 }
615 for child in map.values_mut() {
616 inline_local_refs(child, definitions);
617 }
618 }
619 Value::Array(items) => {
620 for item in items {
621 inline_local_refs(item, definitions);
622 }
623 }
624 _ => {}
625 }
626}
627
628fn resolve_local_ref(ref_value: &str, definitions: &Value) -> Option<Value> {
629 let key = ref_value
630 .strip_prefix("#/definitions/")
631 .or_else(|| ref_value.strip_prefix("#/$defs/"))?;
632 definitions.get(key).cloned()
633}
634
635fn simplify_tool_schema(value: &mut Value) {
636 match value {
637 Value::Object(map) => {
638 for child in map.values_mut() {
639 simplify_tool_schema(child);
640 }
641 if let Some(compressed) = compress_enum_union(map) {
642 *value = compressed;
643 return;
644 }
645 map.remove("$schema");
646 map.remove("$defs");
647 map.remove("definitions");
648 map.remove("title");
649 map.remove("format");
650 }
651 Value::Array(items) => {
652 for item in items {
653 simplify_tool_schema(item);
654 }
655 }
656 _ => {}
657 }
658}
659
660fn compress_enum_union(map: &Map<String, Value>) -> Option<Value> {
661 let variants = map
662 .get("oneOf")
663 .or_else(|| map.get("anyOf"))
664 .and_then(Value::as_array)?;
665
666 let mut enum_values = Vec::<Value>::new();
667 let mut has_null = false;
668 for variant in variants {
669 let variant = variant.as_object()?;
670 if is_null_schema(variant) {
671 has_null = true;
672 continue;
673 }
674 let values = variant.get("enum")?.as_array()?;
675 if values.is_empty() || !values.iter().all(|value| value.is_string()) {
676 return None;
677 }
678 enum_values.extend(values.iter().cloned());
679 }
680 if enum_values.is_empty() {
681 return None;
682 }
683 if has_null {
684 enum_values.push(Value::Null);
685 }
686
687 let mut compressed = Map::new();
688 if let Some(description) = map.get("description") {
689 compressed.insert("description".to_string(), description.clone());
690 }
691 compressed.insert(
692 "type".to_string(),
693 if has_null {
694 Value::Array(vec![
695 Value::String("string".to_string()),
696 Value::String("null".to_string()),
697 ])
698 } else {
699 Value::String("string".to_string())
700 },
701 );
702 compressed.insert("enum".to_string(), Value::Array(enum_values));
703 Some(Value::Object(compressed))
704}
705
706fn is_null_schema(map: &Map<String, Value>) -> bool {
707 matches!(map.get("type").and_then(Value::as_str), Some("null"))
708 || matches!(
709 map.get("enum").and_then(Value::as_array),
710 Some(values) if values.len() == 1 && values[0].is_null()
711 )
712}
713
714fn map_stop_reason(reason: Option<&str>) -> FinishReason {
715 match reason {
716 Some("end_turn") | Some("stop_sequence") => FinishReason::Stop,
717 Some("tool_use") => FinishReason::ToolCalls,
718 Some("max_tokens") => FinishReason::Length,
719 Some("refusal") => FinishReason::ContentFilter,
720 Some(other) => FinishReason::Unknown(other.to_string()),
721 None => FinishReason::Stop,
722 }
723}
724
725fn map_transport_error(error: reqwest::Error) -> ProviderError {
726 if error.is_timeout() || error.is_connect() || error.is_request() {
727 ProviderError::transient(error.to_string())
728 } else {
729 ProviderError::permanent(error.to_string())
730 }
731}
732
733fn map_status_error(status: StatusCode, headers: &HeaderMap, text: String) -> ProviderError {
734 let message = parse_error_message(&text).unwrap_or(text);
735 match status.as_u16() {
736 408 | 409 | 500..=599 => ProviderError::transient(message),
737 429 => ProviderError::rate_limited(message, retry_after_secs(headers)),
738 _ => ProviderError::permanent(message),
739 }
740}
741
742fn parse_error_message(text: &str) -> Option<String> {
743 let value = serde_json::from_str::<Value>(text).ok()?;
744 value
745 .get("error")
746 .and_then(|error| error.get("message"))
747 .and_then(Value::as_str)
748 .map(str::to_string)
749}
750
751fn retry_after_secs(headers: &HeaderMap) -> Option<u64> {
752 headers
753 .get(reqwest::header::RETRY_AFTER)
754 .and_then(|value| value.to_str().ok())
755 .and_then(|value| value.parse::<u64>().ok())
756}
757
758fn map_stream_error(error: AnthropicErrorBody) -> ProviderError {
759 match error.kind.as_str() {
760 "overloaded_error" | "rate_limit_error" => ProviderError::transient(error.message),
761 _ => ProviderError::permanent(error.message),
762 }
763}
764
765fn take_sse_lines(buffer: &mut Vec<u8>, chunk: &[u8]) -> Result<Vec<String>, ProviderError> {
766 buffer.extend_from_slice(chunk);
767 let mut lines = Vec::new();
768 while let Some(pos) = buffer.iter().position(|byte| *byte == b'\n') {
769 let mut line = buffer.drain(..=pos).collect::<Vec<_>>();
770 if line.last() == Some(&b'\n') {
771 line.pop();
772 }
773 if line.last() == Some(&b'\r') {
774 line.pop();
775 }
776 lines
777 .push(String::from_utf8(line).map_err(|e| {
778 ProviderError::transient(format!("invalid UTF-8 in SSE line: {e}"))
779 })?);
780 }
781 Ok(lines)
782}
783
784#[cfg(test)]
785mod tests {
786 use super::*;
787 use serde_json::json;
788
789 #[test]
790 fn transforms_tool_results_into_user_blocks() {
791 let req = LLMRequest {
792 model: "claude-test".to_string(),
793 system: Some("system".to_string()),
794 messages: vec![
795 Message {
796 role: Role::Assistant,
797 blocks: vec![ContentBlock::ToolUse {
798 id: "toolu_1".to_string(),
799 name: "read_file".to_string(),
800 input: json!({ "path": "Cargo.toml" }),
801 }],
802 },
803 Message {
804 role: Role::Tool,
805 blocks: vec![ContentBlock::ToolResult {
806 tool_use_id: "toolu_1".to_string(),
807 content: vec![ContentBlock::Text {
808 text: "ok".to_string(),
809 }],
810 is_error: false,
811 }],
812 },
813 ],
814 tools: vec![],
815 temperature: None,
816 max_tokens: Some(128),
817 };
818
819 let payload = build_anthropic_payload(req);
820
821 assert_eq!(payload.system.as_deref(), Some("system"));
822 assert_eq!(payload.messages.len(), 2);
823 assert_eq!(payload.messages[0].role, "assistant");
824 assert!(matches!(
825 &payload.messages[0].content[0],
826 AnthropicContentBlock::ToolUse { id, .. } if id == "toolu_1"
827 ));
828 assert_eq!(payload.messages[1].role, "user");
829 assert!(matches!(
830 &payload.messages[1].content[0],
831 AnthropicContentBlock::ToolResult { tool_use_id, content, is_error }
832 if tool_use_id == "toolu_1" && content == "ok" && !is_error
833 ));
834 }
835
836 #[test]
837 fn anthropic_tool_schema_uses_input_schema_without_refs() {
838 let tool = ToolDefinition {
839 name: "write_file".to_string(),
840 desc: "edit files".to_string(),
841 arguments: json!({
842 "$schema": "http://json-schema.org/draft-07/schema#",
843 "title": "WriteFileArgs",
844 "type": "object",
845 "definitions": {
846 "Mode": {
847 "oneOf": [
848 { "enum": ["write"], "type": "string" },
849 { "enum": ["append"], "type": "string" }
850 ]
851 }
852 },
853 "properties": {
854 "mode": {
855 "description": "edit mode",
856 "anyOf": [
857 { "$ref": "#/definitions/Mode" },
858 { "type": "null" }
859 ]
860 }
861 }
862 }),
863 };
864
865 let schema = build_tool_schema(&tool);
866 let input_schema = schema.input_schema;
867 let blob = input_schema.to_string();
868
869 assert!(!blob.contains("$schema"));
870 assert!(!blob.contains("definitions"));
871 assert!(!blob.contains("$ref"));
872 assert!(!blob.contains("title"));
873 assert_eq!(
874 input_schema["properties"]["mode"],
875 json!({
876 "description": "edit mode",
877 "type": ["string", "null"],
878 "enum": ["write", "append", null]
879 })
880 );
881 }
882
883 #[test]
884 fn parses_streaming_tool_use_events() {
885 let events = [
886 r#"{"type":"message_start","message":{"usage":{"input_tokens":1,"output_tokens":1}}}"#,
887 r#"{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}"#,
888 r#"{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"I'll check"}}"#,
889 r#"{"type":"content_block_stop","index":0}"#,
890 r#"{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_1","name":"read_file","input":{}}}"#,
891 r#"{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"path\":"}}"#,
892 r#"{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"\"Cargo.toml\"}"}}"#,
893 r#"{"type":"content_block_stop","index":1}"#,
894 r#"{"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":9}}"#,
895 r#"{"type":"message_stop"}"#,
896 ];
897
898 let mut pending_tools = BTreeMap::<usize, PendingToolBlock>::new();
899 let mut stop_reason = None;
900 let mut usage = None;
901 let mut chunks = Vec::<ProviderStreamChunk>::new();
902 for event in events {
903 let event = serde_json::from_str::<StreamEvent>(event).expect("event should parse");
904 let output =
905 handle_stream_event(event, &mut pending_tools, &mut stop_reason, &mut usage)
906 .expect("event should map");
907 chunks.extend(output.chunks);
908 }
909
910 assert!(matches!(
911 &chunks[0],
912 ProviderStreamChunk::ContentDelta(text) if text == "I'll check"
913 ));
914 assert!(matches!(
915 &chunks[1],
916 ProviderStreamChunk::ToolCallDelta { id: Some(id), name: Some(name), arguments_delta, .. }
917 if id == "toolu_1" && name == "read_file" && arguments_delta == "{\"path\":\"Cargo.toml\"}"
918 ));
919 assert!(matches!(
920 &chunks[2],
921 ProviderStreamChunk::Done { finish_reason: FinishReason::ToolCalls, usage: Some(usage) }
922 if usage.input_tokens == Some(1)
923 && usage.output_tokens == Some(9)
924 && usage.total_tokens == Some(10)
925 ));
926 }
927}