ag_ui/client/transport/
http.rs

1//! The `reqwest`-backed HTTP transport.
2//!
3//! One POST of the [`RunAgentInput`](https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/input/struct.RunAgentInput.html) as JSON, one `text/event-stream` response
4//! decoded by [`crate::client::transport::sse`]. This module is the only place in the
5//! crate that pulls in an HTTP client, and it sits behind the `http` feature so
6//! that a wasm or custom-transport build never sees it.
7
8use std::time::Duration;
9
10use crate::{RunAgentInput, SSE_MEDIA_TYPE};
11use reqwest::header::{ACCEPT, HeaderMap, HeaderName, HeaderValue};
12use reqwest::{Client, Url};
13
14use crate::client::error::{Error, Result};
15use crate::client::transport::sse::decode_events;
16use crate::client::transport::{EventStream, Transport, TransportFuture};
17
18/// How much of a failing response body is kept in the error.
19const MAX_ERROR_BODY: usize = 2048;
20
21/// POSTs a run to an HTTP endpoint and streams the response.
22#[derive(Clone, Debug)]
23pub struct HttpTransport {
24    client: Client,
25    url: Url,
26    headers: HeaderMap,
27    timeout: Option<Duration>,
28}
29
30impl HttpTransport {
31    /// A transport pointed at an agent's run endpoint, with default settings.
32    ///
33    /// # Errors
34    ///
35    /// [`Error::Config`] when the URL does not parse, or the HTTP client cannot
36    /// be built.
37    pub fn new(url: impl AsRef<str>) -> Result<Self> {
38        Self::builder(url).build()
39    }
40
41    /// A builder, for headers, timeouts, or a pre-configured client.
42    pub fn builder(url: impl AsRef<str>) -> HttpTransportBuilder {
43        HttpTransportBuilder::new(url)
44    }
45
46    /// The endpoint this transport posts to.
47    pub fn url(&self) -> &Url {
48        &self.url
49    }
50
51    /// The headers sent with every run.
52    pub fn headers(&self) -> &HeaderMap {
53        &self.headers
54    }
55}
56
57impl Transport for HttpTransport {
58    fn run(&self, input: RunAgentInput) -> TransportFuture {
59        // Cloned rather than borrowed so the future outlives this call; see the
60        // note on [`Transport`]. A `reqwest::Client` is an `Arc` inside.
61        let client = self.client.clone();
62        let url = self.url.clone();
63        let headers = self.headers.clone();
64        let timeout = self.timeout;
65
66        Box::pin(async move {
67            let mut request = client.post(url).headers(headers).json(&input);
68            if let Some(timeout) = timeout {
69                request = request.timeout(timeout);
70            }
71
72            let response = request.send().await.map_err(Error::transport)?;
73            let status = response.status();
74            if !status.is_success() {
75                let body = response.text().await.unwrap_or_default();
76                return Err(Error::Http {
77                    status: status.as_u16(),
78                    body: body.chars().take(MAX_ERROR_BODY).collect(),
79                });
80            }
81
82            Ok(Box::pin(decode_events(response.bytes_stream())) as EventStream)
83        })
84    }
85}
86
87/// Builds an [`HttpTransport`].
88///
89/// Header values are validated when [`HttpTransportBuilder::build`] is called,
90/// so a chain of setters stays a chain and does not thread a `Result` through
91/// every step.
92#[derive(Clone, Debug)]
93pub struct HttpTransportBuilder {
94    url: String,
95    headers: Vec<(String, String)>,
96    timeout: Option<Duration>,
97    connect_timeout: Option<Duration>,
98    client: Option<Client>,
99}
100
101impl HttpTransportBuilder {
102    /// A builder for an agent at `url`.
103    pub fn new(url: impl AsRef<str>) -> Self {
104        Self {
105            url: url.as_ref().to_owned(),
106            headers: Vec::new(),
107            timeout: None,
108            connect_timeout: None,
109            client: None,
110        }
111    }
112
113    /// Adds a header to every request — an API key, a tenant id, a trace
114    /// header.
115    #[must_use]
116    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
117        self.headers.push((name.into(), value.into()));
118        self
119    }
120
121    /// Adds several headers.
122    #[must_use]
123    pub fn headers<K, V>(mut self, headers: impl IntoIterator<Item = (K, V)>) -> Self
124    where
125        K: Into<String>,
126        V: Into<String>,
127    {
128        self.headers
129            .extend(headers.into_iter().map(|(k, v)| (k.into(), v.into())));
130        self
131    }
132
133    /// Bounds the whole run: connecting, headers, *and* streaming the body.
134    ///
135    /// An agent that thinks for longer than this has its stream cut off, so a
136    /// long-running agent wants [`HttpTransportBuilder::connect_timeout`]
137    /// instead.
138    #[must_use]
139    pub fn timeout(mut self, timeout: Duration) -> Self {
140        self.timeout = Some(timeout);
141        self
142    }
143
144    /// Bounds only connection setup, leaving the stream itself unbounded.
145    #[must_use]
146    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
147        self.connect_timeout = Some(timeout);
148        self
149    }
150
151    /// Uses a caller-supplied client, for proxies, custom TLS roots, or a
152    /// connection pool shared with the rest of an application.
153    #[must_use]
154    pub fn client(mut self, client: Client) -> Self {
155        self.client = Some(client);
156        self
157    }
158
159    /// Builds the transport.
160    ///
161    /// # Errors
162    ///
163    /// [`Error::Config`] when the URL, a header name, or a header value is not
164    /// valid, or when the HTTP client cannot be built.
165    pub fn build(self) -> Result<HttpTransport> {
166        let url = Url::parse(&self.url)
167            .map_err(|error| Error::Config(format!("invalid URL {:?}: {error}", self.url)))?;
168
169        let mut headers = HeaderMap::with_capacity(self.headers.len() + 1);
170        headers.insert(ACCEPT, HeaderValue::from_static(SSE_MEDIA_TYPE));
171        for (name, value) in self.headers {
172            let name = HeaderName::try_from(name.as_str())
173                .map_err(|error| Error::Config(format!("invalid header name {name:?}: {error}")))?;
174            let value = HeaderValue::try_from(value.as_str()).map_err(|error| {
175                Error::Config(format!("invalid value for header {name:?}: {error}"))
176            })?;
177            headers.insert(name, value);
178        }
179
180        let client = match self.client {
181            Some(client) => client,
182            None => {
183                let mut builder = Client::builder();
184                if let Some(timeout) = self.connect_timeout {
185                    builder = builder.connect_timeout(timeout);
186                }
187                builder.build().map_err(|error| {
188                    Error::Config(format!("could not build HTTP client: {error}"))
189                })?
190            }
191        };
192
193        Ok(HttpTransport {
194            client,
195            url,
196            headers,
197            timeout: self.timeout,
198        })
199    }
200}
ag_ui/client/transport/http.rs

ag_ui/client/transport/
http.rs
