rust_web_server/proxy_config/
mcp_gateway.rs

1//! `[route.action] type = "mcp"`: the [MCP gateway](crate::mcp::gateway)
2//! from the config file (18.9.0).
3//!
4//! ```toml
5//! [[route]]
6//!   [route.match]
7//!   path = "/mcp"
8//!
9//!   [route.action]
10//!   type = "mcp"
11//!
12//!   [route.action.mcp]
13//!   name        = "gateway"     # announced to clients
14//!   refresh_secs = 300          # re-read upstream tools; 0 = only at start
15//!   audit_log   = true
16//!   allowed_origins = ["https://app.example.com"]
17//!
18//!   [[route.action.mcp.upstream]]
19//!   prefix      = "github"
20//!   url         = "https://mcp.github.example/mcp"
21//!   timeout_ms  = 30000
22//!   headers_env = { Authorization = "GITHUB_MCP_AUTHORIZATION" }  # header = env var holding its value
23//!
24//!   [[route.action.mcp.openapi]]               # an HTTP API's operations as tools (18.19.0)
25//!   prefix      = "billing"
26//!   spec        = "https://billing.internal/openapi.json"   # or a file path
27//!   base_url    = "https://billing.internal"                # calls go only here
28//!   include     = ["getInvoice", "listInvoices"]            # operationIds; default: every GET
29//!   headers_env = { Authorization = "BILLING_API_AUTHORIZATION" }
30//!
31//!   [[route.action.mcp.caller]]
32//!   identity  = "alice"
33//!   token_env = "ALICE_MCP_TOKEN"       # the bearer token alice sends
34//!
35//!   [[route.action.mcp.allow]]
36//!   identity = "alice"
37//!   tools    = ["github__*"]
38//!
39//!   [[route.action.mcp.limit]]
40//!   tools       = "github__create_issue"
41//!   max_calls   = 5
42//!   window_secs = 3600
43//! ```
44//!
45//! Secrets never sit in the file: caller tokens and upstream header values
46//! are read from the environment variables named. The gateway is built in
47//! the background (it lists every upstream's tools); until then, and while
48//! an upstream can't be reached at startup, requests answer 503 and the
49//! build is retried every 10 s.
50
51use std::sync::{Arc, Mutex, Weak};
52use std::time::Duration;
53
54use crate::application::Application;
55use crate::config_check::ConfigReport;
56use crate::mcp::gateway::{McpGateway, Upstream};
57use crate::mcp::McpServer;
58use crate::request::Request;
59use crate::response::Response;
60use crate::server::ConnectionInfo;
61
62use super::parser::{get_array, get_str, get_u32, get_u64, section_exists, SectionMap};
63
64/// How long a failed build waits before the next attempt.
65const RETRY: Duration = Duration::from_secs(10);
66
67#[derive(Debug, Clone, Default, PartialEq)]
68pub struct McpGatewayConfig {
69    pub name: String,
70    pub refresh_secs: u64,
71    pub audit_log: bool,
72    pub allowed_origins: Vec<String>,
73    pub upstreams: Vec<UpstreamConfig>,
74    pub apis: Vec<OpenApiConfig>,
75    pub callers: Vec<CallerConfig>,
76    pub allow: Vec<(String, Vec<String>)>,
77    pub limits: Vec<LimitConfig>,
78}
79
80#[derive(Debug, Clone, Default, PartialEq)]
81pub struct UpstreamConfig {
82    pub prefix: String,
83    pub url: String,
84    pub timeout_ms: u64,
85    /// `(header, environment variable holding its value)`.
86    pub headers_env: Vec<(String, String)>,
87}
88
89/// `[[route.action.mcp.openapi]]`: an HTTP API's operations as tools.
90#[derive(Debug, Clone, Default, PartialEq)]
91pub struct OpenApiConfig {
92    pub prefix: String,
93    pub spec: String,
94    pub base_url: String,
95    pub include: Vec<String>,
96    pub exclude: Vec<String>,
97    pub timeout_ms: u64,
98    pub headers_env: Vec<(String, String)>,
99}
100
101#[derive(Debug, Clone, Default, PartialEq)]
102pub struct CallerConfig {
103    pub identity: String,
104    pub token_env: String,
105}
106
107#[derive(Debug, Clone, Default, PartialEq)]
108pub struct LimitConfig {
109    pub tools: String,
110    pub max_calls: u32,
111    pub window_secs: u64,
112}
113
114/// The keys `check_unknown_keys` accepts in each part.
115pub(crate) const GATEWAY_KEYS: &[&str] = &["name", "refresh_secs", "audit_log", "allowed_origins"];
116pub(crate) const UPSTREAM_KEYS: &[&str] = &["prefix", "url", "timeout_ms"];
117pub(crate) const CALLER_KEYS: &[&str] = &["identity", "token_env"];
118pub(crate) const OPENAPI_KEYS: &[&str] = &["prefix", "spec", "base_url", "include", "exclude", "timeout_ms"];
119pub(crate) const ALLOW_KEYS: &[&str] = &["identity", "tools"];
120pub(crate) const LIMIT_KEYS: &[&str] = &["tools", "max_calls", "window_secs"];
121
122/// `section` is the action's, e.g. `route[0].action`.
123pub(crate) fn parse(map: &SectionMap, section: &str) -> McpGatewayConfig {
124    let base = format!("{section}.mcp");
125    let indexed = |kind: &str| -> Vec<String> {
126        (0..).map(|i| format!("{base}.{kind}[{i}]")).take_while(|s| section_exists(map, s)).collect()
127    };
128    let name = get_str(map, &base, "name");
129    McpGatewayConfig {
130        name: if name.is_empty() { "rws-mcp-gateway".to_string() } else { name },
131        refresh_secs: get_u64(map, &base, "refresh_secs", crate::mcp::gateway::DEFAULT_REFRESH.as_secs()),
132        audit_log: get_str(map, &base, "audit_log").eq_ignore_ascii_case("true"),
133        allowed_origins: get_array(map, &base, "allowed_origins"),
134        upstreams: indexed("upstream")
135            .into_iter()
136            .map(|s| UpstreamConfig {
137                prefix: get_str(map, &s, "prefix"),
138                url: get_str(map, &s, "url"),
139                timeout_ms: get_u64(map, &s, "timeout_ms", 30_000),
140                headers_env: map.get(&format!("{s}.headers_env")).cloned().unwrap_or_default(),
141            })
142            .collect(),
143        apis: indexed("openapi")
144            .into_iter()
145            .map(|s| OpenApiConfig {
146                prefix: get_str(map, &s, "prefix"),
147                spec: get_str(map, &s, "spec"),
148                base_url: get_str(map, &s, "base_url"),
149                include: get_array(map, &s, "include"),
150                exclude: get_array(map, &s, "exclude"),
151                timeout_ms: get_u64(map, &s, "timeout_ms", 30_000),
152                headers_env: map.get(&format!("{s}.headers_env")).cloned().unwrap_or_default(),
153            })
154            .collect(),
155        callers: indexed("caller")
156            .into_iter()
157            .map(|s| CallerConfig { identity: get_str(map, &s, "identity"), token_env: get_str(map, &s, "token_env") })
158            .collect(),
159        allow: indexed("allow").into_iter().map(|s| (get_str(map, &s, "identity"), get_array(map, &s, "tools"))).collect(),
160        limits: indexed("limit")
161            .into_iter()
162            .map(|s| LimitConfig { tools: get_str(map, &s, "tools"), max_calls: get_u32(map, &s, "max_calls", 0), window_secs: get_u64(map, &s, "window_secs", 0) })
163            .collect(),
164    }
165}
166
167/// Semantic checks for `rws --check` and startup.
168pub(crate) fn validate(config: &McpGatewayConfig, route_path: Option<&str>, label: &str, report: &mut ConfigReport) {
169    let offline = crate::config_check::offline();
170    match route_path {
171        Some(path) if path.ends_with('*') => report.error(format!("{label}: an mcp route needs an exact [route.match] path, e.g. \"/mcp\" (not {path:?})")),
172        None => report.error(format!("{label}: an mcp route needs [route.match] path, e.g. \"/mcp\"")),
173        _ => {}
174    }
175    if config.upstreams.is_empty() && config.apis.is_empty() {
176        report.error(format!("{label}: [route.action.mcp] has no [[route.action.mcp.upstream]] or [[route.action.mcp.openapi]]"));
177    }
178    let mut prefixes = Vec::new();
179    for (i, u) in config.upstreams.iter().enumerate() {
180        let what = format!("{label}: upstream #{}", i + 1);
181        if u.prefix.is_empty() || u.prefix.contains("__") || u.prefix.ends_with('_') || !u.prefix.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
182            report.error(format!("{what}: prefix {:?} must be letters, digits, '-' and single '_'", u.prefix));
183        } else if prefixes.contains(&u.prefix) {
184            report.error(format!("{what}: prefix {:?} is used twice", u.prefix));
185        } else {
186            prefixes.push(u.prefix.clone());
187        }
188        if !(u.url.starts_with("http://") || u.url.starts_with("https://")) {
189            report.error(format!("{what}: url {:?} must start with http:// or https://", u.url));
190        }
191        for (header, env) in &u.headers_env {
192            if !offline && std::env::var(env).unwrap_or_default().is_empty() {
193                report.error(format!("{what}: header {header} reads {env}, which is unset or empty"));
194            }
195        }
196    }
197    for (i, a) in config.apis.iter().enumerate() {
198        let what = format!("{label}: openapi #{}", i + 1);
199        if a.prefix.is_empty() || a.prefix.contains("__") || a.prefix.ends_with('_') || !a.prefix.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
200            report.error(format!("{what}: prefix {:?} must be letters, digits, '-' and single '_'", a.prefix));
201        } else if prefixes.contains(&a.prefix) {
202            report.error(format!("{what}: prefix {:?} is used twice", a.prefix));
203        } else {
204            prefixes.push(a.prefix.clone());
205        }
206        if a.spec.is_empty() {
207            report.error(format!("{what}: spec (an OpenAPI document's URL or file) is required"));
208        } else if !offline && !a.spec.starts_with("http") && !std::path::Path::new(&a.spec).is_file() {
209            report.error(format!("{what}: spec file {:?} doesn't exist", a.spec));
210        }
211        if !(a.base_url.starts_with("http://") || a.base_url.starts_with("https://")) {
212            report.error(format!("{what}: base_url {:?} must start with http:// or https://", a.base_url));
213        }
214        for (header, env) in &a.headers_env {
215            if !offline && std::env::var(env).unwrap_or_default().is_empty() {
216                report.error(format!("{what}: header {header} reads {env}, which is unset or empty"));
217            }
218        }
219    }
220    if config.callers.is_empty() {
221        report.error(format!("{label}: no [[route.action.mcp.caller]]: nobody can call the gateway"));
222    }
223    for c in &config.callers {
224        if c.identity.is_empty() || c.token_env.is_empty() {
225            report.error(format!("{label}: a caller needs identity and token_env"));
226        } else if !offline && std::env::var(&c.token_env).unwrap_or_default().trim().is_empty() {
227            report.error(format!("{label}: caller {} reads its token from {}, which is unset or empty", c.identity, c.token_env));
228        }
229    }
230    if config.allow.is_empty() {
231        report.warn(format!("{label}: no [[route.action.mcp.allow]] rule: callers see no tools"));
232    }
233    for (identity, tools) in &config.allow {
234        if identity.is_empty() || tools.is_empty() {
235            report.error(format!("{label}: an allow rule needs identity and tools"));
236        }
237    }
238    for l in &config.limits {
239        if l.tools.is_empty() || l.max_calls == 0 || l.window_secs == 0 {
240            report.error(format!("{label}: a limit needs tools, max_calls and window_secs above 0"));
241        }
242    }
243}
244
245/// The gateway from `config` (lists every upstream's tools: network).
246fn build(config: &McpGatewayConfig, path: &str) -> Result<McpServer, String> {
247    let mut gateway = McpGateway::new(config.name.clone(), env!("CARGO_PKG_VERSION"))
248        .refresh_every((config.refresh_secs > 0).then(|| Duration::from_secs(config.refresh_secs)))
249        .audit_log(config.audit_log);
250    for u in &config.upstreams {
251        let mut upstream = Upstream::new(u.prefix.clone(), u.url.clone()).timeout_ms(u.timeout_ms);
252        for (header, env) in &u.headers_env {
253            if let Ok(value) = std::env::var(env) {
254                upstream = upstream.header(header.clone(), value);
255            }
256        }
257        gateway = gateway.upstream(upstream);
258    }
259    for a in &config.apis {
260        let mut api = crate::mcp::gateway::OpenApiUpstream::new(a.prefix.clone(), a.spec.clone(), a.base_url.clone()).include(a.include.clone()).exclude(a.exclude.clone()).timeout_ms(a.timeout_ms);
261        for (header, env) in &a.headers_env {
262            if let Ok(value) = std::env::var(env) {
263                api = api.header(header.clone(), value);
264            }
265        }
266        gateway = gateway.openapi(api);
267    }
268    for (identity, tools) in &config.allow {
269        gateway = gateway.allow(identity.clone(), tools.clone());
270    }
271    for l in &config.limits {
272        gateway = gateway.limit(l.tools.clone(), l.max_calls, l.window_secs);
273    }
274    let tokens: Vec<(String, String)> = config
275        .callers
276        .iter()
277        .filter_map(|c| std::env::var(&c.token_env).ok().filter(|t| !t.trim().is_empty()).map(|t| (t, c.identity.clone())))
278        .collect();
279    let mut server = gateway.build()?.at(path).require_bearer_tokens(tokens);
280    if !config.allowed_origins.is_empty() {
281        server = server.allowed_origins(config.allowed_origins.clone());
282    }
283    Ok(server)
284}
285
286/// The route's [`Application`]: the gateway once built, 503 until then.
287pub(crate) struct GatewayAdapter {
288    slot: Arc<Mutex<Option<McpServer>>>,
289}
290
291impl GatewayAdapter {
292    /// Starts building in the background; retries every [`RETRY`] until it
293    /// works or the adapter is dropped (a reload replaced the route).
294    pub(crate) fn start(config: McpGatewayConfig, path: String) -> GatewayAdapter {
295        let slot: Arc<Mutex<Option<McpServer>>> = Arc::new(Mutex::new(None));
296        let weak: Weak<Mutex<Option<McpServer>>> = Arc::downgrade(&slot);
297        let spawned = std::thread::Builder::new().name("rws-mcp-gateway-build".to_string()).spawn(move || loop {
298            match build(&config, &path) {
299                Ok(server) => {
300                    if let Some(slot) = weak.upgrade() {
301                        *slot.lock().unwrap_or_else(|e| e.into_inner()) = Some(server);
302                        println!("mcp gateway at {path}: {} upstream(s) federated", config.upstreams.len());
303                    }
304                    return;
305                }
306                Err(e) => {
307                    eprintln!("mcp gateway at {path}: {e}; retrying in {} s", RETRY.as_secs());
308                    std::thread::sleep(RETRY);
309                    if weak.strong_count() == 0 {
310                        return;
311                    }
312                }
313            }
314        });
315        if let Err(e) = spawned {
316            eprintln!("mcp gateway: not started: {e}");
317        }
318        GatewayAdapter { slot }
319    }
320}
321
322impl Application for GatewayAdapter {
323    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
324        let server = self.slot.lock().unwrap_or_else(|e| e.into_inner()).clone();
325        match server {
326            Some(server) => server.execute(request, connection),
327            None => Ok(unavailable()),
328        }
329    }
330}
331
332fn unavailable() -> Response {
333    let mut response = super::service_unavailable();
334    response.headers.retain(|h| !h.name.eq_ignore_ascii_case("retry-after"));
335    response.headers.push(crate::header::Header::new("Retry-After", RETRY.as_secs().to_string()));
336    response
337}
338
339#[cfg(test)]
340mod tests;
rust_web_server/proxy_config/mcp_gateway.rs

rust_web_server/proxy_config/
mcp_gateway.rs
