gateway_core/
anthropic.rs

1use std::collections::BTreeMap;
2
3use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
4use serde::{Deserialize, Serialize};
5use serde_json::{Map, Value, json};
6
7use crate::{
8    Capabilities, ModelUsage, ProviderAdapter, ProviderError, ProviderRequest, ProviderResponse,
9    ProviderStreamDecoder, ProviderStreamEvent, SseEvent, Surface, is_rate_limit_payload,
10};
11
12const DEFAULT_MAX_TOKENS: u64 = 4096;
13const MIN_THINKING_BUDGET: u64 = 1024;
14pub const REASONING_DETAIL_FORMAT: &str = "anthropic-claude-v1";
15
16#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
17pub struct SignedThinking {
18    pub thinking: String,
19    pub signature: String,
20}
21
22pub fn encrypted_reasoning_detail(tool_call_id: &str, blocks: &[SignedThinking]) -> Value {
23    let data = BASE64.encode(serde_json::to_vec(blocks).unwrap_or_default());
24    json!({
25        "type": "reasoning.encrypted",
26        "id": tool_call_id,
27        "data": data,
28        "format": REASONING_DETAIL_FORMAT,
29    })
30}
31
32pub fn signed_thinking_from_details(message: &Value) -> Vec<SignedThinking> {
33    message
34        .get("reasoning_details")
35        .and_then(Value::as_array)
36        .into_iter()
37        .flatten()
38        .filter(|detail| {
39            detail.get("type").and_then(Value::as_str) == Some("reasoning.encrypted")
40                && detail
41                    .get("format")
42                    .and_then(Value::as_str)
43                    .is_none_or(|format| format == REASONING_DETAIL_FORMAT)
44        })
45        .filter_map(|detail| detail.get("data").and_then(Value::as_str))
46        .filter_map(|data| BASE64.decode(data).ok())
47        .filter_map(|decoded| serde_json::from_slice::<Vec<SignedThinking>>(&decoded).ok())
48        .flatten()
49        .collect()
50}
51
52pub fn thinking_blocks_from_details(message: &Value) -> Vec<Value> {
53    signed_thinking_from_details(message)
54        .into_iter()
55        .map(|block| {
56            json!({
57                "type": "thinking",
58                "thinking": block.thinking,
59                "signature": block.signature,
60            })
61        })
62        .collect()
63}
64
65pub struct AnthropicAdapter;
66
67impl AnthropicAdapter {
68    pub const VERSION: &'static str = "2023-06-01";
69
70    pub fn new() -> Self {
71        Self
72    }
73}
74
75impl Default for AnthropicAdapter {
76    fn default() -> Self {
77        Self::new()
78    }
79}
80
81impl ProviderAdapter for AnthropicAdapter {
82    fn name(&self) -> &'static str {
83        "anthropic"
84    }
85
86    fn capabilities(&self) -> Capabilities {
87        Capabilities {
88            chat: true,
89            responses: false,
90            vision: true,
91            reasoning: true,
92            embeddings: false,
93        }
94    }
95
96    fn encode_request(
97        &self,
98        surface: Surface,
99        request: ProviderRequest,
100    ) -> Result<Value, ProviderError> {
101        if surface != Surface::ChatCompletions {
102            return Err(ProviderError::Unsupported(
103                "Anthropic supports chat completions through the native Messages adapter".into(),
104            ));
105        }
106        Ok(build_request(&request.model, &request.body))
107    }
108
109    fn decode_response(
110        &self,
111        surface: Surface,
112        response: Value,
113    ) -> Result<ProviderResponse, ProviderError> {
114        if surface != Surface::ChatCompletions {
115            return Err(ProviderError::Unsupported("Anthropic Responses API".into()));
116        }
117        let usage = anthropic_usage(response.get("usage").unwrap_or(&Value::Null));
118        let mut text = String::new();
119        let mut reasoning = String::new();
120        let mut signed_thinking = Vec::new();
121        let mut tool_calls = Vec::new();
122        for block in response
123            .get("content")
124            .and_then(Value::as_array)
125            .into_iter()
126            .flatten()
127        {
128            match block.get("type").and_then(Value::as_str) {
129                Some("text") => text.push_str(block.get("text").and_then(Value::as_str).unwrap_or_default()),
130                Some("thinking") => {
131                    let thinking = block
132                        .get("thinking")
133                        .and_then(Value::as_str)
134                        .unwrap_or_default();
135                    reasoning.push_str(thinking);
136                    if let Some(signature) = block.get("signature").and_then(Value::as_str) {
137                        signed_thinking.push(SignedThinking {
138                            thinking: thinking.to_owned(),
139                            signature: signature.to_owned(),
140                        });
141                    }
142                }
143                Some("tool_use") => tool_calls.push(json!({
144                    "id": block.get("id").cloned().unwrap_or(Value::Null),
145                    "type": "function",
146                    "function": {
147                        "name": block.get("name").cloned().unwrap_or(Value::Null),
148                        "arguments": serde_json::to_string(block.get("input").unwrap_or(&Value::Null)).unwrap_or_else(|_| "{}".into())
149                    }
150                })),
151                _ => {}
152            }
153        }
154        let finish_reason = match response.get("stop_reason").and_then(Value::as_str) {
155            Some("tool_use") => "tool_calls",
156            Some("max_tokens") => "length",
157            _ => "stop",
158        };
159        let mut message = json!({
160            "role": "assistant",
161            "content": if text.is_empty() { Value::Null } else { json!(text) }
162        });
163        if !reasoning.is_empty() {
164            message["reasoning_content"] = json!(reasoning);
165        }
166        if !tool_calls.is_empty() {
167            if !signed_thinking.is_empty() {
168                let tool_call_id = tool_calls[0]
169                    .get("id")
170                    .and_then(Value::as_str)
171                    .unwrap_or_default();
172                message["reasoning_details"] =
173                    json!([encrypted_reasoning_detail(tool_call_id, &signed_thinking)]);
174            }
175            message["tool_calls"] = Value::Array(tool_calls);
176        }
177        Ok(ProviderResponse {
178            body: json!({
179                "id": response.get("id").cloned().unwrap_or(Value::Null),
180                "object": "chat.completion",
181                "model": response.get("model").cloned().unwrap_or(Value::Null),
182                "choices": [{ "index": 0, "message": message, "finish_reason": finish_reason }],
183                "usage": openai_usage(usage)
184            }),
185            usage,
186        })
187    }
188
189    fn stream_decoder(
190        &self,
191        surface: Surface,
192    ) -> Result<Box<dyn ProviderStreamDecoder>, ProviderError> {
193        if surface != Surface::ChatCompletions {
194            return Err(ProviderError::Unsupported("Anthropic Responses API".into()));
195        }
196        Ok(Box::new(AnthropicStreamDecoder::default()))
197    }
198}
199
200#[derive(Default)]
201struct AnthropicStreamDecoder {
202    usage: ModelUsage,
203    first_delta: bool,
204    pending_thinking: BTreeMap<u64, SignedThinking>,
205    completed_thinking: Vec<SignedThinking>,
206    tool_blocks: BTreeMap<u64, u64>,
207    next_tool_index: u64,
208    thinking_attached: bool,
209    saw_tool_call: bool,
210    stop_reason: Option<String>,
211    terminal_emitted: bool,
212    done: bool,
213}
214
215impl AnthropicStreamDecoder {
216    fn data(&mut self, mut delta: Value) -> ProviderStreamEvent {
217        if !self.first_delta {
218            if let Some(delta) = delta.as_object_mut() {
219                delta.insert("role".into(), json!("assistant"));
220            }
221            self.first_delta = true;
222        }
223        ProviderStreamEvent::Data {
224            event: None,
225            data: json!({
226                "choices": [{
227                    "index": 0,
228                    "delta": delta,
229                    "finish_reason": null
230                }]
231            }),
232        }
233    }
234
235    fn terminal(&self) -> ProviderStreamEvent {
236        let finish_reason = map_stop_reason(self.stop_reason.as_deref(), self.saw_tool_call);
237        ProviderStreamEvent::Data {
238            event: None,
239            data: json!({
240                "choices": [{
241                    "index": 0,
242                    "delta": {},
243                    "finish_reason": finish_reason
244                }],
245                "usage": openai_usage(self.usage)
246            }),
247        }
248    }
249}
250
251impl ProviderStreamDecoder for AnthropicStreamDecoder {
252    fn decode(&mut self, event: SseEvent) -> Result<Vec<ProviderStreamEvent>, ProviderError> {
253        let data: Value = serde_json::from_str(&event.data)
254            .map_err(|error| ProviderError::InvalidStream(error.to_string()))?;
255        let kind = event
256            .event
257            .as_deref()
258            .or_else(|| data.get("type").and_then(Value::as_str));
259        match kind {
260            Some("message_start") => {
261                merge_anthropic_usage(
262                    &mut self.usage,
263                    data.pointer("/message/usage").unwrap_or(&Value::Null),
264                );
265                Ok(Vec::new())
266            }
267            Some("content_block_start") => {
268                let index = data.get("index").and_then(Value::as_u64).unwrap_or(0);
269                let block = data.get("content_block").unwrap_or(&Value::Null);
270                match block.get("type").and_then(Value::as_str) {
271                    Some("thinking") => {
272                        self.pending_thinking.insert(
273                            index,
274                            SignedThinking {
275                                thinking: block
276                                    .get("thinking")
277                                    .and_then(Value::as_str)
278                                    .unwrap_or_default()
279                                    .to_owned(),
280                                signature: block
281                                    .get("signature")
282                                    .and_then(Value::as_str)
283                                    .unwrap_or_default()
284                                    .to_owned(),
285                            },
286                        );
287                        Ok(Vec::new())
288                    }
289                    Some("tool_use") => {
290                        let tool_index = self.next_tool_index;
291                        self.next_tool_index = self.next_tool_index.saturating_add(1);
292                        self.tool_blocks.insert(index, tool_index);
293                        self.saw_tool_call = true;
294                        let id = block.get("id").and_then(Value::as_str).unwrap_or_default();
295                        let mut delta = json!({
296                            "tool_calls": [{
297                                "index": tool_index,
298                                "id": id,
299                                "type": "function",
300                                "function": {
301                                    "name": block
302                                        .get("name")
303                                        .and_then(Value::as_str)
304                                        .unwrap_or_default(),
305                                    "arguments": ""
306                                }
307                            }]
308                        });
309                        if !self.thinking_attached && !self.completed_thinking.is_empty() {
310                            delta["reasoning_details"] =
311                                json!([encrypted_reasoning_detail(id, &self.completed_thinking)]);
312                            self.thinking_attached = true;
313                        }
314                        Ok(vec![self.data(delta)])
315                    }
316                    _ => Ok(Vec::new()),
317                }
318            }
319            Some("content_block_delta") => {
320                let index = data.get("index").and_then(Value::as_u64).unwrap_or(0);
321                let delta = data.get("delta").unwrap_or(&Value::Null);
322                match delta.get("type").and_then(Value::as_str) {
323                    Some("text_delta") => Ok(vec![self.data(json!({
324                        "content": delta.get("text").and_then(Value::as_str).unwrap_or_default()
325                    }))]),
326                    Some("thinking_delta") => {
327                        let thinking = delta
328                            .get("thinking")
329                            .and_then(Value::as_str)
330                            .unwrap_or_default();
331                        self.pending_thinking
332                            .entry(index)
333                            .or_insert_with(|| SignedThinking {
334                                thinking: String::new(),
335                                signature: String::new(),
336                            })
337                            .thinking
338                            .push_str(thinking);
339                        Ok(vec![self.data(json!({ "reasoning_content": thinking }))])
340                    }
341                    Some("signature_delta") => {
342                        let signature = delta
343                            .get("signature")
344                            .and_then(Value::as_str)
345                            .unwrap_or_default();
346                        self.pending_thinking
347                            .entry(index)
348                            .or_insert_with(|| SignedThinking {
349                                thinking: String::new(),
350                                signature: String::new(),
351                            })
352                            .signature
353                            .push_str(signature);
354                        Ok(Vec::new())
355                    }
356                    Some("input_json_delta") => {
357                        let Some(tool_index) = self.tool_blocks.get(&index).copied() else {
358                            return Err(ProviderError::InvalidStream(format!(
359                                "input_json_delta for unknown content block {index}"
360                            )));
361                        };
362                        Ok(vec![self.data(json!({
363                            "tool_calls": [{
364                                "index": tool_index,
365                                "function": {
366                                    "arguments": delta
367                                        .get("partial_json")
368                                        .and_then(Value::as_str)
369                                        .unwrap_or_default()
370                                }
371                            }]
372                        }))])
373                    }
374                    _ => Ok(Vec::new()),
375                }
376            }
377            Some("content_block_stop") => {
378                let index = data.get("index").and_then(Value::as_u64).unwrap_or(0);
379                if let Some(thinking) = self.pending_thinking.remove(&index)
380                    && !thinking.signature.is_empty()
381                {
382                    self.completed_thinking.push(thinking);
383                }
384                self.tool_blocks.remove(&index);
385                Ok(Vec::new())
386            }
387            Some("message_delta") => {
388                merge_anthropic_usage(&mut self.usage, data.get("usage").unwrap_or(&Value::Null));
389                if let Some(reason) = data.pointer("/delta/stop_reason").and_then(Value::as_str) {
390                    self.stop_reason = Some(reason.to_owned());
391                }
392                self.terminal_emitted = true;
393                Ok(vec![self.terminal()])
394            }
395            Some("message_stop") => {
396                let mut events = Vec::with_capacity(2);
397                if !self.terminal_emitted {
398                    events.push(self.terminal());
399                    self.terminal_emitted = true;
400                }
401                if !self.done {
402                    events.push(ProviderStreamEvent::Done(self.usage));
403                    self.done = true;
404                }
405                Ok(events)
406            }
407            Some("error") => {
408                let message = data
409                    .pointer("/error/message")
410                    .and_then(Value::as_str)
411                    .unwrap_or("Anthropic stream error")
412                    .to_owned();
413                if crate::is_rate_limit_payload(&data) {
414                    Err(ProviderError::RateLimitedStream(message))
415                } else {
416                    Err(ProviderError::InvalidStream(message))
417                }
418            }
419            _ => Ok(Vec::new()),
420        }
421    }
422
423    fn finish(&mut self) -> Result<Vec<ProviderStreamEvent>, ProviderError> {
424        if self.done {
425            return Ok(Vec::new());
426        }
427        let mut events = Vec::with_capacity(2);
428        if !self.terminal_emitted {
429            events.push(self.terminal());
430            self.terminal_emitted = true;
431        }
432        events.push(ProviderStreamEvent::Done(self.usage));
433        self.done = true;
434        Ok(events)
435    }
436}
437
438/// Usage from a native Messages response, mapped onto the canonical
439/// [`ModelUsage`]. Anthropic reports `input_tokens`/`output_tokens` alongside
440/// separate cache counters, so a native response needs this rather than the
441/// OpenAI-shaped `prompt_tokens`/`completion_tokens` reader.
442pub fn native_message_usage(response: &Value) -> ModelUsage {
443    anthropic_usage(response.get("usage").unwrap_or(&Value::Null))
444}
445
446/// Decoder for a native Messages stream that is relayed to the caller
447/// unchanged: every event is handed back verbatim, and the only work done is
448/// folding Anthropic's split usage reporting — input tokens on `message_start`,
449/// output tokens on `message_delta` — into one [`ModelUsage`].
450#[derive(Default)]
451pub struct NativeMessagesDecoder {
452    usage: ModelUsage,
453    done: bool,
454}
455
456impl NativeMessagesDecoder {
457    pub fn new() -> Self {
458        Self::default()
459    }
460}
461
462impl ProviderStreamDecoder for NativeMessagesDecoder {
463    fn decode(&mut self, event: SseEvent) -> Result<Vec<ProviderStreamEvent>, ProviderError> {
464        let data: Value = serde_json::from_str(&event.data)
465            .map_err(|error| ProviderError::InvalidStream(error.to_string()))?;
466        let kind = event
467            .event
468            .clone()
469            .or_else(|| data.get("type").and_then(Value::as_str).map(str::to_owned));
470        if is_rate_limit_payload(&data) {
471            return Err(ProviderError::RateLimitedStream(
472                data.pointer("/error/message")
473                    .and_then(Value::as_str)
474                    .unwrap_or("Anthropic stream rate limited")
475                    .to_owned(),
476            ));
477        }
478        match kind.as_deref() {
479            Some("message_start") => merge_anthropic_usage(
480                &mut self.usage,
481                data.pointer("/message/usage").unwrap_or(&Value::Null),
482            ),
483            Some("message_delta") => {
484                merge_anthropic_usage(&mut self.usage, data.get("usage").unwrap_or(&Value::Null))
485            }
486            Some("error") => {
487                return Err(ProviderError::InvalidStream(
488                    data.pointer("/error/message")
489                        .and_then(Value::as_str)
490                        .unwrap_or("Anthropic stream error")
491                        .to_owned(),
492                ));
493            }
494            _ => {}
495        }
496        let terminal = kind.as_deref() == Some("message_stop");
497        let mut events = vec![ProviderStreamEvent::Data { event: kind, data }];
498        if terminal && !self.done {
499            self.done = true;
500            events.push(ProviderStreamEvent::Done(self.usage));
501        }
502        Ok(events)
503    }
504
505    fn finish(&mut self) -> Result<Vec<ProviderStreamEvent>, ProviderError> {
506        if self.done {
507            return Ok(Vec::new());
508        }
509        self.done = true;
510        Ok(vec![ProviderStreamEvent::Done(self.usage)])
511    }
512}
513
514fn build_request(model: &str, body: &Value) -> Value {
515    let max_tokens = body
516        .get("max_tokens")
517        .or_else(|| body.get("max_completion_tokens"))
518        .and_then(Value::as_u64)
519        .unwrap_or(DEFAULT_MAX_TOKENS);
520    let thinking = thinking_budget(body, max_tokens);
521    let mut request = Map::from_iter([
522        ("model".into(), json!(model)),
523        ("max_tokens".into(), json!(max_tokens)),
524        ("messages".into(), Value::Array(Vec::new())),
525    ]);
526    let (system, messages) = build_messages(body, thinking.is_some());
527    request.insert("messages".into(), Value::Array(messages));
528    if !system.is_empty() {
529        request.insert("system".into(), json!(system));
530    }
531    if let Some(tools) = parse_tools(body) {
532        request.insert("tools".into(), Value::Array(tools));
533        if let Some(choice) = parse_tool_choice(body) {
534            request.insert("tool_choice".into(), choice);
535        }
536    }
537    if let Some(budget) = thinking {
538        request.insert(
539            "thinking".into(),
540            json!({ "type": "enabled", "budget_tokens": budget }),
541        );
542    } else {
543        if let Some(value) = body.get("temperature") {
544            request.insert("temperature".into(), value.clone());
545        }
546        if let Some(value) = body.get("top_p") {
547            request.insert("top_p".into(), value.clone());
548        }
549    }
550    if let Some(stops) = parse_stop_sequences(body) {
551        request.insert("stop_sequences".into(), Value::Array(stops));
552    }
553    request.insert(
554        "stream".into(),
555        body.get("stream").cloned().unwrap_or(json!(false)),
556    );
557    Value::Object(request)
558}
559
560fn thinking_budget(body: &Value, max_tokens: u64) -> Option<u64> {
561    let fraction = match body.get("reasoning_effort").and_then(Value::as_str)? {
562        "minimal" => 10,
563        "low" => 20,
564        "medium" => 50,
565        "high" | "xhigh" => 80,
566        _ => return None,
567    };
568    let cap = max_tokens.checked_sub(MIN_THINKING_BUDGET)?;
569    if cap < MIN_THINKING_BUDGET {
570        return None;
571    }
572    let scaled = max_tokens.saturating_mul(fraction) / 100;
573    Some(scaled.clamp(MIN_THINKING_BUDGET, cap))
574}
575
576fn parse_tools(body: &Value) -> Option<Vec<Value>> {
577    let tools = body
578        .get("tools")
579        .and_then(Value::as_array)?
580        .iter()
581        .filter_map(|tool| {
582            let function = tool.get("function")?;
583            let name = function.get("name")?.as_str()?;
584            let mut translated = Map::new();
585            translated.insert("name".into(), json!(name));
586            if let Some(description) = function.get("description").and_then(Value::as_str) {
587                translated.insert("description".into(), json!(description));
588            }
589            translated.insert(
590                "input_schema".into(),
591                function
592                    .get("parameters")
593                    .cloned()
594                    .unwrap_or_else(|| json!({ "type": "object" })),
595            );
596            Some(Value::Object(translated))
597        })
598        .collect::<Vec<_>>();
599    (!tools.is_empty()).then_some(tools)
600}
601
602fn parse_tool_choice(body: &Value) -> Option<Value> {
603    match body.get("tool_choice")? {
604        Value::String(choice) => match choice.as_str() {
605            "auto" => Some(json!({ "type": "auto" })),
606            "required" => Some(json!({ "type": "any" })),
607            "none" => Some(json!({ "type": "none" })),
608            _ => None,
609        },
610        choice @ Value::Object(_) => Some(json!({
611            "type": "tool",
612            "name": choice.pointer("/function/name")?.as_str()?
613        })),
614        _ => None,
615    }
616}
617
618fn parse_stop_sequences(body: &Value) -> Option<Vec<Value>> {
619    match body.get("stop")? {
620        Value::String(stop) => Some(vec![json!(stop)]),
621        Value::Array(stops) => {
622            let stops = stops
623                .iter()
624                .filter(|stop| stop.is_string())
625                .cloned()
626                .collect::<Vec<_>>();
627            (!stops.is_empty()).then_some(stops)
628        }
629        _ => None,
630    }
631}
632
633fn build_messages(body: &Value, thinking_enabled: bool) -> (String, Vec<Value>) {
634    let mut system = String::new();
635    let mut messages: Vec<Value> = Vec::new();
636    for message in body
637        .get("messages")
638        .and_then(Value::as_array)
639        .into_iter()
640        .flatten()
641    {
642        let role = message
643            .get("role")
644            .and_then(Value::as_str)
645            .unwrap_or("user");
646        if matches!(role, "system" | "developer") {
647            let text = content_text(message.get("content"));
648            if !text.is_empty() {
649                if !system.is_empty() {
650                    system.push('\n');
651                }
652                system.push_str(&text);
653            }
654            continue;
655        }
656        let output_role = if role == "assistant" {
657            "assistant"
658        } else {
659            "user"
660        };
661        let mut blocks = if role == "tool" {
662            vec![json!({
663                "type": "tool_result",
664                "tool_use_id": message.get("tool_call_id").cloned().unwrap_or(Value::Null),
665                "content": content_text(message.get("content"))
666            })]
667        } else {
668            content_blocks(message.get("content"))
669        };
670        if role == "assistant" {
671            if thinking_enabled {
672                let mut thinking = thinking_blocks_from_details(message);
673                thinking.append(&mut blocks);
674                blocks = thinking;
675            }
676            for call in message
677                .get("tool_calls")
678                .and_then(Value::as_array)
679                .into_iter()
680                .flatten()
681            {
682                let function = call.get("function").unwrap_or(&Value::Null);
683                let input = function
684                    .get("arguments")
685                    .and_then(Value::as_str)
686                    .and_then(|value| serde_json::from_str(value).ok())
687                    .unwrap_or_else(|| json!({}));
688                blocks.push(json!({
689                    "type": "tool_use",
690                    "id": call.get("id").cloned().unwrap_or(Value::Null),
691                    "name": function.get("name").cloned().unwrap_or(Value::Null),
692                    "input": input
693                }));
694            }
695        }
696        if blocks.is_empty() {
697            continue;
698        }
699        if let Some(previous) = messages.last_mut()
700            && previous.get("role").and_then(Value::as_str) == Some(output_role)
701            && let Some(content) = previous.get_mut("content").and_then(Value::as_array_mut)
702        {
703            content.extend(blocks);
704        } else {
705            messages.push(json!({ "role": output_role, "content": blocks }));
706        }
707    }
708    (system, messages)
709}
710
711fn content_text(content: Option<&Value>) -> String {
712    match content {
713        Some(Value::String(value)) => value.clone(),
714        Some(Value::Array(parts)) => parts
715            .iter()
716            .filter_map(|part| part.get("text").and_then(Value::as_str))
717            .collect(),
718        _ => String::new(),
719    }
720}
721
722fn content_blocks(content: Option<&Value>) -> Vec<Value> {
723    match content {
724        Some(Value::String(value)) if !value.is_empty() => {
725            vec![json!({ "type": "text", "text": value })]
726        }
727        Some(Value::Array(parts)) => parts
728            .iter()
729            .filter_map(|part| match part.get("type").and_then(Value::as_str) {
730                Some("text") => Some(json!({ "type": "text", "text": part.get("text")? })),
731                Some("image_url") => {
732                    let url = part.pointer("/image_url/url").and_then(Value::as_str)?;
733                    Some(image_block(url))
734                }
735                _ => None,
736            })
737            .collect(),
738        _ => Vec::new(),
739    }
740}
741
742fn image_block(url: &str) -> Value {
743    if let Some(data) = url.strip_prefix("data:")
744        && let Some((metadata, payload)) = data.split_once(',')
745    {
746        return json!({
747            "type": "image",
748            "source": {
749                "type": "base64",
750                "media_type": metadata.split(';').next().unwrap_or("image/png"),
751                "data": payload
752            }
753        });
754    }
755    json!({ "type": "image", "source": { "type": "url", "url": url } })
756}
757
758fn map_stop_reason(stop_reason: Option<&str>, saw_tool_call: bool) -> &'static str {
759    match stop_reason {
760        Some("tool_use") => "tool_calls",
761        Some("max_tokens") => "length",
762        Some(_) | None if saw_tool_call => "tool_calls",
763        _ => "stop",
764    }
765}
766
767fn openai_usage(usage: ModelUsage) -> Value {
768    // OpenAI has no cache-write field, so the read/write distinction is lost
769    // when translating the disjoint provider counters into its wire format.
770    json!({
771        "prompt_tokens": usage.input_tokens
772            .saturating_add(usage.cache_read_tokens)
773            .saturating_add(usage.cache_write_tokens),
774        "completion_tokens": usage.output_tokens,
775        "total_tokens": usage.total_tokens(),
776        "completion_tokens_details": {
777            "reasoning_tokens": usage.reasoning_tokens
778        },
779        "prompt_tokens_details": {
780            "cached_tokens": usage.cache_read_tokens,
781            "cache_write_tokens": usage.cache_write_tokens
782        }
783    })
784}
785
786fn merge_anthropic_usage(usage: &mut ModelUsage, value: &Value) {
787    let next = anthropic_usage(value);
788    if value.get("input_tokens").is_some() {
789        usage.input_tokens = next.input_tokens;
790    }
791    if value.get("output_tokens").is_some() {
792        usage.output_tokens = next.output_tokens;
793    }
794    if value.get("reasoning_tokens").is_some() {
795        usage.reasoning_tokens = next.reasoning_tokens;
796    }
797    if value.get("cache_read_input_tokens").is_some() {
798        usage.cache_read_tokens = next.cache_read_tokens;
799    }
800    if value.get("cache_creation_input_tokens").is_some() {
801        usage.cache_write_tokens = next.cache_write_tokens;
802    }
803}
804
805fn anthropic_usage(value: &Value) -> ModelUsage {
806    ModelUsage {
807        input_tokens: value
808            .get("input_tokens")
809            .and_then(Value::as_u64)
810            .unwrap_or_default(),
811        output_tokens: value
812            .get("output_tokens")
813            .and_then(Value::as_u64)
814            .unwrap_or_default(),
815        reasoning_tokens: value
816            .get("reasoning_tokens")
817            .and_then(Value::as_u64)
818            .unwrap_or_default(),
819        cache_read_tokens: value
820            .get("cache_read_input_tokens")
821            .and_then(Value::as_u64)
822            .unwrap_or_default(),
823        cache_write_tokens: value
824            .get("cache_creation_input_tokens")
825            .and_then(Value::as_u64)
826            .unwrap_or_default(),
827    }
828}
829
830#[cfg(test)]
831mod tests {
832    use super::*;
833    use crate::provider::chat_usage;
834
835    #[test]
836    fn cached_usage_round_trips_through_both_provider_conventions() {
837        for wire in [
838            json!({
839                "prompt_tokens": 22,
840                "completion_tokens": 9,
841                "prompt_tokens_details": { "cached_tokens": 3 }
842            }),
843            json!({
844                "input_tokens": 19,
845                "output_tokens": 9,
846                "cache_read_input_tokens": 3
847            }),
848        ] {
849            let usage = chat_usage(&wire);
850            assert_eq!(chat_usage(&openai_usage(usage)), usage);
851        }
852    }
853
854    #[test]
855    fn cache_writes_fold_into_openai_prompt_total() {
856        let usage = ModelUsage {
857            input_tokens: 19,
858            output_tokens: 9,
859            cache_read_tokens: 3,
860            cache_write_tokens: 4,
861            ..ModelUsage::default()
862        };
863        let wire = openai_usage(usage);
864        assert_eq!(wire["prompt_tokens"], 26);
865        assert_eq!(wire["completion_tokens"], 9);
866        assert_eq!(wire["total_tokens"], 35);
867        assert_eq!(
868            wire["total_tokens"],
869            wire["prompt_tokens"].as_u64().unwrap() + wire["completion_tokens"].as_u64().unwrap()
870        );
871    }
872
873    #[test]
874    fn translates_openai_messages_tools_and_usage() {
875        let adapter = AnthropicAdapter::new();
876        let request = adapter
877            .encode_request(
878                Surface::ChatCompletions,
879                ProviderRequest {
880                    model: "claude-sonnet".into(),
881                    body: json!({
882                        "messages": [
883                            { "role": "system", "content": "safe" },
884                            { "role": "user", "content": "hello" }
885                        ],
886                        "tools": [{ "type": "function", "function": { "name": "lookup", "parameters": { "type": "object" } } }]
887                    }),
888                },
889            )
890            .unwrap();
891        assert_eq!(request["system"], "safe");
892        assert_eq!(request["messages"][0]["content"][0]["text"], "hello");
893        assert_eq!(request["tools"][0]["name"], "lookup");
894
895        let response = adapter
896            .decode_response(
897                Surface::ChatCompletions,
898                json!({
899                    "id": "msg_1",
900                    "model": "claude-sonnet",
901                    "content": [{ "type": "text", "text": "answer" }],
902                    "stop_reason": "end_turn",
903                    "usage": { "input_tokens": 10, "output_tokens": 4 }
904                }),
905            )
906            .unwrap();
907        assert_eq!(response.body["choices"][0]["message"]["content"], "answer");
908        assert_eq!(response.usage.total_tokens(), 14);
909    }
910
911    #[test]
912    fn translates_reasoning_tool_choice_and_stop_sequences() {
913        let request = build_request(
914            "claude",
915            &json!({
916                "messages": [{ "role": "user", "content": "hello" }],
917                "max_tokens": 8000,
918                "reasoning_effort": "medium",
919                "temperature": 0.2,
920                "top_p": 0.9,
921                "stop": ["END", 42],
922                "tools": [{
923                    "type": "function",
924                    "function": { "name": "lookup", "parameters": { "type": "object" } }
925                }],
926                "tool_choice": {
927                    "type": "function",
928                    "function": { "name": "lookup" }
929                }
930            }),
931        );
932        assert_eq!(request["thinking"]["budget_tokens"], 4000);
933        assert!(request.get("temperature").is_none());
934        assert!(request.get("top_p").is_none());
935        assert_eq!(
936            request["tool_choice"],
937            json!({ "type": "tool", "name": "lookup" })
938        );
939        assert_eq!(request["stop_sequences"], json!(["END"]));
940        assert_eq!(
941            thinking_budget(&json!({ "reasoning_effort": "minimal" }), 4096),
942            Some(1024)
943        );
944        assert_eq!(
945            thinking_budget(&json!({ "reasoning_effort": "high" }), 4096),
946            Some(3072)
947        );
948        assert_eq!(
949            thinking_budget(&json!({ "reasoning_effort": "medium" }), 1500),
950            None
951        );
952    }
953
954    #[test]
955    fn signed_thinking_round_trips_through_encrypted_details() {
956        let adapter = AnthropicAdapter::new();
957        let response = adapter
958            .decode_response(
959                Surface::ChatCompletions,
960                json!({
961                    "content": [
962                        { "type": "thinking", "thinking": "check", "signature": "sig" },
963                        { "type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {} }
964                    ],
965                    "stop_reason": "tool_use"
966                }),
967            )
968            .unwrap();
969        let message = &response.body["choices"][0]["message"];
970        assert_eq!(
971            signed_thinking_from_details(message),
972            vec![SignedThinking {
973                thinking: "check".into(),
974                signature: "sig".into()
975            }]
976        );
977        let request = build_request(
978            "claude",
979            &json!({
980                "reasoning_effort": "low",
981                "max_tokens": 8000,
982                "messages": [message]
983            }),
984        );
985        assert_eq!(
986            request["messages"][0]["content"][0],
987            json!({ "type": "thinking", "thinking": "check", "signature": "sig" })
988        );
989        assert_eq!(request["messages"][0]["content"][1]["type"], "tool_use");
990    }
991
992    fn sse(data: Value) -> SseEvent {
993        SseEvent {
994            event: None,
995            data: data.to_string(),
996        }
997    }
998
999    #[test]
1000    fn stream_preserves_fragmented_tools_signed_reasoning_and_terminal_usage() {
1001        let adapter = AnthropicAdapter::new();
1002        let mut decoder = adapter.stream_decoder(Surface::ChatCompletions).unwrap();
1003        let upstream = [
1004            json!({ "type": "message_start", "message": { "usage": {
1005                "input_tokens": 12,
1006                "output_tokens": 0,
1007                "cache_read_input_tokens": 3,
1008                "cache_creation_input_tokens": 2
1009            }}}),
1010            json!({ "type": "content_block_start", "index": 0, "content_block": { "type": "thinking" }}),
1011            json!({ "type": "content_block_delta", "index": 0, "delta": { "type": "thinking_delta", "thinking": "check " }}),
1012            json!({ "type": "content_block_delta", "index": 0, "delta": { "type": "thinking_delta", "thinking": "weather" }}),
1013            json!({ "type": "content_block_delta", "index": 0, "delta": { "type": "signature_delta", "signature": "sig-" }}),
1014            json!({ "type": "content_block_delta", "index": 0, "delta": { "type": "signature_delta", "signature": "1" }}),
1015            json!({ "type": "content_block_stop", "index": 0 }),
1016            json!({ "type": "content_block_start", "index": 2, "content_block": {
1017                "type": "tool_use", "id": "toolu_1", "name": "lookup"
1018            }}),
1019            json!({ "type": "content_block_delta", "index": 2, "delta": {
1020                "type": "input_json_delta", "partial_json": "{\"city\":"
1021            }}),
1022            json!({ "type": "content_block_delta", "index": 2, "delta": {
1023                "type": "input_json_delta", "partial_json": "\"Paris\"}"
1024            }}),
1025            json!({ "type": "content_block_stop", "index": 2 }),
1026            json!({ "type": "message_delta", "delta": { "stop_reason": "tool_use" }, "usage": {
1027                "output_tokens": 5, "reasoning_tokens": 2
1028            }}),
1029            json!({ "type": "message_stop" }),
1030        ];
1031        let mut events = Vec::new();
1032        for event in upstream {
1033            events.extend(decoder.decode(sse(event)).unwrap());
1034        }
1035
1036        let mut assembler = crate::ToolCallAssembler::new();
1037        for event in &events {
1038            assembler.push_event(event).unwrap();
1039        }
1040        let calls = assembler.finish().unwrap();
1041        assert_eq!(calls.len(), 1);
1042        assert_eq!(calls[0].id, "toolu_1");
1043        assert_eq!(calls[0].name, "lookup");
1044        assert_eq!(
1045            calls[0].arguments_json().unwrap(),
1046            json!({ "city": "Paris" })
1047        );
1048
1049        let details = events.iter().find_map(|event| match event {
1050            ProviderStreamEvent::Data { data, .. }
1051                if data.pointer("/choices/0/delta/reasoning_details/0/type")
1052                    == Some(&json!("reasoning.encrypted")) =>
1053            {
1054                Some(&data["choices"][0]["delta"])
1055            }
1056            _ => None,
1057        });
1058        let details = details.expect("encrypted reasoning detail on tool start");
1059        assert_eq!(
1060            signed_thinking_from_details(details),
1061            vec![SignedThinking {
1062                thinking: "check weather".into(),
1063                signature: "sig-1".into()
1064            }]
1065        );
1066
1067        let terminal = events.iter().find_map(|event| match event {
1068            ProviderStreamEvent::Data { data, .. }
1069                if data.pointer("/choices/0/finish_reason") == Some(&json!("tool_calls")) =>
1070            {
1071                Some(data)
1072            }
1073            _ => None,
1074        });
1075        let terminal = terminal.expect("terminal chunk");
1076        assert_eq!(terminal["usage"]["prompt_tokens"], 17);
1077        assert_eq!(terminal["usage"]["completion_tokens"], 5);
1078        assert_eq!(terminal["usage"]["total_tokens"], 22);
1079        assert_eq!(
1080            terminal["usage"]["completion_tokens_details"]["reasoning_tokens"],
1081            2
1082        );
1083        assert_eq!(
1084            terminal["usage"]["prompt_tokens_details"]["cache_write_tokens"],
1085            2
1086        );
1087        assert!(matches!(
1088            events.last(),
1089            Some(ProviderStreamEvent::Done(ModelUsage {
1090                input_tokens: 12,
1091                output_tokens: 5,
1092                reasoning_tokens: 2,
1093                cache_read_tokens: 3,
1094                cache_write_tokens: 2,
1095            }))
1096        ));
1097    }
1098
1099    #[test]
1100    fn stream_finish_closes_an_upstream_stream_without_message_stop() {
1101        let mut decoder = AnthropicAdapter::new()
1102            .stream_decoder(Surface::ChatCompletions)
1103            .unwrap();
1104        decoder
1105            .decode(sse(json!({
1106                "type": "content_block_delta",
1107                "index": 0,
1108                "delta": { "type": "text_delta", "text": "answer" }
1109            })))
1110            .unwrap();
1111        let terminal = decoder.finish().unwrap();
1112        assert_eq!(terminal.len(), 2);
1113        assert!(matches!(terminal[0], ProviderStreamEvent::Data { .. }));
1114        assert!(matches!(terminal[1], ProviderStreamEvent::Done(_)));
1115        assert!(decoder.finish().unwrap().is_empty());
1116    }
1117
1118    #[test]
1119    fn native_response_usage_maps_cache_counters() {
1120        assert_eq!(
1121            native_message_usage(&json!({
1122                "id": "msg_1",
1123                "content": [{ "type": "text", "text": "answer" }],
1124                "usage": {
1125                    "input_tokens": 11,
1126                    "output_tokens": 4,
1127                    "cache_creation_input_tokens": 7,
1128                    "cache_read_input_tokens": 5
1129                }
1130            })),
1131            ModelUsage {
1132                input_tokens: 11,
1133                output_tokens: 4,
1134                reasoning_tokens: 0,
1135                cache_read_tokens: 5,
1136                cache_write_tokens: 7,
1137            }
1138        );
1139        assert_eq!(native_message_usage(&json!({})), ModelUsage::default());
1140    }
1141
1142    #[test]
1143    fn native_stream_forwards_events_verbatim_and_folds_split_usage() {
1144        let mut decoder = NativeMessagesDecoder::new();
1145        let thinking = json!({
1146            "type": "content_block_start",
1147            "index": 0,
1148            "content_block": { "type": "thinking", "thinking": "why", "signature": "sig-1" }
1149        });
1150        let upstream = vec![
1151            json!({ "type": "message_start", "message": { "usage": {
1152                "input_tokens": 12,
1153                "cache_read_input_tokens": 3,
1154                "cache_creation_input_tokens": 2
1155            }}}),
1156            thinking.clone(),
1157            json!({ "type": "content_block_stop", "index": 0 }),
1158            json!({ "type": "message_delta", "delta": { "stop_reason": "end_turn" }, "usage": {
1159                "output_tokens": 9
1160            }}),
1161            json!({ "type": "message_stop" }),
1162        ];
1163        let mut events = Vec::new();
1164        for event in &upstream {
1165            events.extend(decoder.decode(sse(event.clone())).unwrap());
1166        }
1167
1168        let forwarded: Vec<&Value> = events
1169            .iter()
1170            .filter_map(|event| match event {
1171                ProviderStreamEvent::Data { data, .. } => Some(data),
1172                ProviderStreamEvent::Done(_) => None,
1173            })
1174            .collect();
1175        assert_eq!(forwarded, upstream.iter().collect::<Vec<_>>());
1176        // The signed thinking block survives untouched, which is the whole point
1177        // of serving the native wire rather than translating it.
1178        assert_eq!(forwarded[1]["content_block"], thinking["content_block"]);
1179        assert_eq!(
1180            events.last(),
1181            Some(&ProviderStreamEvent::Done(ModelUsage {
1182                input_tokens: 12,
1183                output_tokens: 9,
1184                reasoning_tokens: 0,
1185                cache_read_tokens: 3,
1186                cache_write_tokens: 2,
1187            }))
1188        );
1189        assert!(decoder.finish().unwrap().is_empty());
1190    }
1191
1192    #[test]
1193    fn native_stream_reports_an_error_event_and_closes_a_truncated_stream() {
1194        let mut decoder = NativeMessagesDecoder::new();
1195        let error = decoder
1196            .decode(sse(
1197                json!({ "type": "error", "error": { "message": "overloaded" } }),
1198            ))
1199            .unwrap_err();
1200        assert!(matches!(error, ProviderError::InvalidStream(message) if message == "overloaded"));
1201
1202        let mut truncated = NativeMessagesDecoder::new();
1203        truncated
1204            .decode(sse(
1205                json!({ "type": "message_start", "message": { "usage": { "input_tokens": 4 } } }),
1206            ))
1207            .unwrap();
1208        assert_eq!(
1209            truncated.finish().unwrap(),
1210            vec![ProviderStreamEvent::Done(ModelUsage {
1211                input_tokens: 4,
1212                ..ModelUsage::default()
1213            })]
1214        );
1215    }
1216}
gateway_core/anthropic.rs

gateway_core/
anthropic.rs
