1use std::time::Duration;
32
33use crate::tool_executor::ToolResult;
34use crate::tool_registry::ToolEntry;
35
36fn parse_timeout(s: &str) -> Option<Duration> {
41 let s = s.trim();
42 if s.is_empty() {
43 return None;
44 }
45
46 if let Some(secs) = s.strip_suffix("ms") {
47 secs.trim().parse::<u64>().ok().map(Duration::from_millis)
48 } else if let Some(secs) = s.strip_suffix('s') {
49 secs.trim().parse::<u64>().ok().map(Duration::from_secs)
50 } else if let Some(mins) = s.strip_suffix('m') {
51 mins.trim()
52 .parse::<u64>()
53 .ok()
54 .map(|m| Duration::from_secs(m * 60))
55 } else {
56 s.parse::<u64>().ok().map(Duration::from_secs)
58 }
59}
60
61pub fn parse_timeout_pub(s: &str) -> Option<Duration> {
63 parse_timeout(s)
64}
65
66const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
67
68pub fn dispatch_http(entry: &ToolEntry, argument: &str) -> ToolResult {
78 let url = entry.runtime.trim();
79
80 if url.is_empty() {
81 return ToolResult {
82 success: false,
83 output: format!(
84 "HTTP tool '{}': no endpoint URL. Set runtime: \"https://...\" in tool definition.",
85 entry.name
86 ),
87 tool_name: entry.name.clone(),
88 };
89 }
90
91 if !url.starts_with("http://") && !url.starts_with("https://") {
93 return ToolResult {
94 success: false,
95 output: format!(
96 "HTTP tool '{}': invalid URL '{}'. Must start with http:// or https://.",
97 entry.name, url
98 ),
99 tool_name: entry.name.clone(),
100 };
101 }
102
103 let timeout = parse_timeout(&entry.timeout).unwrap_or(DEFAULT_TIMEOUT);
104
105 let body = if argument.trim_start().starts_with('{') || argument.trim_start().starts_with('[') {
107 argument.to_string()
108 } else {
109 serde_json::json!({ "input": argument }).to_string()
110 };
111
112 match execute_request(url, &entry.name, &body, timeout) {
114 Ok(response) => response,
115 Err(e) => ToolResult {
116 success: false,
117 output: format!("HTTP tool '{}': {}", entry.name, e),
118 tool_name: entry.name.clone(),
119 },
120 }
121}
122
123fn execute_request(
125 url: &str,
126 tool_name: &str,
127 body: &str,
128 timeout: Duration,
129) -> Result<ToolResult, String> {
130 let client = reqwest::blocking::Client::builder()
131 .timeout(timeout)
132 .build()
133 .map_err(|e| format!("failed to create HTTP client: {e}"))?;
134
135 let response = client
136 .post(url)
137 .header("Content-Type", "application/json")
138 .header("X-Axon-Tool", tool_name)
139 .body(body.to_string())
140 .send()
141 .map_err(|e| {
142 if e.is_timeout() {
143 format!("request timed out after {}s", timeout.as_secs())
144 } else if e.is_connect() {
145 format!("connection failed to {url}")
146 } else {
147 format!("request failed: {e}")
148 }
149 })?;
150
151 let status = response.status();
152 let response_body = response
153 .text()
154 .map_err(|e| format!("failed to read response body: {e}"))?;
155
156 if status.is_success() {
157 Ok(ToolResult {
158 success: true,
159 output: response_body,
160 tool_name: tool_name.to_string(),
161 })
162 } else {
163 Ok(ToolResult {
164 success: false,
165 output: format!(
166 "HTTP {}: {}",
167 status.as_u16(),
168 if response_body.len() > 200 {
169 format!("{}...", &response_body[..200])
170 } else {
171 response_body
172 }
173 ),
174 tool_name: tool_name.to_string(),
175 })
176 }
177}
178
179use async_trait::async_trait;
185use bytes::Bytes;
186use futures::StreamExt;
187
188use crate::backends::sse_streaming::{LineBuffer, SseEventParser};
189use crate::tool_trait::{Tool, ToolChunk, ToolContext, ToolFinishReason, ToolStream};
190
191pub struct HttpStreamingTool {
228 name: String,
229 url: String,
230 timeout: Duration,
231}
232
233impl HttpStreamingTool {
234 pub fn from_entry(entry: &ToolEntry) -> Result<Self, String> {
238 let url = entry.runtime.trim();
239 if url.is_empty() {
240 return Err(format!(
241 "HTTP tool '{}': no endpoint URL. Set runtime: \"https://...\" in tool definition.",
242 entry.name
243 ));
244 }
245 if !url.starts_with("http://") && !url.starts_with("https://") {
246 return Err(format!(
247 "HTTP tool '{}': invalid URL '{}'. Must start with http:// or https://.",
248 entry.name, url
249 ));
250 }
251 let timeout = parse_timeout(&entry.timeout).unwrap_or(DEFAULT_TIMEOUT);
252 Ok(Self {
253 name: entry.name.clone(),
254 url: url.to_string(),
255 timeout,
256 })
257 }
258
259 pub fn new(name: String, url: String, timeout: Duration) -> Self {
262 Self { name, url, timeout }
263 }
264}
265
266fn build_request_body(args: &str) -> String {
269 let trimmed = args.trim_start();
270 if trimmed.starts_with('{') || trimmed.starts_with('[') {
271 args.to_string()
272 } else {
273 serde_json::json!({ "input": args }).to_string()
274 }
275}
276
277#[derive(Debug, Clone, Copy, PartialEq, Eq)]
280enum FramingMode {
281 Sse,
285 Ndjson,
288 Single,
292}
293
294fn classify_framing(content_type: &str) -> FramingMode {
295 let lc = content_type.to_ascii_lowercase();
296 if lc.contains("text/event-stream") {
297 FramingMode::Sse
298 } else if lc.contains("application/x-ndjson") || lc.contains("application/jsonl") {
299 FramingMode::Ndjson
300 } else {
301 FramingMode::Single
302 }
303}
304
305#[async_trait]
306impl Tool for HttpStreamingTool {
307 async fn execute(&self, args: String, _ctx: ToolContext) -> ToolResult {
308 let entry = ToolEntry {
318 name: self.name.clone(),
319 provider: "http".to_string(),
320 timeout: format!("{}s", self.timeout.as_secs()),
321 runtime: self.url.clone(),
322 sandbox: None,
323 max_results: None,
324 output_schema: String::new(),
325 effect_row: Vec::new(),
326 parameters: Vec::new(),
329 secret: String::new(),
330 secret_partition: String::new(),
331 source: crate::tool_registry::ToolSource::Program,
332 is_streaming: false,
333 scrape: None,
334 };
335 match tokio::task::spawn_blocking(move || dispatch_http(&entry, &args)).await {
336 Ok(result) => result,
337 Err(e) => ToolResult {
338 success: false,
339 output: format!("HTTP tool '{}': blocking task join failed: {e}", self.name),
340 tool_name: self.name.clone(),
341 },
342 }
343 }
344
345 async fn stream(&self, args: String, ctx: ToolContext) -> ToolStream {
346 let url = self.url.clone();
347 let name = self.name.clone();
348 let timeout = self.timeout;
349 let cancel = ctx.cancel.clone();
350 let body = build_request_body(&args);
351
352 let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<ToolChunk>();
358
359 tokio::spawn(async move {
360 let send_terminator = |reason: ToolFinishReason| {
365 let _ = tx.send(ToolChunk::terminator("", reason));
366 };
367
368 if cancel.is_cancelled() {
370 send_terminator(ToolFinishReason::Cancelled);
371 return;
372 }
373
374 let client = match reqwest::Client::builder().timeout(timeout).build() {
376 Ok(c) => c,
377 Err(e) => {
378 send_terminator(ToolFinishReason::Error {
379 message: format!(
380 "HTTP tool '{name}': failed to build async client: {e}"
381 ),
382 });
383 return;
384 }
385 };
386
387 let response = match client
389 .post(&url)
390 .header("Content-Type", "application/json")
391 .header("X-Axon-Tool", &name)
392 .body(body)
393 .send()
394 .await
395 {
396 Ok(r) => r,
397 Err(e) => {
398 let message = if e.is_timeout() {
399 format!(
400 "HTTP tool '{name}': request timed out after {}s",
401 timeout.as_secs()
402 )
403 } else if e.is_connect() {
404 format!("HTTP tool '{name}': connection failed to {url}")
405 } else {
406 format!("HTTP tool '{name}': request failed: {e}")
407 };
408 send_terminator(ToolFinishReason::Error { message });
409 return;
410 }
411 };
412
413 let status = response.status();
417 if !status.is_success() {
418 let body_text = response.text().await.unwrap_or_default();
419 let truncated = if body_text.len() > 200 {
420 format!("{}...", &body_text[..200])
421 } else {
422 body_text
423 };
424 send_terminator(ToolFinishReason::Error {
425 message: format!("HTTP {}: {}", status.as_u16(), truncated),
426 });
427 return;
428 }
429
430 let content_type = response
432 .headers()
433 .get(reqwest::header::CONTENT_TYPE)
434 .and_then(|v| v.to_str().ok())
435 .unwrap_or("")
436 .to_string();
437 let framing = classify_framing(&content_type);
438
439 let mut byte_stream = response.bytes_stream();
441 let drain_result = match framing {
442 FramingMode::Sse => {
443 drain_sse(&mut byte_stream, &cancel, &tx).await
444 }
445 FramingMode::Ndjson => {
446 drain_ndjson(&mut byte_stream, &cancel, &tx).await
447 }
448 FramingMode::Single => {
449 drain_single(&mut byte_stream, &cancel, &tx).await
450 }
451 };
452
453 match drain_result {
454 DrainOutcome::Completed => send_terminator(ToolFinishReason::Stop),
455 DrainOutcome::Cancelled => send_terminator(ToolFinishReason::Cancelled),
456 DrainOutcome::Error(message) => {
457 send_terminator(ToolFinishReason::Error { message })
458 }
459 }
460 });
461
462 Box::pin(futures::stream::unfold(rx, |mut rx| async move {
466 rx.recv().await.map(|chunk| (chunk, rx))
467 }))
468 }
469
470 fn is_streaming(&self) -> bool {
471 true
472 }
473}
474
475enum DrainOutcome {
478 Completed,
479 Cancelled,
480 Error(String),
481}
482
483async fn drain_sse<S>(
489 byte_stream: &mut S,
490 cancel: &crate::cancel_token::CancellationFlag,
491 tx: &tokio::sync::mpsc::UnboundedSender<ToolChunk>,
492) -> DrainOutcome
493where
494 S: futures::Stream<Item = reqwest::Result<Bytes>> + Unpin + Send,
495{
496 let mut line_buf = LineBuffer::new();
497 let mut sse_parser = SseEventParser::new();
498 loop {
499 if cancel.is_cancelled() {
500 return DrainOutcome::Cancelled;
501 }
502 match byte_stream.next().await {
503 None => break,
504 Some(Err(e)) => {
505 return DrainOutcome::Error(format!("SSE stream chunk error: {e}"))
506 }
507 Some(Ok(bytes)) => {
508 let lines = line_buf.push(&bytes);
509 for line in lines {
510 if let Some(event) = sse_parser.push_line(&line) {
511 if let Some(data) = event.data {
512 if tx
513 .send(ToolChunk::intermediate(data))
514 .is_err()
515 {
516 return DrainOutcome::Cancelled;
517 }
518 }
519 }
520 }
521 }
522 }
523 }
524 if let Some(line) = line_buf.flush() {
528 if let Some(event) = sse_parser.push_line(&line) {
529 if let Some(data) = event.data {
530 let _ = tx.send(ToolChunk::intermediate(data));
531 }
532 }
533 }
534 DrainOutcome::Completed
535}
536
537async fn drain_ndjson<S>(
541 byte_stream: &mut S,
542 cancel: &crate::cancel_token::CancellationFlag,
543 tx: &tokio::sync::mpsc::UnboundedSender<ToolChunk>,
544) -> DrainOutcome
545where
546 S: futures::Stream<Item = reqwest::Result<Bytes>> + Unpin + Send,
547{
548 let mut line_buf = LineBuffer::new();
549 loop {
550 if cancel.is_cancelled() {
551 return DrainOutcome::Cancelled;
552 }
553 match byte_stream.next().await {
554 None => break,
555 Some(Err(e)) => {
556 return DrainOutcome::Error(format!("NDJSON stream chunk error: {e}"))
557 }
558 Some(Ok(bytes)) => {
559 let lines = line_buf.push(&bytes);
560 for line in lines {
561 if !line.is_empty()
562 && tx.send(ToolChunk::intermediate(line)).is_err()
563 {
564 return DrainOutcome::Cancelled;
565 }
566 }
567 }
568 }
569 }
570 if let Some(line) = line_buf.flush() {
571 if !line.is_empty() {
572 let _ = tx.send(ToolChunk::intermediate(line));
573 }
574 }
575 DrainOutcome::Completed
576}
577
578async fn drain_single<S>(
582 byte_stream: &mut S,
583 cancel: &crate::cancel_token::CancellationFlag,
584 tx: &tokio::sync::mpsc::UnboundedSender<ToolChunk>,
585) -> DrainOutcome
586where
587 S: futures::Stream<Item = reqwest::Result<Bytes>> + Unpin + Send,
588{
589 let mut acc: Vec<u8> = Vec::new();
590 loop {
591 if cancel.is_cancelled() {
592 return DrainOutcome::Cancelled;
593 }
594 match byte_stream.next().await {
595 None => break,
596 Some(Err(e)) => {
597 return DrainOutcome::Error(format!("HTTP body chunk error: {e}"))
598 }
599 Some(Ok(bytes)) => acc.extend_from_slice(&bytes),
600 }
601 }
602 let body_text = String::from_utf8_lossy(&acc).into_owned();
603 if !body_text.is_empty()
604 && tx
605 .send(ToolChunk::intermediate(body_text))
606 .is_err()
607 {
608 return DrainOutcome::Cancelled;
609 }
610 DrainOutcome::Completed
611}
612
613#[cfg(test)]
616mod tests {
617 use super::*;
618 use crate::tool_registry::{ToolEntry, ToolSource};
619
620 fn make_http_entry(name: &str, url: &str, timeout: &str) -> ToolEntry {
621 ToolEntry {
622 name: name.to_string(),
623 provider: "http".to_string(),
624 timeout: timeout.to_string(),
625 runtime: url.to_string(),
626 sandbox: None,
627 max_results: None,
628 output_schema: "JSON".to_string(),
629 effect_row: vec!["network".to_string()],
630 parameters: Vec::new(),
631 secret: String::new(),
632 secret_partition: String::new(),
633 source: ToolSource::Program,
634 is_streaming: false,
638 scrape: None,
639 }
640 }
641
642 #[test]
645 fn parse_timeout_seconds() {
646 assert_eq!(parse_timeout("10s"), Some(Duration::from_secs(10)));
647 assert_eq!(parse_timeout("30s"), Some(Duration::from_secs(30)));
648 }
649
650 #[test]
651 fn parse_timeout_milliseconds() {
652 assert_eq!(parse_timeout("500ms"), Some(Duration::from_millis(500)));
653 assert_eq!(parse_timeout("100ms"), Some(Duration::from_millis(100)));
654 }
655
656 #[test]
657 fn parse_timeout_minutes() {
658 assert_eq!(parse_timeout("2m"), Some(Duration::from_secs(120)));
659 }
660
661 #[test]
662 fn parse_timeout_raw_number() {
663 assert_eq!(parse_timeout("15"), Some(Duration::from_secs(15)));
664 }
665
666 #[test]
667 fn parse_timeout_empty() {
668 assert_eq!(parse_timeout(""), None);
669 assert_eq!(parse_timeout(" "), None);
670 }
671
672 #[test]
673 fn parse_timeout_invalid() {
674 assert_eq!(parse_timeout("abc"), None);
675 assert_eq!(parse_timeout("10x"), None);
676 }
677
678 #[test]
681 fn dispatch_empty_url_fails() {
682 let entry = make_http_entry("DataAPI", "", "10s");
683 let result = dispatch_http(&entry, "test query");
684 assert!(!result.success);
685 assert!(result.output.contains("no endpoint URL"));
686 }
687
688 #[test]
689 fn dispatch_invalid_url_scheme_fails() {
690 let entry = make_http_entry("DataAPI", "ftp://example.com", "10s");
691 let result = dispatch_http(&entry, "test query");
692 assert!(!result.success);
693 assert!(result.output.contains("invalid URL"));
694 assert!(result.output.contains("http://"));
695 }
696
697 #[test]
700 fn dispatch_connection_refused() {
701 let entry = make_http_entry("TestTool", "http://127.0.0.1:1/api", "2s");
703 let result = dispatch_http(&entry, "test");
704 assert!(!result.success);
705 assert!(
706 result.output.contains("connection failed")
707 || result.output.contains("request failed")
708 || result.output.contains("timed out"),
709 "unexpected error: {}",
710 result.output
711 );
712 }
713
714 #[test]
717 fn json_body_passthrough() {
718 let arg = r#"{"query": "test"}"#;
720 let body = if arg.trim_start().starts_with('{') {
721 arg.to_string()
722 } else {
723 serde_json::json!({ "input": arg }).to_string()
724 };
725 assert_eq!(body, r#"{"query": "test"}"#);
726 }
727
728 #[test]
729 fn plain_text_wrapped() {
730 let arg = "search for cats";
732 let body = if arg.trim_start().starts_with('{') || arg.trim_start().starts_with('[') {
733 arg.to_string()
734 } else {
735 serde_json::json!({ "input": arg }).to_string()
736 };
737 let parsed: serde_json::Value = serde_json::from_str(&body).unwrap();
738 assert_eq!(parsed["input"], "search for cats");
739 }
740
741 #[test]
742 fn array_body_passthrough() {
743 let arg = r#"[1, 2, 3]"#;
744 let body = if arg.trim_start().starts_with('{') || arg.trim_start().starts_with('[') {
745 arg.to_string()
746 } else {
747 serde_json::json!({ "input": arg }).to_string()
748 };
749 assert_eq!(body, "[1, 2, 3]");
750 }
751}