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

rust_web_server/proxy_config/
mcp_gateway.rs
