1use anyhow::{Context, Result};
4use async_trait::async_trait;
5use futures::StreamExt;
6use std::env;
7use std::pin::Pin;
8use std::sync::Arc;
9use std::time::Duration;
10use tokio_util::sync::CancellationToken;
11
12#[derive(Debug, thiserror::Error)]
16pub enum HttpClientError {
17 #[error("{operation} was cancelled")]
18 Cancelled { operation: String },
19 #[error("{operation} transport failed: {message}")]
20 Transport { operation: String, message: String },
21 #[error("{operation} request was invalid: {message}")]
22 InvalidRequest { operation: String, message: String },
23}
24
25impl HttpClientError {
26 pub fn cancelled(operation: impl Into<String>) -> Self {
27 Self::Cancelled {
28 operation: operation.into(),
29 }
30 }
31
32 pub fn transport(operation: impl Into<String>, message: impl Into<String>) -> Self {
33 Self::Transport {
34 operation: operation.into(),
35 message: message.into(),
36 }
37 }
38
39 fn from_reqwest(operation: &str, error: reqwest::Error) -> Self {
40 if error.is_builder() {
41 Self::InvalidRequest {
42 operation: operation.to_string(),
43 message: error.to_string(),
44 }
45 } else if error.is_timeout() {
46 Self::transport(operation, format!("timed out: {error}"))
51 } else {
52 Self::transport(operation, error.to_string())
53 }
54 }
55
56 pub fn is_retryable(&self) -> bool {
57 match self {
58 Self::Transport { message, .. } => !is_hard_connect_failure(message),
62 Self::Cancelled { .. } | Self::InvalidRequest { .. } => false,
63 }
64 }
65}
66
67pub(crate) fn is_hard_connect_failure(message: &str) -> bool {
70 let lower = message.to_ascii_lowercase();
71 lower.contains("connection reset")
72 || lower.contains("connection refused")
73 || lower.contains("network is unreachable")
74 || lower.contains("no route to host")
75 || lower.contains("name or service not known")
76 || lower.contains("nodename nor servname provided")
77 || lower.contains("failed to lookup address information")
78 || lower.contains("dns error")
79 || lower.contains("could not resolve host")
80 || lower.contains("temporary failure in name resolution")
81}
82
83pub(crate) fn is_retryable_http_failure(error: &anyhow::Error) -> bool {
84 error
85 .downcast_ref::<HttpClientError>()
86 .is_some_and(HttpClientError::is_retryable)
87}
88
89pub struct HttpResponse {
91 pub status: u16,
92 pub body: String,
93}
94
95pub struct StreamingHttpResponse {
97 pub status: u16,
98 pub retry_after: Option<String>,
100 pub byte_stream: Pin<Box<dyn futures::Stream<Item = Result<bytes::Bytes>> + Send>>,
102 pub error_body: String,
104}
105
106#[derive(Debug, Clone)]
108pub struct HttpMetricsRecord {
109 pub url: String,
111 pub method: String,
113 pub status: u16,
115 pub duration_ms: f64,
117 pub request_bytes: u64,
119 pub response_bytes: u64,
121 pub streaming: bool,
123}
124
125pub type HttpMetricsCallback = Arc<dyn Fn(HttpMetricsRecord) + Send + Sync>;
128
129static HTTP_METRICS_CALLBACK: std::sync::RwLock<Option<HttpMetricsCallback>> =
134 std::sync::RwLock::new(None);
135
136pub fn set_http_metrics_callback(callback: HttpMetricsCallback) {
139 *HTTP_METRICS_CALLBACK.write().unwrap() = Some(callback);
140}
141
142pub fn clear_http_metrics_callback() {
144 *HTTP_METRICS_CALLBACK.write().unwrap() = None;
145}
146
147fn maybe_record_metrics(record: HttpMetricsRecord) {
148 if let Some(callback) = HTTP_METRICS_CALLBACK.read().unwrap().as_ref() {
149 callback(record);
150 }
151}
152
153#[async_trait]
157pub trait HttpClient: Send + Sync {
158 async fn post(
160 &self,
161 url: &str,
162 headers: Vec<(&str, &str)>,
163 body: &serde_json::Value,
164 cancel_token: CancellationToken,
165 ) -> Result<HttpResponse>;
166
167 async fn post_streaming(
170 &self,
171 url: &str,
172 headers: Vec<(&str, &str)>,
173 body: &serde_json::Value,
174 cancel_token: CancellationToken,
175 ) -> Result<StreamingHttpResponse>;
176}
177
178pub struct ReqwestHttpClient {
180 client: reqwest::Client,
181}
182
183impl ReqwestHttpClient {
184 pub fn new() -> Self {
185 Self {
186 client: build_reqwest_client(None, None).expect("failed to build default HTTP client"),
187 }
188 }
189
190 pub fn with_timeout(timeout: Duration) -> Result<Self> {
191 Ok(Self {
192 client: build_reqwest_client(Some(timeout), None)?,
193 })
194 }
195}
196
197impl Default for ReqwestHttpClient {
198 fn default() -> Self {
199 Self::new()
200 }
201}
202
203#[async_trait]
204impl HttpClient for ReqwestHttpClient {
205 async fn post(
206 &self,
207 url: &str,
208 headers: Vec<(&str, &str)>,
209 body: &serde_json::Value,
210 cancel_token: CancellationToken,
211 ) -> Result<HttpResponse> {
212 let start = std::time::Instant::now();
213 let request_body = serde_json::to_string(body).unwrap_or_default();
214 let request_bytes = request_body.len() as u64;
215
216 tracing::debug!(
217 "HTTP POST to {}: {}",
218 url,
219 serde_json::to_string_pretty(body)?
220 );
221
222 let mut request = self.client.post(url);
223 for (key, value) in headers {
224 request = request.header(key, value);
225 }
226 request = request.json(body);
227
228 let response = tokio::select! {
229 _ = cancel_token.cancelled() => {
230 return Err(anyhow::Error::new(HttpClientError::cancelled("HTTP request")));
231 }
232 result = request.send() => {
233 result.map_err(|error| {
234 anyhow::Error::new(HttpClientError::from_reqwest("HTTP request", error))
235 })?
236 }
237 };
238
239 let status = response.status().as_u16();
240 let response_body = response.text().await.map_err(|error| {
241 anyhow::Error::new(HttpClientError::from_reqwest("HTTP response body", error))
242 })?;
243 let response_bytes = response_body.len() as u64;
244 let duration_ms = start.elapsed().as_secs_f64() * 1000.0;
245
246 maybe_record_metrics(HttpMetricsRecord {
247 url: url.to_string(),
248 method: "POST".to_string(),
249 status,
250 duration_ms,
251 request_bytes,
252 response_bytes,
253 streaming: false,
254 });
255
256 Ok(HttpResponse {
257 status,
258 body: response_body,
259 })
260 }
261
262 async fn post_streaming(
263 &self,
264 url: &str,
265 headers: Vec<(&str, &str)>,
266 body: &serde_json::Value,
267 cancel_token: CancellationToken,
268 ) -> Result<StreamingHttpResponse> {
269 let start = std::time::Instant::now();
270 let request_body = serde_json::to_string(body).unwrap_or_default();
271 let request_bytes = request_body.len() as u64;
272
273 let mut request = self.client.post(url);
274 for (key, value) in headers {
275 request = request.header(key, value);
276 }
277 request = request.json(body);
278
279 let response = tokio::select! {
280 _ = cancel_token.cancelled() => {
281 return Err(anyhow::Error::new(HttpClientError::cancelled(
282 "HTTP streaming request",
283 )));
284 }
285 result = request.send() => {
286 result.map_err(|error| {
287 anyhow::Error::new(HttpClientError::from_reqwest(
288 "HTTP streaming request",
289 error,
290 ))
291 })?
292 }
293 };
294
295 let status = response.status().as_u16();
296 let retry_after = response
297 .headers()
298 .get("retry-after")
299 .and_then(|v| v.to_str().ok())
300 .map(String::from);
301
302 let duration_ms = start.elapsed().as_secs_f64() * 1000.0;
305 maybe_record_metrics(HttpMetricsRecord {
306 url: url.to_string(),
307 method: "POST".to_string(),
308 status,
309 duration_ms,
310 request_bytes,
311 response_bytes: 0, streaming: true,
313 });
314
315 if (200..300).contains(&status) {
316 let byte_stream = response.bytes_stream().map(|result| {
317 result.map_err(|error| {
318 anyhow::Error::new(HttpClientError::from_reqwest("HTTP response stream", error))
319 })
320 });
321 Ok(StreamingHttpResponse {
322 status,
323 retry_after,
324 byte_stream: Box::pin(byte_stream),
325 error_body: String::new(),
326 })
327 } else {
328 let error_body = response.text().await.unwrap_or_default();
329 let empty: futures::stream::Empty<Result<bytes::Bytes>> = futures::stream::empty();
331 Ok(StreamingHttpResponse {
332 status,
333 retry_after,
334 byte_stream: Box::pin(empty),
335 error_body,
336 })
337 }
338 }
339}
340
341pub fn default_http_client() -> Arc<dyn HttpClient> {
343 Arc::new(ReqwestHttpClient::new())
344}
345
346#[derive(Debug, Clone, Default, PartialEq, Eq)]
347struct ExplicitProxyConfig {
348 http: Option<String>,
349 https: Option<String>,
350}
351
352pub(crate) fn build_reqwest_client(
359 timeout: Option<Duration>,
360 default_headers: Option<reqwest::header::HeaderMap>,
361) -> Result<reqwest::Client> {
362 let mut builder = reqwest::Client::builder().no_proxy();
363
364 if let Some(timeout) = timeout {
365 builder = builder.timeout(timeout);
366 }
367
368 if let Some(default_headers) = default_headers {
369 builder = builder.default_headers(default_headers);
370 }
371
372 let proxy_config = explicit_proxy_config_from_env();
373 let no_proxy = reqwest::NoProxy::from_env();
377 if let Some(http_proxy) = proxy_config.http.as_deref() {
378 builder = builder.proxy(
379 reqwest::Proxy::http(http_proxy)
380 .with_context(|| format!("Invalid HTTP proxy URL: {http_proxy}"))?
381 .no_proxy(no_proxy.clone()),
382 );
383 }
384 if let Some(https_proxy) = proxy_config.https.as_deref() {
385 builder = builder.proxy(
386 reqwest::Proxy::https(https_proxy)
387 .with_context(|| format!("Invalid HTTPS proxy URL: {https_proxy}"))?
388 .no_proxy(no_proxy),
389 );
390 }
391
392 builder.build().context("Failed to build reqwest client")
393}
394
395fn explicit_proxy_config_from_env() -> ExplicitProxyConfig {
396 let http = first_non_empty_env(&["http_proxy", "HTTP_PROXY"]);
397 let https = first_non_empty_env(&["https_proxy", "HTTPS_PROXY"]).or_else(|| http.clone());
398
399 ExplicitProxyConfig { http, https }
400}
401
402fn first_non_empty_env(keys: &[&str]) -> Option<String> {
403 keys.iter().find_map(|key| {
404 env::var(key)
405 .ok()
406 .map(|value| value.trim().to_string())
407 .filter(|value| !value.is_empty())
408 })
409}
410
411pub(crate) fn normalize_base_url(base_url: &str) -> String {
413 base_url
414 .trim_end_matches('/')
415 .trim_end_matches("/v1")
416 .trim_end_matches('/')
417 .to_string()
418}
419
420fn base_has_versioned_api_root(base: &str) -> bool {
421 base.contains("/api/")
422 || base.ends_with("/v4")
423 || base.ends_with("/v3")
424 || base.ends_with("/v2")
425}
426
427pub(crate) fn join_chat_completions_url(base_url: &str, chat_path: &str) -> String {
433 let base = base_url.trim_end_matches('/');
434 if chat_path.is_empty() {
435 return base.to_string();
436 }
437 let mut path = if chat_path.starts_with('/') {
438 chat_path.to_string()
439 } else {
440 format!("/{chat_path}")
441 };
442
443 if base.ends_with("/chat/completions") {
444 return base.to_string();
445 }
446
447 if path.starts_with("/v1/") && base_has_versioned_api_root(base) {
451 path = path.replacen("/v1", "", 1);
452 }
453
454 const PAAS_V4: &str = "/paas/v4";
458 if base.contains(PAAS_V4) {
459 if let Some(idx) = path.find(PAAS_V4) {
460 let suffix = &path[idx + PAAS_V4.len()..];
461 return format!("{base}{suffix}");
462 }
463 }
464
465 format!("{base}{path}")
466}
467
468#[cfg(test)]
469mod tests {
470 use super::*;
471 use std::sync::{Mutex, OnceLock};
472 use tokio::io::{AsyncReadExt, AsyncWriteExt};
473
474 #[test]
475 fn retryable_http_failure_requires_a_typed_transport_error() {
476 let prose = anyhow::anyhow!(
477 "Human-readable text says timeout, connection reset, and TLS handshake."
478 );
479 assert!(!is_retryable_http_failure(&prose));
480
481 let transport = anyhow::Error::new(HttpClientError::transport(
482 "stream request",
483 "opaque diagnostic",
484 ));
485 assert!(is_retryable_http_failure(&transport));
486
487 let cancelled = anyhow::Error::new(HttpClientError::cancelled("stream request"));
488 assert!(!is_retryable_http_failure(&cancelled));
489 }
490
491 #[test]
492 fn hard_connect_failures_are_not_retryable() {
493 let reset = anyhow::Error::new(HttpClientError::transport(
494 "API request",
495 "error sending request for url (https://api.deepseek.com/v1/chat/completions): connection reset",
496 ));
497 assert!(!is_retryable_http_failure(&reset));
498 assert!(is_hard_connect_failure(
499 "error sending request … connection reset by peer"
500 ));
501
502 let refused = anyhow::Error::new(HttpClientError::transport(
503 "API request",
504 "tcp connect error: Connection refused (os error 61)",
505 ));
506 assert!(!is_retryable_http_failure(&refused));
507
508 let timeout = anyhow::Error::new(HttpClientError::transport(
509 "API request",
510 "timed out: operation timed out",
511 ));
512 assert!(is_retryable_http_failure(&timeout));
513 }
514
515 fn proxy_env_lock() -> &'static Mutex<()> {
516 static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
517 LOCK.get_or_init(|| Mutex::new(()))
518 }
519
520 fn clear_proxy_env() {
521 for key in [
522 "http_proxy",
523 "HTTP_PROXY",
524 "https_proxy",
525 "HTTPS_PROXY",
526 "no_proxy",
527 "NO_PROXY",
528 ] {
529 unsafe { env::remove_var(key) };
530 }
531 }
532
533 #[test]
534 fn test_normalize_base_url() {
535 assert_eq!(
536 normalize_base_url("https://api.example.com"),
537 "https://api.example.com"
538 );
539 assert_eq!(
540 normalize_base_url("https://api.example.com/"),
541 "https://api.example.com"
542 );
543 assert_eq!(
544 normalize_base_url("https://api.example.com/v1"),
545 "https://api.example.com"
546 );
547 assert_eq!(
548 normalize_base_url("https://api.example.com/v1/"),
549 "https://api.example.com"
550 );
551 }
552
553 #[test]
554 fn test_normalize_base_url_edge_cases() {
555 assert_eq!(
556 normalize_base_url("http://localhost:8080/v1"),
557 "http://localhost:8080"
558 );
559 assert_eq!(
560 normalize_base_url("http://localhost:8080"),
561 "http://localhost:8080"
562 );
563 assert_eq!(
564 normalize_base_url("https://api.example.com/v1/"),
565 "https://api.example.com"
566 );
567 }
568
569 #[test]
570 fn test_normalize_base_url_multiple_trailing_slashes() {
571 assert_eq!(
572 normalize_base_url("https://api.example.com//"),
573 "https://api.example.com"
574 );
575 }
576
577 #[test]
578 fn test_normalize_base_url_with_port() {
579 assert_eq!(
580 normalize_base_url("http://localhost:11434/v1/"),
581 "http://localhost:11434"
582 );
583 }
584
585 #[test]
586 fn test_normalize_base_url_already_normalized() {
587 assert_eq!(
588 normalize_base_url("https://api.openai.com"),
589 "https://api.openai.com"
590 );
591 }
592
593 #[test]
594 fn test_normalize_base_url_empty_string() {
595 assert_eq!(normalize_base_url(""), "");
596 }
597
598 #[test]
599 fn join_chat_completions_url_openai_default() {
600 assert_eq!(
601 join_chat_completions_url("https://api.openai.com", "/v1/chat/completions"),
602 "https://api.openai.com/v1/chat/completions"
603 );
604 }
605
606 #[test]
607 fn join_chat_completions_url_zhipu_default() {
608 assert_eq!(
609 join_chat_completions_url("https://open.bigmodel.cn", "/api/paas/v4/chat/completions"),
610 "https://open.bigmodel.cn/api/paas/v4/chat/completions"
611 );
612 }
613
614 #[test]
615 fn join_chat_completions_url_coding_plan_openai_compatible() {
616 assert_eq!(
618 join_chat_completions_url(
619 "https://open.bigmodel.cn/api/coding/paas/v4",
620 "/v1/chat/completions"
621 ),
622 "https://open.bigmodel.cn/api/coding/paas/v4/chat/completions"
623 );
624 }
625
626 #[test]
627 fn join_chat_completions_url_coding_plan_builtin_zhipu() {
628 assert_eq!(
630 join_chat_completions_url(
631 "https://open.bigmodel.cn/api/coding/paas/v4",
632 "/api/paas/v4/chat/completions"
633 ),
634 "https://open.bigmodel.cn/api/coding/paas/v4/chat/completions"
635 );
636 }
637
638 #[test]
639 fn join_chat_completions_url_full_endpoint_as_base() {
640 assert_eq!(
641 join_chat_completions_url(
642 "https://open.bigmodel.cn/api/coding/paas/v4/chat/completions",
643 "/v1/chat/completions"
644 ),
645 "https://open.bigmodel.cn/api/coding/paas/v4/chat/completions"
646 );
647 }
648
649 #[test]
650 fn test_default_http_client_creation() {
651 let _client = default_http_client();
652 }
653
654 #[tokio::test]
655 async fn test_reqwest_http_client_timeout_applies_to_api_call() {
656 let mut last_refused = None;
657 for _ in 0..3 {
658 let (elapsed, err) = post_to_slow_local_server().await;
659 assert!(
660 elapsed < Duration::from_secs(1),
661 "API timeout should fail quickly, elapsed={elapsed:?}"
662 );
663
664 let msg = format!("{err:?}").to_ascii_lowercase();
665 if msg.contains("connection refused") {
666 last_refused = Some(err);
667 continue;
668 }
669
670 assert!(
671 msg.contains("timed out") || msg.contains("timeout"),
672 "expected timeout error, got: {err:?}"
673 );
674 return;
675 }
676
677 panic!(
678 "local timeout server was not reachable after retries; last error: {:?}",
679 last_refused.expect("at least one connection-refused error")
680 );
681 }
682
683 async fn post_to_slow_local_server() -> (Duration, anyhow::Error) {
684 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
685 let addr = listener.local_addr().unwrap();
686
687 let server = tokio::spawn(async move {
688 let (mut stream, _) = listener.accept().await.unwrap();
689 let mut buf = [0_u8; 1024];
690 let _ = stream.read(&mut buf).await;
691 tokio::time::sleep(Duration::from_millis(250)).await;
692 let _ = stream
693 .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
694 .await;
695 });
696
697 let client = {
702 let _guard = proxy_env_lock().lock().unwrap();
703 clear_proxy_env();
704 ReqwestHttpClient::with_timeout(Duration::from_millis(50)).unwrap()
705 };
706 let started = std::time::Instant::now();
707 let err = match client
708 .post(
709 &format!("http://{addr}/v1/chat/completions"),
710 Vec::new(),
711 &serde_json::json!({"model": "test"}),
712 CancellationToken::new(),
713 )
714 .await
715 {
716 Ok(_) => panic!("expected API timeout error"),
717 Err(err) => err,
718 };
719
720 server.abort();
721 (started.elapsed(), err)
722 }
723
724 #[test]
725 #[cfg(not(windows))]
726 fn test_explicit_proxy_config_from_env_prefers_lowercase_vars() {
727 let _guard = proxy_env_lock().lock().unwrap();
728 clear_proxy_env();
729 unsafe {
730 env::set_var("http_proxy", "http://lower-http:3128");
731 env::set_var("HTTP_PROXY", "http://upper-http:3128");
732 env::set_var("https_proxy", "http://lower-https:3128");
733 env::set_var("HTTPS_PROXY", "http://upper-https:3128");
734 }
735
736 let proxy_config = explicit_proxy_config_from_env();
737
738 assert_eq!(
739 proxy_config,
740 ExplicitProxyConfig {
741 http: Some("http://lower-http:3128".to_string()),
742 https: Some("http://lower-https:3128".to_string()),
743 }
744 );
745 clear_proxy_env();
746 }
747
748 #[test]
749 fn test_explicit_proxy_config_from_env_falls_back_to_http_for_https() {
750 let _guard = proxy_env_lock().lock().unwrap();
751 clear_proxy_env();
752 unsafe {
753 env::set_var("HTTP_PROXY", "http://proxy.example:3128");
754 }
755
756 let proxy_config = explicit_proxy_config_from_env();
757
758 assert_eq!(
759 proxy_config,
760 ExplicitProxyConfig {
761 http: Some("http://proxy.example:3128".to_string()),
762 https: Some("http://proxy.example:3128".to_string()),
763 }
764 );
765 clear_proxy_env();
766 }
767
768 #[test]
769 fn test_build_reqwest_client_accepts_proxy_env_urls() {
770 let _guard = proxy_env_lock().lock().unwrap();
771 clear_proxy_env();
772 unsafe {
773 env::set_var("http_proxy", "http://127.0.0.1:3128");
774 env::set_var("https_proxy", "http://127.0.0.1:3128");
775 }
776
777 let client = build_reqwest_client(None, None);
778 assert!(client.is_ok());
779 clear_proxy_env();
780 }
781
782 #[test]
783 fn test_build_reqwest_client_accepts_no_proxy_with_proxies() {
784 let _guard = proxy_env_lock().lock().unwrap();
785 clear_proxy_env();
786 unsafe {
787 env::set_var("HTTP_PROXY", "http://proxy.example:3128");
788 env::set_var("HTTPS_PROXY", "http://proxy.example:3128");
789 env::set_var("NO_PROXY", "playwright-mcp,localhost,127.0.0.1");
790 }
791
792 assert!(
793 reqwest::NoProxy::from_env().is_some(),
794 "NO_PROXY must parse into a reqwest exclusion list"
795 );
796 let client = build_reqwest_client(None, None);
797 assert!(
798 client.is_ok(),
799 "client must build when proxies and NO_PROXY are both set"
800 );
801 clear_proxy_env();
802 }
803
804 #[test]
805 #[cfg(not(windows))]
806 fn test_no_proxy_from_env_prefers_lowercase_var() {
807 let _guard = proxy_env_lock().lock().unwrap();
808 clear_proxy_env();
809 unsafe {
810 env::set_var("no_proxy", "lower.example");
811 env::set_var("NO_PROXY", "upper.example");
812 }
813
814 assert!(reqwest::NoProxy::from_env().is_some());
817 clear_proxy_env();
818 }
819}