rustapi_mcp/
runner.rs

1//! Concurrent execution helpers to run a normal RustAPI HTTP server
2//! side-by-side with an MCP server (on a separate address).
3//!
4//! This is modeled after `rustapi-grpc` for consistency.
5
6use crate::server::McpServer;
7use rustapi_core::RustApi;
8use std::error::Error;
9use std::future::Future;
10use std::pin::Pin;
11use tokio::sync::watch;
12
13/// Boxed error type used by runner helpers.
14pub type BoxError = Box<dyn Error + Send + Sync>;
15
16/// Result type for runner helpers.
17pub type Result<T> = std::result::Result<T, BoxError>;
18
19/// Shutdown future type.
20pub type ShutdownFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
21
22fn to_boxed_error<E>(err: E) -> BoxError
23where
24    E: Error + Send + Sync + 'static,
25{
26    Box::new(err)
27}
28
29/// Run two independent futures concurrently (HTTP + MCP).
30pub async fn run_concurrently<HF, MF, HE, ME>(http_future: HF, mcp_future: MF) -> Result<()>
31where
32    HF: Future<Output = std::result::Result<(), HE>> + Send,
33    MF: Future<Output = std::result::Result<(), ME>> + Send,
34    HE: Error + Send + Sync + 'static,
35    ME: Error + Send + Sync + 'static,
36{
37    let http_task = async move { http_future.await.map_err(to_boxed_error) };
38    let mcp_task = async move { mcp_future.await.map_err(to_boxed_error) };
39
40    let (_http_ok, _mcp_ok) = tokio::try_join!(http_task, mcp_task)?;
41    Ok(())
42}
43
44/// Run a `RustApi` HTTP server and an `McpServer` side-by-side on separate addresses.
45///
46/// This is the primary helper for running your normal API and the MCP endpoint
47/// (for LLM/agent clients) at the same time.
48pub async fn run_rustapi_and_mcp(
49    app: RustApi,
50    http_addr: impl AsRef<str>,
51    mcp: McpServer,
52    mcp_addr: impl AsRef<str>,
53) -> Result<()>
54where
55{
56    let http_addr = http_addr.as_ref().to_string();
57    let mcp_addr = mcp_addr.as_ref().to_string();
58
59    // Automatically configure the MCP server to proxy tool calls back to the main HTTP API.
60    // This makes end-to-end tool invocation work out of the box.
61    let http_base = format!(
62        "http://127.0.0.1:{}",
63        extract_port_or_default(&http_addr, 8080)
64    );
65    let mcp = mcp.with_http_base(http_base);
66
67    let http_task = async move { app.run(&http_addr).await };
68    let mcp_task = async move { mcp.serve(&mcp_addr).await.map_err(to_boxed_error) };
69
70    let (_http_ok, _mcp_ok) = tokio::try_join!(http_task, mcp_task)?;
71    Ok(())
72}
73
74/// Run RustAPI HTTP + MCP with a shared shutdown signal.
75///
76/// Useful with `tokio::signal::ctrl_c()` so that Ctrl+C stops both servers cleanly.
77pub async fn run_rustapi_and_mcp_with_shutdown<SF>(
78    app: RustApi,
79    http_addr: impl AsRef<str>,
80    mcp: McpServer,
81    mcp_addr: impl AsRef<str>,
82    shutdown_signal: SF,
83) -> Result<()>
84where
85    SF: Future<Output = ()> + Send + 'static,
86{
87    let http_addr = http_addr.as_ref().to_string();
88    let mcp_addr = mcp_addr.as_ref().to_string();
89
90    let http_base = format!(
91        "http://127.0.0.1:{}",
92        extract_port_or_default(&http_addr, 8080)
93    );
94    let mcp = mcp.with_http_base(http_base);
95
96    let (shutdown_tx, shutdown_rx) = watch::channel(false);
97
98    let shutdown_dispatch = tokio::spawn(async move {
99        shutdown_signal.await;
100        let _ = shutdown_tx.send(true);
101    });
102
103    let http_shutdown = shutdown_notifier(shutdown_rx.clone());
104    let mcp_shutdown = shutdown_notifier(shutdown_rx);
105
106    let http_task = async move { app.run_with_shutdown(&http_addr, http_shutdown).await };
107
108    let mcp_task = async move {
109        mcp.serve_with_shutdown(&mcp_addr, mcp_shutdown)
110            .await
111            .map_err(to_boxed_error)
112    };
113
114    let joined = tokio::try_join!(http_task, mcp_task).map(|_| ());
115
116    shutdown_dispatch.abort();
117    let _ = shutdown_dispatch.await;
118
119    joined
120}
121
122async fn shutdown_notifier(mut rx: watch::Receiver<bool>) {
123    if *rx.borrow() {
124        return;
125    }
126    while rx.changed().await.is_ok() {
127        if *rx.borrow() {
128            break;
129        }
130    }
131}
132
133/// Very small helper to turn "0.0.0.0:9090" or "[::]:9090" into a port for the localhost base URL.
134fn extract_port_or_default(addr: &str, default: u16) -> u16 {
135    // Try to find the last ':' and parse what follows as port
136    if let Some(colon) = addr.rfind(':') {
137        let after = &addr[colon + 1..];
138        // strip any trailing path or query (shouldn't be there for addr)
139        let port_str = after
140            .split(|c: char| !c.is_ascii_digit())
141            .next()
142            .unwrap_or("");
143        if let Ok(p) = port_str.parse::<u16>() {
144            return p;
145        }
146    }
147    default
148}
rustapi_mcp/runner.rs

rustapi_mcp/
runner.rs
