rws_proxy/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.
115#[doc(hidden)]
116pub const GATEWAY_KEYS: &[&str] = &["name", "refresh_secs", "audit_log", "allowed_origins"];
117#[doc(hidden)]
118pub const UPSTREAM_KEYS: &[&str] = &["prefix", "url", "timeout_ms"];
119#[doc(hidden)]
120pub const CALLER_KEYS: &[&str] = &["identity", "token_env"];
121#[doc(hidden)]
122pub const OPENAPI_KEYS: &[&str] = &["prefix", "spec", "base_url", "include", "exclude", "timeout_ms"];
123#[doc(hidden)]
124pub const ALLOW_KEYS: &[&str] = &["identity", "tools"];
125#[doc(hidden)]
126pub const LIMIT_KEYS: &[&str] = &["tools", "max_calls", "window_secs"];
127
128/// `section` is the action's, e.g. `route[0].action`.
129#[doc(hidden)]
130pub fn parse(map: &SectionMap, section: &str) -> McpGatewayConfig {
131    let base = format!("{section}.mcp");
132    let indexed = |kind: &str| -> Vec<String> {
133        (0..).map(|i| format!("{base}.{kind}[{i}]")).take_while(|s| section_exists(map, s)).collect()
134    };
135    let name = get_str(map, &base, "name");
136    McpGatewayConfig {
137        name: if name.is_empty() { "rws-mcp-gateway".to_string() } else { name },
138        refresh_secs: get_u64(map, &base, "refresh_secs", crate::mcp::gateway::DEFAULT_REFRESH.as_secs()),
139        audit_log: get_str(map, &base, "audit_log").eq_ignore_ascii_case("true"),
140        allowed_origins: get_array(map, &base, "allowed_origins"),
141        upstreams: indexed("upstream")
142            .into_iter()
143            .map(|s| UpstreamConfig {
144                prefix: get_str(map, &s, "prefix"),
145                url: get_str(map, &s, "url"),
146                timeout_ms: get_u64(map, &s, "timeout_ms", 30_000),
147                headers_env: map.get(&format!("{s}.headers_env")).cloned().unwrap_or_default(),
148            })
149            .collect(),
150        apis: indexed("openapi")
151            .into_iter()
152            .map(|s| OpenApiConfig {
153                prefix: get_str(map, &s, "prefix"),
154                spec: get_str(map, &s, "spec"),
155                base_url: get_str(map, &s, "base_url"),
156                include: get_array(map, &s, "include"),
157                exclude: get_array(map, &s, "exclude"),
158                timeout_ms: get_u64(map, &s, "timeout_ms", 30_000),
159                headers_env: map.get(&format!("{s}.headers_env")).cloned().unwrap_or_default(),
160            })
161            .collect(),
162        callers: indexed("caller")
163            .into_iter()
164            .map(|s| CallerConfig { identity: get_str(map, &s, "identity"), token_env: get_str(map, &s, "token_env") })
165            .collect(),
166        allow: indexed("allow").into_iter().map(|s| (get_str(map, &s, "identity"), get_array(map, &s, "tools"))).collect(),
167        limits: indexed("limit")
168            .into_iter()
169            .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) })
170            .collect(),
171    }
172}
173
174/// Semantic checks for `rws --check` and startup.
175#[doc(hidden)]
176pub fn validate(config: &McpGatewayConfig, route_path: Option<&str>, label: &str, report: &mut ConfigReport) {
177    let offline = crate::config_check::offline();
178    match route_path {
179        Some(path) if path.ends_with('*') => report.error(format!("{label}: an mcp route needs an exact [route.match] path, e.g. \"/mcp\" (not {path:?})")),
180        None => report.error(format!("{label}: an mcp route needs [route.match] path, e.g. \"/mcp\"")),
181        _ => {}
182    }
183    if config.upstreams.is_empty() && config.apis.is_empty() {
184        report.error(format!("{label}: [route.action.mcp] has no [[route.action.mcp.upstream]] or [[route.action.mcp.openapi]]"));
185    }
186    let mut prefixes = Vec::new();
187    for (i, u) in config.upstreams.iter().enumerate() {
188        let what = format!("{label}: upstream #{}", i + 1);
189        if u.prefix.is_empty() || u.prefix.contains("__") || u.prefix.ends_with('_') || !u.prefix.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
190            report.error(format!("{what}: prefix {:?} must be letters, digits, '-' and single '_'", u.prefix));
191        } else if prefixes.contains(&u.prefix) {
192            report.error(format!("{what}: prefix {:?} is used twice", u.prefix));
193        } else {
194            prefixes.push(u.prefix.clone());
195        }
196        if !(u.url.starts_with("http://") || u.url.starts_with("https://")) {
197            report.error(format!("{what}: url {:?} must start with http:// or https://", u.url));
198        }
199        for (header, env) in &u.headers_env {
200            if !offline && std::env::var(env).unwrap_or_default().is_empty() {
201                report.error(format!("{what}: header {header} reads {env}, which is unset or empty"));
202            }
203        }
204    }
205    for (i, a) in config.apis.iter().enumerate() {
206        let what = format!("{label}: openapi #{}", i + 1);
207        if a.prefix.is_empty() || a.prefix.contains("__") || a.prefix.ends_with('_') || !a.prefix.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
208            report.error(format!("{what}: prefix {:?} must be letters, digits, '-' and single '_'", a.prefix));
209        } else if prefixes.contains(&a.prefix) {
210            report.error(format!("{what}: prefix {:?} is used twice", a.prefix));
211        } else {
212            prefixes.push(a.prefix.clone());
213        }
214        if a.spec.is_empty() {
215            report.error(format!("{what}: spec (an OpenAPI document's URL or file) is required"));
216        } else if !offline && !a.spec.starts_with("http") && !std::path::Path::new(&a.spec).is_file() {
217            report.error(format!("{what}: spec file {:?} doesn't exist", a.spec));
218        }
219        if !(a.base_url.starts_with("http://") || a.base_url.starts_with("https://")) {
220            report.error(format!("{what}: base_url {:?} must start with http:// or https://", a.base_url));
221        }
222        for (header, env) in &a.headers_env {
223            if !offline && std::env::var(env).unwrap_or_default().is_empty() {
224                report.error(format!("{what}: header {header} reads {env}, which is unset or empty"));
225            }
226        }
227    }
228    if config.callers.is_empty() {
229        report.error(format!("{label}: no [[route.action.mcp.caller]]: nobody can call the gateway"));
230    }
231    for c in &config.callers {
232        if c.identity.is_empty() || c.token_env.is_empty() {
233            report.error(format!("{label}: a caller needs identity and token_env"));
234        } else if !offline && std::env::var(&c.token_env).unwrap_or_default().trim().is_empty() {
235            report.error(format!("{label}: caller {} reads its token from {}, which is unset or empty", c.identity, c.token_env));
236        }
237    }
238    if config.allow.is_empty() {
239        report.warn(format!("{label}: no [[route.action.mcp.allow]] rule: callers see no tools"));
240    }
241    for (identity, tools) in &config.allow {
242        if identity.is_empty() || tools.is_empty() {
243            report.error(format!("{label}: an allow rule needs identity and tools"));
244        }
245    }
246    for l in &config.limits {
247        if l.tools.is_empty() || l.max_calls == 0 || l.window_secs == 0 {
248            report.error(format!("{label}: a limit needs tools, max_calls and window_secs above 0"));
249        }
250    }
251}
252
253/// The gateway from `config` (lists every upstream's tools: network).
254fn build(config: &McpGatewayConfig, path: &str) -> Result<McpServer, String> {
255    let mut gateway = McpGateway::new(config.name.clone(), env!("CARGO_PKG_VERSION"))
256        .refresh_every((config.refresh_secs > 0).then(|| Duration::from_secs(config.refresh_secs)))
257        .audit_log(config.audit_log);
258    for u in &config.upstreams {
259        let mut upstream = Upstream::new(u.prefix.clone(), u.url.clone()).timeout_ms(u.timeout_ms);
260        for (header, env) in &u.headers_env {
261            if let Ok(value) = std::env::var(env) {
262                upstream = upstream.header(header.clone(), value);
263            }
264        }
265        gateway = gateway.upstream(upstream);
266    }
267    for a in &config.apis {
268        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);
269        for (header, env) in &a.headers_env {
270            if let Ok(value) = std::env::var(env) {
271                api = api.header(header.clone(), value);
272            }
273        }
274        gateway = gateway.openapi(api);
275    }
276    for (identity, tools) in &config.allow {
277        gateway = gateway.allow(identity.clone(), tools.clone());
278    }
279    for l in &config.limits {
280        gateway = gateway.limit(l.tools.clone(), l.max_calls, l.window_secs);
281    }
282    let tokens: Vec<(String, String)> = config
283        .callers
284        .iter()
285        .filter_map(|c| std::env::var(&c.token_env).ok().filter(|t| !t.trim().is_empty()).map(|t| (t, c.identity.clone())))
286        .collect();
287    let mut server = gateway.build()?.at(path).require_bearer_tokens(tokens);
288    if !config.allowed_origins.is_empty() {
289        server = server.allowed_origins(config.allowed_origins.clone());
290    }
291    Ok(server)
292}
293
294/// The route's [`Application`]: the gateway once built, 503 until then.
295#[doc(hidden)]
296pub struct GatewayAdapter {
297    slot: Arc<Mutex<Option<McpServer>>>,
298}
299
300impl GatewayAdapter {
301    /// Starts building in the background; retries every [`RETRY`] until it
302    /// works or the adapter is dropped (a reload replaced the route).
303    #[doc(hidden)]
304    pub fn start(config: McpGatewayConfig, path: String) -> GatewayAdapter {
305        let slot: Arc<Mutex<Option<McpServer>>> = Arc::new(Mutex::new(None));
306        let weak: Weak<Mutex<Option<McpServer>>> = Arc::downgrade(&slot);
307        let spawned = std::thread::Builder::new().name("rws-mcp-gateway-build".to_string()).spawn(move || loop {
308            match build(&config, &path) {
309                Ok(server) => {
310                    if let Some(slot) = weak.upgrade() {
311                        *slot.lock().unwrap_or_else(|e| e.into_inner()) = Some(server);
312                        println!("mcp gateway at {path}: {} upstream(s) federated", config.upstreams.len());
313                    }
314                    return;
315                }
316                Err(e) => {
317                    eprintln!("mcp gateway at {path}: {e}; retrying in {} s", RETRY.as_secs());
318                    std::thread::sleep(RETRY);
319                    if weak.strong_count() == 0 {
320                        return;
321                    }
322                }
323            }
324        });
325        if let Err(e) = spawned {
326            eprintln!("mcp gateway: not started: {e}");
327        }
328        GatewayAdapter { slot }
329    }
330}
331
332impl Application for GatewayAdapter {
333    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
334        let server = self.slot.lock().unwrap_or_else(|e| e.into_inner()).clone();
335        match server {
336            Some(server) => server.execute(request, connection),
337            None => Ok(unavailable()),
338        }
339    }
340}
341
342fn unavailable() -> Response {
343    let mut response = super::service_unavailable();
344    response.headers.retain(|h| !h.name.eq_ignore_ascii_case("retry-after"));
345    response.headers.push(crate::header::Header::new("Retry-After", RETRY.as_secs().to_string()));
346    response
347}
348
349#[cfg(test)]
350mod tests;
rws_proxy/proxy_config/mcp_gateway.rs

rws_proxy/proxy_config/
mcp_gateway.rs
