cera_client/
client.rs

1//! Main HTTP client implementation for OpenAI and OpenRouter APIs.
2
3use std::time::Duration;
4
5use reqwest::Response;
6use reqwest::header::HeaderMap;
7
8use crate::error::{ApiErrorEnvelope, ClientError};
9use crate::provider::{OPENAI_API_KEY_ENV, OPENAI_BASE_URL_ENV, OPENROUTER_API_KEY_ENV, Provider};
10use crate::stream::ChatCompletionStream;
11use crate::types::{
12    ChatCompletionRequest, ChatCompletionResponse, EmbeddingRequest, EmbeddingResponse,
13    ListModelsResponse,
14};
15
16/// Builder for constructing a configured [`Client`].
17#[derive(Clone)]
18pub struct ClientBuilder {
19    provider: Provider,
20    api_key: Option<String>,
21    timeout: Option<Duration>,
22    connect_timeout: Option<Duration>,
23}
24
25impl std::fmt::Debug for ClientBuilder {
26    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        f.debug_struct("ClientBuilder")
28            .field("provider", &self.provider)
29            .field("api_key", &self.api_key.as_ref().map(|_| "[REDACTED]"))
30            .field("timeout", &self.timeout)
31            .field("connect_timeout", &self.connect_timeout)
32            .finish()
33    }
34}
35
36impl ClientBuilder {
37    /// Create a new builder with the specified provider.
38    pub fn new(provider: Provider) -> Self {
39        Self {
40            provider,
41            api_key: None,
42            timeout: Some(Duration::from_secs(60)),
43            connect_timeout: Some(Duration::from_secs(10)),
44        }
45    }
46
47    /// Set the API key for authorization.
48    pub fn api_key(mut self, key: impl Into<String>) -> Self {
49        self.api_key = Some(key.into());
50        self
51    }
52
53    /// Set request timeout.
54    pub fn timeout(mut self, timeout: Duration) -> Self {
55        self.timeout = Some(timeout);
56        self
57    }
58
59    /// Disable request timeout.
60    pub fn no_timeout(mut self) -> Self {
61        self.timeout = None;
62        self
63    }
64
65    /// Set socket connection timeout.
66    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
67        self.connect_timeout = Some(timeout);
68        self
69    }
70
71    /// Disable socket connection timeout.
72    pub fn no_connect_timeout(mut self) -> Self {
73        self.connect_timeout = None;
74        self
75    }
76
77    /// Build the client instance.
78    pub fn build(self) -> Result<Client, ClientError> {
79        #[allow(unused_mut)]
80        let mut http_builder = reqwest::Client::builder();
81        #[cfg(not(target_arch = "wasm32"))]
82        {
83            if let Some(ct) = self.connect_timeout {
84                http_builder = http_builder.connect_timeout(ct);
85            }
86            http_builder =
87                http_builder.user_agent(concat!("cera-client/", env!("CARGO_PKG_VERSION")));
88        }
89
90        let http = http_builder.build()?;
91        Ok(Client {
92            http,
93            provider: self.provider,
94            api_key: self.api_key,
95            timeout: self.timeout,
96        })
97    }
98}
99
100/// Asynchronous API client for querying OpenAI and OpenRouter endpoints.
101#[derive(Clone)]
102pub struct Client {
103    http: reqwest::Client,
104    provider: Provider,
105    api_key: Option<String>,
106    #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
107    timeout: Option<Duration>,
108}
109
110impl std::fmt::Debug for Client {
111    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
112        f.debug_struct("Client")
113            .field("http", &self.http)
114            .field("provider", &self.provider)
115            .field("api_key", &self.api_key.as_ref().map(|_| "[REDACTED]"))
116            .field("timeout", &self.timeout)
117            .finish()
118    }
119}
120
121impl Client {
122    /// Create a new client targeting OpenAI with an API key.
123    pub fn new(api_key: impl Into<String>) -> Result<Self, ClientError> {
124        Self::openai(api_key)
125    }
126
127    /// Create a client targeting the official OpenAI API.
128    pub fn openai(api_key: impl Into<String>) -> Result<Self, ClientError> {
129        Self::builder(Provider::OpenAi).api_key(api_key).build()
130    }
131
132    /// Create a client targeting the OpenRouter API.
133    pub fn openrouter(api_key: impl Into<String>) -> Result<Self, ClientError> {
134        Self::builder(Provider::openrouter())
135            .api_key(api_key)
136            .build()
137    }
138
139    /// Create a client targeting OpenRouter with application attribution metadata.
140    pub fn openrouter_with_attribution(
141        api_key: impl Into<String>,
142        app_url: impl Into<String>,
143        app_name: impl Into<String>,
144    ) -> Result<Self, ClientError> {
145        Self::builder(Provider::openrouter_with_attribution(app_url, app_name))
146            .api_key(api_key)
147            .build()
148    }
149
150    /// Create a client targeting a custom OpenAI-compatible server.
151    pub fn custom(
152        base_url: impl Into<String>,
153        api_key: Option<String>,
154    ) -> Result<Self, ClientError> {
155        let mut b = Self::builder(Provider::custom(base_url));
156        if let Some(key) = api_key {
157            b = b.api_key(key);
158        }
159        b.build()
160    }
161
162    /// Automatically configure a client from available environment variables.
163    ///
164    /// Checks `OPENROUTER_API_KEY` first; if present, connects to OpenRouter.
165    /// Then checks `OPENAI_BASE_URL`; if present, connects to that custom endpoint
166    /// with optional `OPENAI_API_KEY`.
167    /// Otherwise checks `OPENAI_API_KEY` to connect to OpenAI.
168    pub fn from_env() -> Result<Self, ClientError> {
169        let or_key = std::env::var(OPENROUTER_API_KEY_ENV)
170            .ok()
171            .filter(|k| !k.trim().is_empty());
172        let oa_base = std::env::var(OPENAI_BASE_URL_ENV)
173            .ok()
174            .filter(|k| !k.trim().is_empty());
175        let oa_key = std::env::var(OPENAI_API_KEY_ENV)
176            .ok()
177            .filter(|k| !k.trim().is_empty());
178
179        if or_key.is_some() && (oa_base.is_some() || oa_key.is_some()) {
180            tracing::warn!(
181                target: "cera_client",
182                "Both OpenRouter ({OPENROUTER_API_KEY_ENV}) and OpenAI ({OPENAI_API_KEY_ENV}/{OPENAI_BASE_URL_ENV}) environment variables are set; defaulting to OpenRouter precedence"
183            );
184        }
185
186        if let Some(key) = or_key {
187            return Self::openrouter(key);
188        }
189
190        if let Some(base_url) = oa_base {
191            return Self::custom(base_url, oa_key);
192        }
193
194        if let Some(key) = oa_key {
195            return Self::openai(key);
196        }
197
198        Err(ClientError::MissingApiKey(format!(
199            "Neither {OPENROUTER_API_KEY_ENV}, {OPENAI_BASE_URL_ENV}, nor {OPENAI_API_KEY_ENV} is set in the environment"
200        )))
201    }
202
203    /// Initialize a builder with a specific provider.
204    pub fn builder(provider: Provider) -> ClientBuilder {
205        ClientBuilder::new(provider)
206    }
207
208    /// Returns a reference to the active provider.
209    pub fn provider(&self) -> &Provider {
210        &self.provider
211    }
212
213    /// Returns the configured API key if any.
214    pub fn api_key(&self) -> Option<&str> {
215        self.api_key.as_deref()
216    }
217
218    /// Sends a non-streaming chat completion request to `/chat/completions`.
219    pub async fn chat(
220        &self,
221        mut request: ChatCompletionRequest,
222    ) -> Result<ChatCompletionResponse, ClientError> {
223        request.stream = Some(false);
224        let url = self.provider.endpoint_url("chat/completions")?;
225        let mut headers = HeaderMap::new();
226        self.provider
227            .apply_headers(&mut headers, self.api_key.as_deref())?;
228
229        #[allow(unused_mut)]
230        let mut req = self.http.post(url).headers(headers).json(&request);
231        #[cfg(not(target_arch = "wasm32"))]
232        if let Some(to) = self.timeout {
233            req = req.timeout(to);
234        }
235        let response = req.send().await?;
236
237        let checked = Self::check_response(response).await?;
238        let body = checked.json::<ChatCompletionResponse>().await?;
239        Ok(body)
240    }
241
242    /// Sends a streaming chat completion request to `/chat/completions`, returning an SSE chunk stream.
243    ///
244    /// The request handshake awaits HTTP response headers. Once headers arrive and status is
245    /// validated, the connection remains open for streaming tokens. Total request timeout is omitted
246    /// to support long generations; connection timeout is governed by [`ClientBuilder::connect_timeout`].
247    ///
248    /// # Stalls and Inactivity Timeouts
249    ///
250    /// Because total request timeouts are omitted during streaming to permit long generations,
251    /// a network disruption or hung server connection mid-stream may not terminate automatically.
252    /// Callers in production environments should wrap stream consumption in an inactivity or idle
253    /// timeout (for example, using `tokio::time::timeout` between successive chunk yields) so that
254    /// mid-stream stalls can be detected and recovered from.
255    pub async fn chat_stream(
256        &self,
257        mut request: ChatCompletionRequest,
258    ) -> Result<
259        ChatCompletionStream<
260            impl futures_core::Stream<Item = Result<bytes::Bytes, reqwest::Error>> + Unpin,
261        >,
262        ClientError,
263    > {
264        request.stream = Some(true);
265        let url = self.provider.endpoint_url("chat/completions")?;
266        let mut headers = HeaderMap::new();
267        headers.insert(
268            reqwest::header::ACCEPT,
269            reqwest::header::HeaderValue::from_static("text/event-stream"),
270        );
271        headers.insert(
272            reqwest::header::CACHE_CONTROL,
273            reqwest::header::HeaderValue::from_static("no-cache"),
274        );
275        self.provider
276            .apply_headers(&mut headers, self.api_key.as_deref())?;
277
278        let response = self
279            .http
280            .post(url)
281            .headers(headers)
282            .json(&request)
283            .send()
284            .await?;
285
286        let checked = Self::check_response(response).await?;
287        let content_type = checked
288            .headers()
289            .get(reqwest::header::CONTENT_TYPE)
290            .and_then(|ct| ct.to_str().ok())
291            .map(|s| s.trim().to_lowercase());
292
293        if let Some(ct_lower) = content_type
294            && !ct_lower.starts_with("text/event-stream")
295            && !ct_lower.starts_with("application/x-ndjson")
296        {
297            let status = checked.status();
298            #[cfg(not(target_arch = "wasm32"))]
299            let text = {
300                // Read at most 64 KB to avoid unbounded memory usage on unexpected non-SSE streams
301                let mut body_bytes = Vec::new();
302                let mut checked = checked;
303                while let Ok(Some(chunk)) = checked.chunk().await {
304                    let remaining = 64 * 1024 - body_bytes.len();
305                    if chunk.len() <= remaining {
306                        body_bytes.extend_from_slice(&chunk);
307                    } else {
308                        body_bytes.extend_from_slice(&chunk[..remaining]);
309                        break;
310                    }
311                    if body_bytes.len() >= 64 * 1024 {
312                        break;
313                    }
314                }
315                String::from_utf8(body_bytes)
316                    .unwrap_or_else(|err| String::from_utf8_lossy(err.as_bytes()).into_owned())
317            };
318            #[cfg(target_arch = "wasm32")]
319            let text = {
320                // In WASM browser environments, response buffering is managed by the browser
321                // Fetch API where Response::chunk is unavailable. We read text and enforce
322                // preview truncation below.
323                checked.text().await.unwrap_or_default()
324            };
325            if let Ok(envelope) = serde_json::from_str::<ApiErrorEnvelope>(&text) {
326                return Err(ClientError::Api {
327                    status: Some(status),
328                    message: envelope.error.message,
329                    error_type: envelope.error.error_type,
330                    code: envelope.error.code,
331                    param: envelope.error.param,
332                });
333            }
334            let preview = if text.len() > 512 {
335                let mut end = 512;
336                while !text.is_char_boundary(end) {
337                    end -= 1;
338                }
339                format!("{}... [truncated]", &text[..end])
340            } else {
341                text
342            };
343            return Err(ClientError::Stream(format!(
344                "Expected text/event-stream content type, received {ct_lower}: {preview}"
345            )));
346        }
347
348        let stream = checked.bytes_stream();
349        Ok(ChatCompletionStream::new(stream))
350    }
351
352    /// Generates vector embeddings for input text via `/embeddings`.
353    pub async fn embeddings(
354        &self,
355        request: EmbeddingRequest,
356    ) -> Result<EmbeddingResponse, ClientError> {
357        let url = self.provider.endpoint_url("embeddings")?;
358        let mut headers = HeaderMap::new();
359        self.provider
360            .apply_headers(&mut headers, self.api_key.as_deref())?;
361
362        #[allow(unused_mut)]
363        let mut req = self.http.post(url).headers(headers).json(&request);
364        #[cfg(not(target_arch = "wasm32"))]
365        if let Some(to) = self.timeout {
366            req = req.timeout(to);
367        }
368        let response = req.send().await?;
369
370        let checked = Self::check_response(response).await?;
371        let body = checked.json::<EmbeddingResponse>().await?;
372        Ok(body)
373    }
374
375    /// Lists models available from the provider via `/models`.
376    pub async fn models(&self) -> Result<ListModelsResponse, ClientError> {
377        let url = self.provider.endpoint_url("models")?;
378        let mut headers = HeaderMap::new();
379        self.provider
380            .apply_headers(&mut headers, self.api_key.as_deref())?;
381
382        #[allow(unused_mut)]
383        let mut req = self.http.get(url).headers(headers);
384        #[cfg(not(target_arch = "wasm32"))]
385        if let Some(to) = self.timeout {
386            req = req.timeout(to);
387        }
388        let response = req.send().await?;
389
390        let checked = Self::check_response(response).await?;
391        let body = checked.json::<ListModelsResponse>().await?;
392        Ok(body)
393    }
394
395    /// Validates response status, parsing error payloads if unsuccessful.
396    async fn check_response(response: Response) -> Result<Response, ClientError> {
397        let status = response.status();
398        if status.is_success() {
399            return Ok(response);
400        }
401
402        let raw_text = response.text().await.unwrap_or_default();
403        if let Ok(envelope) = serde_json::from_str::<ApiErrorEnvelope>(&raw_text) {
404            return Err(ClientError::Api {
405                status: Some(status),
406                message: envelope.error.message,
407                error_type: envelope.error.error_type,
408                code: envelope.error.code,
409                param: envelope.error.param,
410            });
411        }
412
413        Err(ClientError::Api {
414            status: Some(status),
415            message: if raw_text.is_empty() {
416                format!("HTTP error {status}")
417            } else {
418                raw_text
419            },
420            error_type: None,
421            code: None,
422            param: None,
423        })
424    }
425}
426
427#[cfg(test)]
428mod tests {
429    use super::*;
430
431    #[test]
432    fn test_client_constructors() {
433        let client_oa = Client::openai("sk-12345").unwrap();
434        assert_eq!(client_oa.provider(), &Provider::OpenAi);
435        assert_eq!(client_oa.api_key(), Some("sk-12345"));
436
437        let client_or = Client::openrouter("or-67890").unwrap();
438        assert_eq!(client_or.provider(), &Provider::openrouter());
439        assert_eq!(client_or.api_key(), Some("or-67890"));
440
441        let client_custom = Client::custom("http://localhost:11434/v1", None).unwrap();
442        assert_eq!(
443            client_custom.provider(),
444            &Provider::custom("http://localhost:11434/v1")
445        );
446        assert_eq!(client_custom.api_key(), None);
447    }
448
449    #[test]
450    fn test_client_builder_customization() {
451        let client = Client::builder(Provider::openrouter_with_attribution(
452            "https://test.com",
453            "TestApp",
454        ))
455        .api_key("key-abc")
456        .timeout(Duration::from_secs(120))
457        .connect_timeout(Duration::from_secs(10))
458        .build()
459        .unwrap();
460
461        assert_eq!(client.api_key(), Some("key-abc"));
462        match client.provider() {
463            Provider::OpenRouter { app_url, app_name } => {
464                assert_eq!(app_url.as_deref(), Some("https://test.com"));
465                assert_eq!(app_name.as_deref(), Some("TestApp"));
466            }
467            _ => panic!("expected OpenRouter provider"),
468        }
469
470        let unbounded_client = Client::builder(Provider::OpenAi)
471            .no_timeout()
472            .no_connect_timeout()
473            .build()
474            .unwrap();
475        assert_eq!(unbounded_client.provider(), &Provider::OpenAi);
476    }
477}
cera_client/client.rs

cera_client/
client.rs
