rig_core/http_client/
mod.rs

1use crate::http_client::sse::BoxedStream;
2use bytes::Bytes;
3pub use http::{HeaderMap, HeaderValue, Method, Request, Response, Uri, request::Builder};
4use http::{HeaderName, StatusCode};
5use reqwest::Body;
6pub mod multipart;
7pub mod retry;
8pub mod sse;
9use crate::wasm_compat::*;
10pub use multipart::MultipartForm;
11pub use reqwest::Client as ReqwestClient;
12use std::pin::Pin;
13
14#[derive(Debug, thiserror::Error)]
15pub enum Error {
16    #[error("Http error: {0}")]
17    Protocol(#[from] http::Error),
18    #[error("Invalid status code: {0}")]
19    InvalidStatusCode(StatusCode),
20    #[error("Invalid status code {0} with message: {1}")]
21    InvalidStatusCodeWithMessage(StatusCode, String),
22    /// A non-success HTTP response whose headers were preserved alongside the
23    /// body, so provider layers can read transport metadata — e.g. their
24    /// request-id contract — off the failed response (rig#2314). Displays
25    /// identically to [`Self::InvalidStatusCodeWithMessage`].
26    #[error("Invalid status code {status} with message: {body}")]
27    InvalidStatusCodeWithDetails {
28        /// The non-success status.
29        status: StatusCode,
30        /// The raw response body.
31        body: String,
32        /// The failed response's headers, verbatim.
33        headers: Box<http::HeaderMap>,
34    },
35    #[error("Header value outside of legal range: {0}")]
36    InvalidHeaderValue(#[from] http::header::InvalidHeaderValue),
37    #[error("Request in error state, cannot access headers")]
38    NoHeaders,
39    #[error("Stream ended")]
40    StreamEnded,
41    #[error("Invalid content type was returned: {0:?}")]
42    InvalidContentType(HeaderValue),
43    #[cfg(not(target_family = "wasm"))]
44    #[error("Http client error: {0}")]
45    Instance(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
46
47    #[cfg(target_family = "wasm")]
48    #[error("Http client error: {0}")]
49    Instance(#[from] Box<dyn std::error::Error + 'static>),
50}
51
52impl Error {
53    pub(crate) fn non_success_status(&self) -> Option<StatusCode> {
54        match self {
55            Self::InvalidStatusCode(status) | Self::InvalidStatusCodeWithMessage(status, _) => {
56                Some(*status)
57            }
58            Self::InvalidStatusCodeWithDetails { status, .. } => Some(*status),
59            _ => None,
60        }
61    }
62
63    pub(crate) fn non_success_body(&self) -> Option<&str> {
64        match self {
65            Self::InvalidStatusCodeWithMessage(_, body) => Some(body.as_str()),
66            Self::InvalidStatusCodeWithDetails { body, .. } => Some(body.as_str()),
67            _ => None,
68        }
69    }
70
71    /// Returns the failed response's headers, when this error preserved them.
72    ///
73    /// Rig's bundled HTTP clients capture the full [`HeaderMap`] whenever a
74    /// non-success status error is built from a live response, so rate-limit
75    /// metadata such as `Retry-After` or `x-ratelimit-*` stays readable
76    /// (rig#2210). This is the accessor a [`retry::RetryPolicy`] uses to honor
77    /// a server-supplied backoff, since it is handed this error directly:
78    ///
79    /// ```
80    /// # use rig_core::http_client::{Error, retry::RetryPolicy};
81    /// # use std::time::Duration;
82    /// fn retry_after(error: &Error) -> Option<Duration> {
83    ///     let seconds = error
84    ///         .non_success_headers()?
85    ///         .get(http::header::RETRY_AFTER)?
86    ///         .to_str()
87    ///         .ok()?
88    ///         .parse()
89    ///         .ok()?;
90    ///     Some(Duration::from_secs(seconds))
91    /// }
92    /// ```
93    ///
94    /// Returns `None` when the error carries no captured headers: transports
95    /// that report a non-success status without them, and errors built from
96    /// only a status and body.
97    pub fn non_success_headers(&self) -> Option<&HeaderMap> {
98        match self {
99            Self::InvalidStatusCodeWithDetails { headers, .. } => Some(headers),
100            _ => None,
101        }
102    }
103}
104
105pub type Result<T> = std::result::Result<T, Error>;
106
107#[cfg(not(target_family = "wasm"))]
108pub(crate) fn instance_error<E: std::error::Error + Send + Sync + 'static>(error: E) -> Error {
109    Error::Instance(error.into())
110}
111
112#[cfg(target_family = "wasm")]
113fn instance_error<E: std::error::Error + 'static>(error: E) -> Error {
114    Error::Instance(error.into())
115}
116
117async fn non_success_status_error(response: reqwest::Response) -> Error {
118    let status = response.status();
119    // Preserve the failed response's headers: provider layers read their
120    // request-id contract off them (rig#2314). The Display is identical to
121    // the header-less variant, so surfaced error text is unchanged.
122    let headers = Box::new(response.headers().clone());
123    let body = response
124        .text()
125        .await
126        .unwrap_or_else(|error| format!("failed to read error response body: {error}"));
127    Error::InvalidStatusCodeWithDetails {
128        status,
129        body,
130        headers,
131    }
132}
133
134pub type LazyBytes = WasmBoxedFuture<'static, Result<Bytes>>;
135pub type LazyBody<T> = WasmBoxedFuture<'static, Result<T>>;
136
137pub type StreamingResponse = Response<BoxedStream>;
138
139#[derive(Debug, Clone, Copy)]
140pub struct NoBody;
141
142impl From<NoBody> for Bytes {
143    fn from(_: NoBody) -> Self {
144        Bytes::new()
145    }
146}
147
148impl From<NoBody> for Body {
149    fn from(_: NoBody) -> Self {
150        reqwest::Body::default()
151    }
152}
153
154pub async fn text(response: Response<LazyBody<Vec<u8>>>) -> Result<String> {
155    let text = response.into_body().await?;
156    Ok(String::from(String::from_utf8_lossy(&text)))
157}
158
159pub fn make_auth_header(key: impl AsRef<str>) -> Result<(HeaderName, HeaderValue)> {
160    Ok((
161        http::header::AUTHORIZATION,
162        HeaderValue::from_str(&format!("Bearer {}", key.as_ref()))?,
163    ))
164}
165
166pub fn bearer_auth_header(headers: &mut HeaderMap, key: impl AsRef<str>) -> Result<()> {
167    let (k, v) = make_auth_header(key)?;
168
169    headers.insert(k, v);
170
171    Ok(())
172}
173
174/// A helper trait to make generic requests (both regular and SSE) possible.
175pub trait HttpClientExt: WasmCompatSend + WasmCompatSync {
176    /// Send a HTTP request, get a response back (as bytes). Response must be able to be turned back into Bytes.
177    fn send<T, U>(
178        &self,
179        req: Request<T>,
180    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
181    where
182        T: Into<Bytes>,
183        T: WasmCompatSend,
184        U: From<Bytes>,
185        U: WasmCompatSend + 'static;
186
187    /// Send a HTTP request with a multipart body, get a response back (as bytes). Response must be able to be turned back into Bytes (although usually for the response, you will probably want to specify Bytes anyway).
188    fn send_multipart<U>(
189        &self,
190        req: Request<MultipartForm>,
191    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
192    where
193        U: From<Bytes>,
194        U: WasmCompatSend + 'static;
195
196    /// Send a HTTP request, get a streamed response back (as a stream of [`bytes::Bytes`].)
197    fn send_streaming<T>(
198        &self,
199        req: Request<T>,
200    ) -> impl Future<Output = Result<StreamingResponse>> + WasmCompatSend
201    where
202        T: Into<Bytes> + WasmCompatSend;
203}
204
205async fn into_lazy_response<U>(response: reqwest::Response) -> Result<Response<LazyBody<U>>>
206where
207    U: From<Bytes>,
208    U: WasmCompatSend + 'static,
209{
210    if !response.status().is_success() {
211        return Err(non_success_status_error(response).await);
212    }
213
214    let mut res = Response::builder().status(response.status());
215
216    if let Some(headers) = res.headers_mut() {
217        *headers = response.headers().clone();
218    }
219
220    let body: LazyBody<U> = Box::pin(async {
221        let bytes = response.bytes().await.map_err(instance_error)?;
222        Ok(U::from(bytes))
223    });
224
225    res.body(body).map_err(Error::Protocol)
226}
227
228macro_rules! impl_http_client_ext {
229    ($(#[$attribute:meta])* $client:ty) => {
230        $(#[$attribute])*
231        impl HttpClientExt for $client {
232            fn send<T, U>(
233                &self,
234                req: Request<T>,
235            ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
236            where
237                T: Into<Bytes>,
238                U: From<Bytes> + WasmCompatSend + 'static,
239            {
240                let (parts, body) = req.into_parts();
241                let req = self
242                    .request(parts.method, parts.uri.to_string())
243                    .headers(parts.headers)
244                    .body(body.into());
245
246                async move {
247                    let response = req.send().await.map_err(instance_error)?;
248                    into_lazy_response(response).await
249                }
250            }
251
252            fn send_multipart<U>(
253                &self,
254                req: Request<MultipartForm>,
255            ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
256            where
257                U: From<Bytes>,
258                U: WasmCompatSend + 'static,
259            {
260                let (parts, body) = req.into_parts();
261                let body = reqwest::multipart::Form::from(body);
262
263                let req = self
264                    .request(parts.method, parts.uri.to_string())
265                    .headers(parts.headers)
266                    .multipart(body);
267
268                async move {
269                    let response = req.send().await.map_err(instance_error)?;
270                    into_lazy_response(response).await
271                }
272            }
273
274            fn send_streaming<T>(
275                &self,
276                req: Request<T>,
277            ) -> impl Future<Output = Result<StreamingResponse>> + WasmCompatSend
278            where
279                T: Into<Bytes> + WasmCompatSend,
280            {
281                let (parts, body) = req.into_parts();
282
283                let client = self.clone();
284
285                async move {
286                    let req = self
287                        .request(parts.method, parts.uri.to_string())
288                        .headers(parts.headers)
289                        .body(body.into())
290                        .build()
291                        .map_err(|error| Error::Instance(error.into()))?;
292                    let response: reqwest::Response =
293                        client.execute(req).await.map_err(instance_error)?;
294                    if !response.status().is_success() {
295                        return Err(non_success_status_error(response).await);
296                    }
297
298                    #[cfg(not(target_family = "wasm"))]
299                    let mut res = Response::builder()
300                        .status(response.status())
301                        .version(response.version());
302
303                    #[cfg(target_family = "wasm")]
304                    let mut res = Response::builder().status(response.status());
305
306                    if let Some(hs) = res.headers_mut() {
307                        *hs = response.headers().clone();
308                    }
309
310                    use futures::StreamExt;
311
312                    let mapped_stream: Pin<
313                        Box<dyn WasmCompatSendStream<InnerItem = Result<Bytes>>>,
314                    > = Box::pin(
315                        response
316                            .bytes_stream()
317                            .map(|chunk| chunk.map_err(|e| Error::Instance(Box::new(e)))),
318                    );
319
320                    res.body(mapped_stream).map_err(Error::Protocol)
321                }
322            }
323        }
324    };
325}
326
327impl_http_client_ext!(reqwest::Client);
328
329impl_http_client_ext!(
330    #[cfg(feature = "reqwest-middleware")]
331    #[cfg_attr(docsrs, doc(cfg(feature = "reqwest-middleware")))]
332    reqwest_middleware::ClientWithMiddleware
333);
334
335#[cfg(test)]
336mod non_success_header_tests {
337    use super::*;
338
339    /// rig#2210: the bundled transport's own error constructor is where the
340    /// headers are captured, so drive it with a real `reqwest::Response`.
341    #[tokio::test]
342    async fn non_success_status_error_preserves_response_headers() {
343        let response = http::Response::builder()
344            .status(StatusCode::TOO_MANY_REQUESTS)
345            .header("retry-after", "20")
346            .header("x-ratelimit-remaining", "0")
347            .body(r#"{"error":{"message":"rate limited"}}"#)
348            .expect("valid response");
349
350        let error = non_success_status_error(reqwest::Response::from(response)).await;
351
352        assert_eq!(
353            error.non_success_status(),
354            Some(StatusCode::TOO_MANY_REQUESTS)
355        );
356        assert_eq!(
357            error.non_success_body(),
358            Some(r#"{"error":{"message":"rate limited"}}"#)
359        );
360        let headers = error
361            .non_success_headers()
362            .expect("headers captured at error construction");
363        assert_eq!(
364            headers.get("retry-after").and_then(|v| v.to_str().ok()),
365            Some("20")
366        );
367        assert_eq!(
368            headers
369                .get("x-ratelimit-remaining")
370                .and_then(|v| v.to_str().ok()),
371            Some("0")
372        );
373    }
374
375    /// `None` means "not captured" and must not be confused with an empty map:
376    /// every other shape of this error reports it.
377    #[test]
378    fn non_success_headers_absent_when_not_captured() {
379        for error in [
380            Error::InvalidStatusCodeWithMessage(
381                StatusCode::TOO_MANY_REQUESTS,
382                "rate limited".to_string(),
383            ),
384            Error::InvalidStatusCode(StatusCode::TOO_MANY_REQUESTS),
385            Error::StreamEnded,
386        ] {
387            assert!(error.non_success_headers().is_none());
388        }
389
390        // A captured-but-empty map is `Some`, not `None`.
391        let error = Error::InvalidStatusCodeWithDetails {
392            status: StatusCode::TOO_MANY_REQUESTS,
393            body: "rate limited".to_string(),
394            headers: Box::new(HeaderMap::new()),
395        };
396        assert!(error.non_success_headers().is_some_and(HeaderMap::is_empty));
397    }
398}
rig_core/http_client/mod.rs

rig_core/http_client/
mod.rs
