infinity_provider_models_dev/
anthropic.rs

1//! Anthropic Messages backend with SSE streaming.
2//!
3//! Used for `ANTHROPIC_API_KEY` (and compatible env aliases). The rest of
4//! the presets go through the OpenAI-compatible backend in `openai.rs`.
5
6use async_stream::stream;
7use futures_util::StreamExt;
8use infinity_provider_protocol::{
9    AssistantContent, CompletionError, CompletionRequest, ErrorClass, FinalResponse, Message,
10    ModelStream, Reasoning, StreamChunk, ToolCall, ToolCallDeltaContent, Usage, UserContent,
11};
12use reqwest::Client;
13use serde_json::{Value, json};
14
15use crate::convert::{classify_http_status, error_message_from_body};
16use crate::oauth::{self, OAuthSessionHandle};
17
18const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
19const ANTHROPIC_VERSION: &str = "2023-06-01";
20
21/// How to reach and authenticate one `Anthropic` backend.
22pub struct Backend<'a> {
23    /// Shared HTTP client.
24    pub client: &'a Client,
25    /// `https://api.anthropic.com` (or override).
26    pub base_url: &'a str,
27    /// API key; ignored when `oauth` is present.
28    pub api_key: &'a str,
29    /// Extra headers (e.g. `OpenRouter` attribution).
30    pub extra_headers: &'a [(String, String)],
31    /// Subscription session; `None` for API-key auth.
32    pub oauth: Option<&'a OAuthSessionHandle>,
33}
34
35/// Stream one completion from the `Anthropic` Messages endpoint.
36///
37/// # Errors
38///
39/// Returns [`CompletionError::provider`] when the endpoint rejects the
40/// request (with the provider's retry classification) and
41/// [`CompletionError::ResponseError`] when a live stream breaks mid-flight.
42/// Individual stream items may also yield errors.
43pub async fn invoke(
44    backend: &Backend<'_>,
45    upstream_model: &str,
46    default_max_tokens: Option<u64>,
47    request: CompletionRequest,
48) -> Result<ModelStream, CompletionError> {
49    let body = request_body(upstream_model, default_max_tokens, &request);
50
51    // Subscription sessions mint a bearer token (refreshing when needed);
52    // plain API keys pass through untouched.
53    let oauth_token: Option<String> = match backend.oauth {
54        Some(handle) => Some(ensure_bearer(backend.client, handle, false).await?),
55        None => None,
56    };
57    let mut resp = post_messages(
58        backend.client,
59        backend.base_url,
60        backend.api_key,
61        oauth_token.as_deref(),
62        backend.extra_headers,
63        &body,
64    )
65    .await?;
66    // A `401` means the bearer died between the freshness check and the
67    // request: refresh once and retry before surfacing the failure.
68    if resp.status() == reqwest::StatusCode::UNAUTHORIZED {
69        if let Some(handle) = backend.oauth {
70            let token = ensure_bearer(backend.client, handle, true).await?;
71            resp = post_messages(
72                backend.client,
73                backend.base_url,
74                backend.api_key,
75                Some(&token),
76                backend.extra_headers,
77                &body,
78            )
79            .await?;
80        }
81    }
82
83    if !resp.status().is_success() {
84        let status = resp.status().as_u16();
85        let text = resp.text().await.unwrap_or_default();
86        let class = classify_http_status(status, &text);
87        return Err(CompletionError::provider(
88            class,
89            format!(
90                "{} -> {status}: {}",
91                backend.base_url,
92                error_message_from_body(&text)
93            ),
94        ));
95    }
96
97    let mut byte_stream = resp.bytes_stream();
98
99    let s = stream! {
100        let mut buf = String::new();
101        let mut state = StreamState::new();
102
103        while let Some(chunk) = byte_stream.next().await {
104            let bytes = match chunk {
105                Ok(b) => b,
106                Err(e) => {
107                    yield Err(CompletionError::ResponseError(format!("stream read failed: {e}")));
108                    break;
109                }
110            };
111            buf.push_str(&String::from_utf8_lossy(&bytes));
112            while let Some(end) = buf.find("\n\n") {
113                let frame_end = end.saturating_add(2);
114                let frame: String = buf.drain(..frame_end).collect();
115                buf = buf.trim_start_matches(['\n', '\r']).to_owned();
116                let Some((event, data)) = split_event(&frame) else {
117                    continue;
118                };
119                if data.is_empty() || event == "ping" {
120                    continue;
121                }
122                let Ok(v) = serde_json::from_str::<Value>(&data) else {
123                    continue;
124                };
125                for item in state.handle_event(&event, &v) {
126                    yield item;
127                }
128            }
129        }
130        yield Ok(StreamChunk::Final(FinalResponse {
131            usage: Some(state.usage),
132        }));
133    };
134    Ok(Box::pin(s))
135}
136
137/// Split one SSE frame into its `event:` name and `data:` payload.
138fn split_event(frame: &str) -> Option<(String, String)> {
139    let mut event = String::new();
140    let mut data = String::new();
141    for line in frame.lines() {
142        let line = line.trim();
143        if let Some(e) = line.strip_prefix("event:") {
144            e.trim().clone_into(&mut event);
145        } else if let Some(d) = line.strip_prefix("data:") {
146            if !data.is_empty() {
147                data.push('\n');
148            }
149            data.push_str(d.trim());
150        }
151    }
152    if event.is_empty() && data.is_empty() {
153        return None;
154    }
155    Some((event, data))
156}
157
158/// Mutable state for one `Anthropic` streaming response.
159struct StreamState {
160    usage: Usage,
161    tool_index: usize,
162    tool_id: String,
163    tool_name: String,
164    tool_args: String,
165    tool_name_sent: bool,
166}
167
168impl StreamState {
169    fn new() -> Self {
170        Self {
171            usage: Usage::default(),
172            tool_index: 0,
173            tool_id: String::new(),
174            tool_name: String::new(),
175            tool_args: String::new(),
176            tool_name_sent: false,
177        }
178    }
179
180    fn handle_event(
181        &mut self,
182        event: &str,
183        v: &Value,
184    ) -> Vec<Result<StreamChunk, CompletionError>> {
185        match event {
186            "message_start" => {
187                if let Some(u) = v.pointer("/message/usage").map(parse_anthropic_usage) {
188                    self.usage = u;
189                }
190                Vec::new()
191            }
192            "content_block_start" => self.on_block_start(v),
193            "content_block_delta" => self.on_block_delta(v),
194            "content_block_stop" => self.on_block_stop(),
195            "message_delta" => {
196                if let Some(u) = v.get("usage").map(parse_anthropic_usage) {
197                    self.usage.output_tokens = self.usage.output_tokens.max(u.output_tokens);
198                    self.usage.total_tokens = self
199                        .usage
200                        .input_tokens
201                        .saturating_add(self.usage.output_tokens);
202                    if u.cached_input_tokens > 0 {
203                        self.usage.cached_input_tokens = u.cached_input_tokens;
204                    }
205                }
206                Vec::new()
207            }
208            "message_stop" => {
209                if let Some(u) = v.get("usage").map(parse_anthropic_usage) {
210                    self.usage = u;
211                }
212                Vec::new()
213            }
214            "error" => {
215                let msg = v
216                    .pointer("/error/message")
217                    .and_then(Value::as_str)
218                    .or_else(|| v.get("message").and_then(Value::as_str))
219                    .unwrap_or("anthropic stream error");
220                vec![Err(CompletionError::provider(
221                    classify_http_status(400, msg),
222                    msg.to_owned(),
223                ))]
224            }
225            _ => Vec::new(),
226        }
227    }
228
229    fn on_block_start(&mut self, v: &Value) -> Vec<Result<StreamChunk, CompletionError>> {
230        let block = v.get("content_block").unwrap_or(&Value::Null);
231        if block.get("type").and_then(Value::as_str) != Some("tool_use") {
232            return Vec::new();
233        }
234        block
235            .get("id")
236            .and_then(Value::as_str)
237            .unwrap_or("")
238            .clone_into(&mut self.tool_id);
239        block
240            .get("name")
241            .and_then(Value::as_str)
242            .unwrap_or("")
243            .clone_into(&mut self.tool_name);
244        self.tool_args.clear();
245        self.tool_name_sent = false;
246        if self.tool_id.is_empty() {
247            self.tool_id = format!("tool-{}", self.tool_index);
248        }
249        if self.tool_name.is_empty() {
250            return Vec::new();
251        }
252        self.tool_name_sent = true;
253        vec![Ok(StreamChunk::ToolCallDelta {
254            id: self.tool_id.clone(),
255            content: ToolCallDeltaContent::Name(self.tool_name.clone()),
256        })]
257    }
258
259    fn on_block_delta(&mut self, v: &Value) -> Vec<Result<StreamChunk, CompletionError>> {
260        let delta = v.get("delta").unwrap_or(&Value::Null);
261        match delta.get("type").and_then(Value::as_str) {
262            Some("text_delta") => {
263                if let Some(t) = delta.get("text").and_then(Value::as_str) {
264                    if !t.is_empty() {
265                        return vec![Ok(StreamChunk::Text(t.to_owned()))];
266                    }
267                }
268                Vec::new()
269            }
270            Some("thinking_delta" | "signature_delta") => {
271                if let Some(t) = delta.get("thinking").and_then(Value::as_str) {
272                    if !t.is_empty() {
273                        return vec![Ok(StreamChunk::ReasoningDelta {
274                            id: None,
275                            text: t.to_owned(),
276                        })];
277                    }
278                }
279                Vec::new()
280            }
281            Some("input_json_delta") => self.on_tool_fragment(delta),
282            _ => Vec::new(),
283        }
284    }
285
286    fn on_tool_fragment(&mut self, delta: &Value) -> Vec<Result<StreamChunk, CompletionError>> {
287        let Some(frag) = delta.get("partial_json").and_then(Value::as_str) else {
288            return Vec::new();
289        };
290        self.tool_args.push_str(frag);
291        let mut out = Vec::new();
292        if !self.tool_name_sent && !self.tool_name.is_empty() {
293            self.tool_name_sent = true;
294            out.push(Ok(StreamChunk::ToolCallDelta {
295                id: self.tool_id.clone(),
296                content: ToolCallDeltaContent::Name(self.tool_name.clone()),
297            }));
298        }
299        if !frag.is_empty() {
300            out.push(Ok(StreamChunk::ToolCallDelta {
301                id: self.tool_id.clone(),
302                content: ToolCallDeltaContent::Delta(frag.to_owned()),
303            }));
304        }
305        out
306    }
307
308    fn on_block_stop(&mut self) -> Vec<Result<StreamChunk, CompletionError>> {
309        if self.tool_name.is_empty() && self.tool_args.is_empty() {
310            return Vec::new();
311        }
312        let args: Value = if self.tool_args.trim().is_empty() {
313            json!({})
314        } else {
315            serde_json::from_str(&self.tool_args)
316                .unwrap_or_else(|_| json!({"_raw": self.tool_args}))
317        };
318        let id = if self.tool_id.is_empty() {
319            format!("tool-{}", self.tool_index)
320        } else {
321            self.tool_id.clone()
322        };
323        self.tool_index = self.tool_index.saturating_add(1);
324        let call = StreamChunk::ToolCall(ToolCall::new(id, self.tool_name.clone(), args));
325        self.tool_id.clear();
326        self.tool_name.clear();
327        self.tool_args.clear();
328        self.tool_name_sent = false;
329        vec![Ok(call)]
330    }
331}
332
333/// Resolve the subscription bearer token, refreshing when stale (`force`
334/// refreshes unconditionally after a `401`).
335async fn ensure_bearer(
336    client: &Client,
337    handle: &OAuthSessionHandle,
338    force: bool,
339) -> Result<String, CompletionError> {
340    let refreshed = if force {
341        oauth::force_refresh(client, handle).await
342    } else {
343        oauth::ensure_fresh_token(client, handle).await
344    };
345    refreshed.map_err(|e| {
346        let class = if e.retryable {
347            ErrorClass::Transient
348        } else {
349            ErrorClass::Fatal
350        };
351        CompletionError::provider(class, format!("oauth: {}", e.message))
352    })
353}
354
355/// Whether a key is a subscription bearer rather than an API key.
356fn is_oauth_token(api_key: &str) -> bool {
357    api_key.starts_with("sk-ant-oauth")
358}
359
360/// Send one `Messages` request with API-key or subscription auth.
361///
362/// Subscription (`Bearer`) requests omit `x-api-key` and carry the `OAuth`
363/// beta plus `Claude Code` identity headers, mirroring the community bridges.
364async fn post_messages(
365    client: &Client,
366    base_url: &str,
367    api_key: &str,
368    oauth_token: Option<&str>,
369    extra_headers: &[(String, String)],
370    body: &Value,
371) -> Result<reqwest::Response, CompletionError> {
372    let url = format!("{}/messages", base_url.trim_end_matches('/'));
373    let mut req = client
374        .post(&url)
375        .timeout(REQUEST_TIMEOUT)
376        .header("Content-Type", "application/json")
377        .header("Accept", "text/event-stream")
378        .header("anthropic-version", ANTHROPIC_VERSION);
379    if let Some(token) = oauth_token {
380        req = req
381            .bearer_auth(token)
382            .header("anthropic-beta", oauth::OAUTH_BETA)
383            .header("anthropic-dangerous-direct-browser-access", "true")
384            .header("User-Agent", oauth::CLAUDE_CODE_USER_AGENT)
385            .header("x-app", "cli");
386    } else if is_oauth_token(api_key) {
387        // Static subscription token via the `anthropic` preset.
388        req = req
389            .bearer_auth(api_key)
390            .header("anthropic-beta", oauth::OAUTH_BETA);
391    } else {
392        req = req.header("x-api-key", api_key);
393    }
394    for (k, v) in extra_headers {
395        req = req.header(k.as_str(), v.as_str());
396    }
397    req.json(body).send().await.map_err(transport_error)
398}
399
400fn request_body(
401    upstream_model: &str,
402    default_max_tokens: Option<u64>,
403    request: &CompletionRequest,
404) -> Value {
405    let (system, messages) =
406        to_anthropic_messages(request.preamble.as_deref(), &request.chat_history);
407    let mut map = serde_json::Map::new();
408    map.insert("model".to_owned(), json!(upstream_model));
409    map.insert("messages".to_owned(), Value::Array(messages));
410    map.insert("stream".to_owned(), Value::Bool(true));
411    if let Some(system) = system {
412        map.insert("system".to_owned(), system);
413    }
414    let max = request.max_tokens.or(default_max_tokens).unwrap_or(16_384);
415    map.insert("max_tokens".to_owned(), json!(max));
416    if !request.tools.is_empty() {
417        let tools: Vec<Value> = request
418            .tools
419            .iter()
420            .map(|t| {
421                json!({
422                    "name": t.name,
423                    "description": t.description,
424                    "input_schema": t.parameters,
425                })
426            })
427            .collect();
428        map.insert("tools".to_owned(), Value::Array(tools));
429    }
430    let mut body = Value::Object(map);
431    if let Some(extra) = request.additional_params.as_ref() {
432        merge_value(&mut body, extra);
433    }
434    body
435}
436
437fn parse_anthropic_usage(u: &Value) -> Usage {
438    let input = u.get("input_tokens").and_then(Value::as_u64).unwrap_or(0);
439    let output = u.get("output_tokens").and_then(Value::as_u64).unwrap_or(0);
440    Usage {
441        input_tokens: input,
442        output_tokens: output,
443        total_tokens: input.saturating_add(output),
444        cached_input_tokens: u
445            .get("cache_read_input_tokens")
446            .and_then(Value::as_u64)
447            .unwrap_or(0),
448    }
449}
450
451fn to_anthropic_messages(
452    preamble: Option<&str>,
453    history: &[Message],
454) -> (Option<Value>, Vec<Value>) {
455    let system = preamble.map(|s| json!([{"type": "text", "text": s}]));
456    let mut messages = Vec::new();
457    for msg in history {
458        match msg {
459            Message::User { content } => {
460                let mut blocks = Vec::new();
461                for c in content {
462                    match c {
463                        UserContent::Text(t) => {
464                            blocks.push(json!({"type": "text", "text": t.text}));
465                        }
466                        UserContent::Image(img) => {
467                            blocks.push(image_block(img));
468                        }
469                        UserContent::ToolResult(r) => {
470                            let text = r
471                                .content
472                                .iter()
473                                .map(|b| match b {
474                                    infinity_provider_protocol::ToolResultContent::Text(t) => {
475                                        t.text.clone()
476                                    }
477                                    infinity_provider_protocol::ToolResultContent::Image(_) => {
478                                        "[image omitted]".to_owned()
479                                    }
480                                })
481                                .collect::<Vec<_>>()
482                                .join("\n");
483                            blocks.push(json!({
484                                "type": "tool_result",
485                                "tool_use_id": r.id,
486                                "content": text,
487                            }));
488                        }
489                    }
490                }
491                if !blocks.is_empty() {
492                    messages.push(json!({"role": "user", "content": blocks}));
493                }
494            }
495            Message::Assistant { content } => {
496                let mut blocks = Vec::new();
497                for c in content {
498                    match c {
499                        AssistantContent::Text(t) => {
500                            blocks.push(json!({"type": "text", "text": t.text}));
501                        }
502                        AssistantContent::ToolCall(tc) => blocks.push(json!({
503                            "type": "tool_use",
504                            "id": tc.id,
505                            "name": tc.function.name,
506                            "input": tc.function.arguments,
507                        })),
508                        AssistantContent::Reasoning(r) => {
509                            blocks.push(reasoning_block(r));
510                        }
511                    }
512                }
513                if !blocks.is_empty() {
514                    messages.push(json!({"role": "assistant", "content": blocks}));
515                }
516            }
517        }
518    }
519    (system, messages)
520}
521
522fn image_block(img: &infinity_provider_protocol::Image) -> Value {
523    match &img.data {
524        infinity_provider_protocol::ImageSource::Url(u) => json!({
525            "type": "image",
526            "source": {"type": "url", "url": u},
527        }),
528        infinity_provider_protocol::ImageSource::Base64(b) => {
529            let mime = img.media_type.as_ref().map_or(
530                "image/png",
531                infinity_provider_protocol::ImageMediaType::to_mime_type,
532            );
533            json!({
534                "type": "image",
535                "source": {"type": "base64", "media_type": mime, "data": b},
536            })
537        }
538    }
539}
540
541fn reasoning_block(r: &Reasoning) -> Value {
542    let text = r.display_text();
543    let signature = r.first_signature().unwrap_or("");
544    json!({"type": "thinking", "thinking": text, "signature": signature})
545}
546
547fn merge_value(dst: &mut Value, src: &Value) {
548    match (dst, src) {
549        (Value::Object(d), Value::Object(s)) => {
550            for (k, v) in s {
551                merge_value(d.entry(k.clone()).or_insert(Value::Null), v);
552            }
553        }
554        (d, s) => *d = s.clone(),
555    }
556}
557
558fn transport_error(e: reqwest::Error) -> CompletionError {
559    if e.is_timeout() || e.is_connect() || e.is_body() || e.is_decode() {
560        CompletionError::provider(ErrorClass::Transient, format!("transport error: {e}"))
561    } else if e.is_request() {
562        CompletionError::RequestError(Box::new(e))
563    } else {
564        CompletionError::provider(ErrorClass::Transient, format!("transport error: {e}"))
565    }
566}
infinity_provider_models_dev/anthropic.rs

infinity_provider_models_dev/
anthropic.rs
