1use crate::error;
4use eventsource_stream::Event as MessageEvent;
5use futures::{SinkExt, Stream, StreamExt};
6use reqwest_eventsource::{Event, RequestBuilderExt};
7use std::sync::Arc;
8use tokio_tungstenite::tungstenite;
9
10#[derive(Debug, Clone)]
34pub struct HttpClient {
35 pub http_client: reqwest::Client,
37 pub address: String,
39 pub authorization: Option<Arc<String>>,
41 pub user_agent: Option<String>,
43 pub x_title: Option<String>,
45 pub http_referer: Option<String>,
47 pub x_github_authorization: Option<Arc<String>>,
49 pub x_openrouter_authorization: Option<Arc<String>>,
51 pub x_mcp_authorization:
53 Option<Arc<std::collections::HashMap<String, String>>>,
54 pub agent_instance_hierarchy: Option<Arc<String>>,
56 pub mcp_session_id: Option<Arc<String>>,
60 pub mcp_call_timeout_ms: Option<u64>,
66}
67
68impl HttpClient {
69 pub fn new(
87 http_client: reqwest::Client,
88 address: Option<impl Into<String>>,
89 authorization: Option<impl Into<String>>,
90 user_agent: Option<impl Into<String>>,
91 x_title: Option<impl Into<String>>,
92 http_referer: Option<impl Into<String>>,
93 x_github_authorization: Option<impl Into<String>>,
94 x_openrouter_authorization: Option<impl Into<String>>,
95 x_mcp_authorization: Option<std::collections::HashMap<String, String>>,
96 agent_instance_hierarchy: Option<impl Into<String>>,
97 mcp_session_id: Option<impl Into<String>>,
98 mcp_call_timeout_ms: Option<u64>,
99 ) -> Self {
100 #[cfg(feature = "env")]
101 let env = |name: &str| -> Option<String> { std::env::var(name).ok() };
102
103 Self {
104 http_client,
105 address: match address {
106 Some(base) => base.into(),
107 #[cfg(feature = "env")]
108 None => env("OBJECTIVEAI_ADDRESS").unwrap_or_else(|| {
109 "https://api.objectiveai.dev".to_string()
110 }),
111 #[cfg(not(feature = "env"))]
112 None => "https://api.objectiveai.dev".to_string(),
113 },
114 authorization: authorization.map(|k| Arc::new(k.into())).or_else(
115 || {
116 #[cfg(feature = "env")]
117 {
118 env("OBJECTIVEAI_AUTHORIZATION").map(Arc::new)
119 }
120 #[cfg(not(feature = "env"))]
121 {
122 None
123 }
124 },
125 ),
126 user_agent: user_agent.map(Into::into).or_else(|| {
127 #[cfg(feature = "env")]
128 {
129 env("USER_AGENT")
130 }
131 #[cfg(not(feature = "env"))]
132 {
133 None
134 }
135 }),
136 x_title: x_title.map(Into::into).or_else(|| {
137 #[cfg(feature = "env")]
138 {
139 env("X_TITLE")
140 }
141 #[cfg(not(feature = "env"))]
142 {
143 None
144 }
145 }),
146 http_referer: http_referer.map(Into::into).or_else(|| {
147 #[cfg(feature = "env")]
148 {
149 env("HTTP_REFERER")
150 }
151 #[cfg(not(feature = "env"))]
152 {
153 None
154 }
155 }),
156 x_github_authorization: x_github_authorization
157 .map(|v| Arc::new(v.into()))
158 .or_else(|| {
159 #[cfg(feature = "env")]
160 {
161 env("GITHUB_AUTHORIZATION").map(Arc::new)
162 }
163 #[cfg(not(feature = "env"))]
164 {
165 None
166 }
167 }),
168 x_openrouter_authorization: x_openrouter_authorization
169 .map(|v| Arc::new(v.into()))
170 .or_else(|| {
171 #[cfg(feature = "env")]
172 {
173 env("OPENROUTER_AUTHORIZATION").map(Arc::new)
174 }
175 #[cfg(not(feature = "env"))]
176 {
177 None
178 }
179 }),
180 x_mcp_authorization: x_mcp_authorization.map(Arc::new).or_else(
181 || {
182 #[cfg(feature = "env")]
183 {
184 env("MCP_AUTHORIZATION")
185 .and_then(|v| serde_json::from_str(&v).ok())
186 .map(Arc::new)
187 }
188 #[cfg(not(feature = "env"))]
189 {
190 None
191 }
192 },
193 ),
194 agent_instance_hierarchy: agent_instance_hierarchy.map(|v| Arc::new(v.into())).or_else(|| {
195 #[cfg(feature = "env")]
196 {
197 env("OBJECTIVEAI_AGENT_INSTANCE_HIERARCHY").map(Arc::new)
198 }
199 #[cfg(not(feature = "env"))]
200 {
201 None
202 }
203 }),
204 mcp_session_id: mcp_session_id.map(|v| Arc::new(v.into())).or_else(
205 || {
206 #[cfg(feature = "env")]
207 {
208 env(crate::mcp::MCP_SESSION_ID_ENV).map(Arc::new)
209 }
210 #[cfg(not(feature = "env"))]
211 {
212 None
213 }
214 },
215 ),
216 mcp_call_timeout_ms,
219 }
220 }
221
222 fn request(
224 &self,
225 method: reqwest::Method,
226 path: &str,
227 body: Option<impl serde::Serialize>,
228 ) -> reqwest::RequestBuilder {
229 let url = format!(
230 "{}/{}",
231 self.address.trim_end_matches('/'),
232 path.trim_start_matches('/')
233 );
234 let mut request = self.http_client.request(method, &url);
235 if let Some(authorization) = &self.authorization {
236 let key = authorization
237 .strip_prefix("Bearer ")
238 .unwrap_or(authorization);
239 request =
240 request.header("authorization", format!("Bearer {}", key));
241 }
242 if let Some(user_agent) = &self.user_agent {
243 request = request.header("user-agent", user_agent);
244 }
245 if let Some(x_title) = &self.x_title {
246 request = request.header("x-title", x_title);
247 }
248 if let Some(http_referer) = &self.http_referer {
249 request = request.header("referer", http_referer);
250 request = request.header("http-referer", http_referer);
251 }
252 if let Some(token) = &self.x_github_authorization {
253 request = request.header("X-GITHUB-AUTHORIZATION", token.as_str());
254 }
255 if let Some(token) = &self.x_openrouter_authorization {
256 request =
257 request.header("X-OPENROUTER-AUTHORIZATION", token.as_str());
258 }
259 if let Some(headers) = &self.x_mcp_authorization {
260 if let Ok(json) = serde_json::to_string(headers.as_ref()) {
261 request = request.header("X-MCP-AUTHORIZATION", json);
262 }
263 }
264 if let Some(id) = &self.agent_instance_hierarchy {
265 request = request.header("X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY", id.as_str());
266 }
267 if let Some(s) = &self.mcp_session_id {
268 request =
269 request.header(crate::mcp::MCP_SESSION_ID_HEADER, s.as_str());
270 }
271 if let Some(ms) = self.mcp_call_timeout_ms {
272 request = request.header("X-MCP-CALL-TIMEOUT", ms.to_string());
273 }
274 if let Some(body) = body {
275 request = request.json(&body);
276 }
277 request
278 }
279
280 pub async fn send_unary<T: serde::de::DeserializeOwned + Send + 'static>(
291 &self,
292 method: reqwest::Method,
293 path: impl AsRef<str>,
294 body: Option<impl serde::Serialize>,
295 ) -> Result<T, super::HttpError> {
296 let response = self
297 .http_client
298 .execute(
299 self.request(method, path.as_ref(), body)
300 .build()
301 .map_err(super::HttpError::RequestError)?,
302 )
303 .await
304 .map_err(super::HttpError::HttpError)?;
305 let code = response.status();
306 if code.is_success() {
307 let text =
308 response.text().await.map_err(super::HttpError::HttpError)?;
309 let mut de = serde_json::Deserializer::from_str(&text);
310 match serde_path_to_error::deserialize::<_, T>(&mut de) {
311 Ok(value) => Ok(value),
312 Err(e) => Err(super::HttpError::DeserializationError(e)),
313 }
314 } else {
315 match response.text().await {
316 Ok(text) => Err(super::HttpError::BadStatus {
317 code,
318 body: match serde_json::from_str::<serde_json::Value>(&text)
319 {
320 Ok(body) => body,
321 Err(_) => serde_json::Value::String(text),
322 },
323 }),
324 Err(_) => Err(super::HttpError::BadStatus {
325 code,
326 body: serde_json::Value::Null,
327 }),
328 }
329 }
330 }
331
332 pub async fn send_unary_no_response(
340 &self,
341 method: reqwest::Method,
342 path: impl AsRef<str>,
343 body: Option<impl serde::Serialize>,
344 ) -> Result<(), super::HttpError> {
345 let response = self
346 .http_client
347 .execute(
348 self.request(method, path.as_ref(), body)
349 .build()
350 .map_err(super::HttpError::RequestError)?,
351 )
352 .await
353 .map_err(super::HttpError::HttpError)?;
354 let code = response.status();
355 if code.is_success() {
356 Ok(())
357 } else {
358 match response.text().await {
359 Ok(text) => Err(super::HttpError::BadStatus {
360 code,
361 body: match serde_json::from_str::<serde_json::Value>(&text)
362 {
363 Ok(body) => body,
364 Err(_) => serde_json::Value::String(text),
365 },
366 }),
367 Err(_) => Err(super::HttpError::BadStatus {
368 code,
369 body: serde_json::Value::Null,
370 }),
371 }
372 }
373 }
374
375 pub async fn send_streaming<
391 T: serde::de::DeserializeOwned + Send + 'static,
392 P: AsRef<str> + Send,
393 B: serde::Serialize + Send,
394 >(
395 &self,
396 method: reqwest::Method,
397 path: P,
398 body: Option<B>,
399 ) -> Result<
400 impl Stream<Item = Result<T, super::HttpError>>
401 + Send
402 + 'static
403 + use<T, P, B>,
404 super::HttpError,
405 > {
406 Ok(
412 self.request(method, path.as_ref(), body)
413 .header("X-Transport", "sse")
414 .eventsource()?
415 .take_while(|result| {
416 let dominated = matches!(
417 result,
418 Ok(Event::Message(MessageEvent { data, .. })) if data == "[DONE]"
419 );
420 async move { !dominated }
421 })
422 .then(|result| async {
423 match result {
424 Ok(Event::Open) => None,
425 Ok(Event::Message(MessageEvent { data, .. }))
426 if data.starts_with(":")
427 || data.is_empty() =>
428 {
429 None
430 }
431 Ok(Event::Message(MessageEvent { data, .. })) => {
432 let mut de =
433 serde_json::Deserializer::from_str(&data);
434 Some(
435 match serde_path_to_error::deserialize::<_, T>(
436 &mut de,
437 ) {
438 Ok(value) => Ok(value),
439 Err(e) => match serde_json::from_str::<error::ResponseError>(&data) {
440 Ok(err) => Err(super::HttpError::ApiError(err)),
441 Err(_) => Err(super::HttpError::DeserializationError(e)),
442 },
443 }
444 )
445 }
446 Err(reqwest_eventsource::Error::InvalidStatusCode(
447 code,
448 response,
449 )) => match response.text().await {
450 Ok(body) => {
451 Some(Err(super::HttpError::BadStatus {
452 code,
453 body: match serde_json::from_str::<
454 serde_json::Value,
455 >(
456 &body
457 ) {
458 Ok(body) => body,
459 Err(_) => {
460 serde_json::Value::String(body)
461 }
462 },
463 }))
464 }
465 Err(_) => Some(Err(super::HttpError::BadStatus {
466 code,
467 body: serde_json::Value::Null,
468 })),
469 },
470 Err(e) => Some(Err(super::HttpError::StreamError(e))),
471 }
472 })
473 .filter_map(|x| async { x }),
474 )
475 }
476
477 #[cfg(feature = "mcp")]
494 pub async fn send_streaming_ws<Chunk, B, H, P>(
495 &self,
496 method: reqwest::Method,
497 path: P,
498 body: B,
499 handler: H,
500 ) -> Result<
501 (
502 impl Stream<Item = Result<Chunk, super::HttpError>>
503 + Send
504 + Unpin
505 + 'static
506 + use<Chunk, B, H, P>,
507 super::Notifier,
508 ),
509 super::HttpError,
510 >
511 where
512 Chunk: serde::de::DeserializeOwned + Send + 'static,
513 B: serde::Serialize + Send + 'static,
514 H: super::McpHandler,
515 P: AsRef<str>,
516 {
517 use crate::client_objectiveai_mcp::{
518 client_response::Response as ClientResponse,
519 server_request::Request as ServerRequest,
520 };
521 use futures::stream::SplitStream;
522 use tokio::net::TcpStream;
523 use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
524
525 let url = format!(
528 "{}/{}",
529 self.address.trim_end_matches('/'),
530 path.as_ref().trim_start_matches('/')
531 );
532 let ws_url = if let Some(rest) = url.strip_prefix("https://") {
533 format!("wss://{rest}")
534 } else if let Some(rest) = url.strip_prefix("http://") {
535 format!("ws://{rest}")
536 } else {
537 url.clone()
538 };
539 let _ = method; let mut req = tungstenite::handshake::client::Request::builder()
544 .method("GET")
545 .uri(&ws_url)
546 .header(
547 "Host",
548 reqwest::Url::parse(&url)
549 .ok()
550 .and_then(|u| u.host_str().map(str::to_owned))
551 .unwrap_or_default(),
552 )
553 .header("Upgrade", "websocket")
554 .header("Connection", "Upgrade")
555 .header(
556 "Sec-WebSocket-Key",
557 tungstenite::handshake::client::generate_key(),
558 )
559 .header("Sec-WebSocket-Version", "13")
560 .header("X-Transport", "ws");
561 if let Some(authorization) = &self.authorization {
562 let key = authorization
563 .strip_prefix("Bearer ")
564 .unwrap_or(authorization.as_str());
565 req = req.header("authorization", format!("Bearer {}", key));
566 }
567 if let Some(ua) = &self.user_agent {
568 req = req.header("user-agent", ua);
569 }
570 if let Some(x_title) = &self.x_title {
571 req = req.header("x-title", x_title);
572 }
573 if let Some(http_referer) = &self.http_referer {
574 req = req.header("referer", http_referer);
575 req = req.header("http-referer", http_referer);
576 }
577 if let Some(token) = &self.x_github_authorization {
578 req = req.header("X-GITHUB-AUTHORIZATION", token.as_str());
579 }
580 if let Some(token) = &self.x_openrouter_authorization {
581 req = req.header("X-OPENROUTER-AUTHORIZATION", token.as_str());
582 }
583 if let Some(headers) = &self.x_mcp_authorization {
584 if let Ok(json) = serde_json::to_string(headers.as_ref()) {
585 req = req.header("X-MCP-AUTHORIZATION", json);
586 }
587 }
588 if let Some(id) = &self.agent_instance_hierarchy {
589 req = req.header("X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY", id.as_str());
590 }
591 if let Some(s) = &self.mcp_session_id {
592 req = req.header(crate::mcp::MCP_SESSION_ID_HEADER, s.as_str());
593 }
594 if let Some(ms) = self.mcp_call_timeout_ms {
595 req = req.header("X-MCP-CALL-TIMEOUT", ms.to_string());
596 }
597 let req = req.body(()).map_err(|e| {
598 super::HttpError::WsConnect(tungstenite::Error::Http(
599 tungstenite::http::Response::builder()
600 .status(400)
601 .body(Some(e.to_string().into_bytes()))
602 .unwrap(),
603 ))
604 })?;
605
606 let (ws_stream, _resp) = tokio_tungstenite::connect_async(req).await?;
607 let (mut sink, rx_stream): (
608 _,
609 SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
610 ) = ws_stream.split();
611
612 let body_frame = serde_json::to_string(&body)
614 .map_err(super::HttpError::NotifySerialize)?;
615 sink.send(tungstenite::Message::Text(body_frame.into()))
616 .await
617 .map_err(super::HttpError::NotifySend)?;
618
619 let sink: super::notifier::SharedSink =
621 Arc::new(tokio::sync::Mutex::new(sink));
622 let pending: super::notifier::PendingNotifies =
623 Arc::new(dashmap::DashMap::new());
624
625 let (chunk_tx, chunk_rx) = futures::channel::mpsc::unbounded::<
629 Result<Chunk, super::HttpError>,
630 >();
631
632 let demux_sink = sink.clone();
633 let demux_pending = pending.clone();
634 let handler = Arc::new(handler);
635 tokio::spawn(async move {
636 let mut rx_stream = rx_stream;
637 let mut chunk_tx = chunk_tx;
638 loop {
639 let msg = match rx_stream.next().await {
640 Some(m) => m,
641 None => break,
642 };
643 let text = match msg {
644 Ok(tungstenite::Message::Text(t)) => {
645 let s = t.to_string();
646 s
647 }
648 Ok(tungstenite::Message::Binary(_)) => {
649 continue;
650 }
651 Ok(
652 tungstenite::Message::Ping(_)
653 | tungstenite::Message::Pong(_),
654 ) => continue,
655 Ok(tungstenite::Message::Close(_)) => {
656 break;
657 }
658 Ok(tungstenite::Message::Frame(_)) => continue,
659 Err(_) => {
660 break;
661 }
662 };
663
664 if let Ok(response) =
669 serde_json::from_str::<ClientResponse>(&text)
670 {
671 let id = response.id().to_string();
672 if let Some((_, tx)) = demux_pending.remove(&id) {
673 let _ = tx.send(response);
674 }
675 continue;
676 }
677 if let Ok(request) =
678 serde_json::from_str::<ServerRequest>(&text)
679 {
680 let id = request.id.clone();
681 let handler = handler.clone();
682 let demux_sink = demux_sink.clone();
683 tokio::spawn(async move {
684 let id = id;
685 let response = handler.handle(request).await;
688 let frame = match serde_json::to_string(&response) {
689 Ok(s) => s,
690 Err(_) => {
691 return;
692 }
693 };
694 let mut guard = demux_sink.lock().await;
695 let send_result = guard
696 .send(tungstenite::Message::Text(frame.into()))
697 .await;
698 });
699 continue;
700 }
701
702 let mut de = serde_json::Deserializer::from_str(&text);
704 match serde_path_to_error::deserialize::<_, Chunk>(&mut de) {
705 Ok(chunk) => {
706 if chunk_tx.unbounded_send(Ok(chunk)).is_err() {
707 break;
708 }
709 }
710 Err(e) => {
711 let err = match serde_json::from_str::<
714 error::ResponseError,
715 >(&text)
716 {
717 Ok(api_err) => super::HttpError::ApiError(api_err),
718 Err(_) => super::HttpError::DeserializationError(e),
719 };
720 let _ = chunk_tx.unbounded_send(Err(err));
721 break;
722 }
723 }
724 }
725 drop(demux_pending);
729 drop(chunk_tx);
730 });
731
732 let notifier = super::Notifier::new(sink, pending);
733 Ok((chunk_rx, notifier))
734 }
735}