ferroui_mcp/
lib.rs

1//! Opt-in Model Context Protocol server for FerroUI. Call [`serve`] on the UI thread after
2//! application setup and keep the returned handle alive until shutdown. No server starts merely
3//! by linking the crate. Property writes and input are independently disabled by default.
4//!
5//! The tools run on the shared inspector of the UI thread (`ferroui_inspect::Inspector::shared`),
6//! the one the Chrome DevTools Protocol endpoint uses too: node IDs are the same for both, and a
7//! custom lifetime registers its top levels there (`register_root`).
8mod extensions;
9mod protocol;
10mod services;
11mod transport;
12pub use extensions::UiToolProvider;
13use ferroui_base::threading::{Dispatcher, DispatcherPriority};
14use ferroui_inspect::LogCapture;
15pub use ferroui_inspect::Permissions;
16use serde_json::Value;
17pub use services::{RequestContext, ServiceToolProvider};
18use std::{
19    io,
20    net::{IpAddr, Ipv4Addr, SocketAddr},
21    sync::{
22        atomic::{AtomicBool, Ordering},
23        mpsc, Arc,
24    },
25    time::{Duration, Instant},
26};
27
28/// HTTP listens only on a loopback address. Port zero chooses an unused port.
29#[derive(Clone, Debug)]
30pub enum Transport {
31    Stdio,
32    Http(SocketAddr),
33}
34pub struct Options {
35    pub transport: Transport,
36    pub permissions: Permissions,
37    /// Tee the application's diagnostic log into the bounded ring of `ferroui-inspect`, shared
38    /// with the other servers of the process, while the server is alive.
39    pub capture_logs: bool,
40    /// Previously reserved protocol output; otherwise stdio reserves stdout when serve is called.
41    pub stdio_output: Option<std::fs::File>,
42    /// Maximum time a queued UI operation may wait for the dispatcher.
43    pub dispatch_timeout: Duration,
44}
45impl Default for Options {
46    fn default() -> Self {
47        Self {
48            transport: Transport::Http(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)),
49            permissions: Permissions::default(),
50            capture_logs: true,
51            stdio_output: None,
52            dispatch_timeout: Duration::from_secs(10),
53        }
54    }
55}
56static STDIO_STARTED: AtomicBool = AtomicBool::new(false);
57
58/// Dropping this handle stops accepting requests and invalidates queued work. A stdio reader
59/// blocked in the OS remains parked until EOF; it cannot execute more UI operations.
60/// This inspector-only handle can be moved to another thread after server startup.
61pub struct ServerHandle {
62    client: Client,
63    log_guard: Option<ferroui_inspect::LogGuard>,
64    pub address: Option<SocketAddr>,
65    /// Bearer token for HTTP; printed to stderr at startup, never included in tool responses.
66    pub token: Option<String>,
67}
68impl ServerHandle {
69    pub fn is_running(&self) -> bool {
70        !self.client.stop.load(Ordering::Acquire)
71    }
72}
73impl Drop for ServerHandle {
74    fn drop(&mut self) {
75        self.client.stop.store(true, Ordering::Release);
76        self.log_guard.take();
77    }
78}
79
80/// A server that also owns application tools on the creating UI thread.
81///
82/// Keep and drop this handle on that thread, since providers may own controls and other
83/// thread-affine state. The underlying inspector handle remains accessible through `Deref`.
84pub struct UiServerHandle {
85    // Fields drop in declaration order: cancel queued requests before disposing the provider.
86    server: ServerHandle,
87    _registration: extensions::Registration,
88}
89impl std::ops::Deref for UiServerHandle {
90    type Target = ServerHandle;
91
92    fn deref(&self) -> &Self::Target {
93        &self.server
94    }
95}
96
97/// Start the server on the UI thread. The returned handle is the lifetime of the service.
98pub fn serve(options: Options) -> io::Result<ServerHandle> {
99    serve_registered(options, None, None, None)
100}
101
102/// Serve inspector tools and application tools through the same transport, permissions,
103/// cancellation and UI dispatcher. Keep the returned handle on this UI thread.
104pub fn serve_with_tools(
105    options: Options,
106    provider: std::rc::Rc<dyn UiToolProvider>,
107) -> io::Result<UiServerHandle> {
108    let registration = extensions::Registration::new(provider)?;
109    let server = serve_registered(options, Some(registration.catalog.clone()), None, None)?;
110    Ok(UiServerHandle {
111        server,
112        _registration: registration,
113    })
114}
115
116/// Host an isolated, thread-safe application service through the existing MCP protocol.
117///
118/// Only the provider's tools are exposed: no inspector, input, logs or UI objects are
119/// reachable through this server. Calls run on transport workers and receive cooperative
120/// cancellation and deadlines. This mode requires no application or UI message loop.
121pub fn serve_service(
122    options: Options,
123    provider: Arc<dyn ServiceToolProvider>,
124) -> io::Result<ServerHandle> {
125    let service = services::ServiceCatalog::new(provider)?;
126    serve_registered(options, None, Some(service), None)
127}
128
129/// Use a securely persisted 256-bit bearer for a durable service, so restarting the process
130/// does not invalidate peers' endpoints. Inspector servers continue generating fresh tokens.
131pub fn serve_service_with_token(
132    options: Options,
133    provider: Arc<dyn ServiceToolProvider>,
134    bearer: String,
135) -> io::Result<ServerHandle> {
136    if bearer.len() != 64 || !bearer.bytes().all(|byte| byte.is_ascii_hexdigit()) {
137        return Err(io::Error::new(
138            io::ErrorKind::InvalidInput,
139            "Service bearer must be 256-bit hexadecimal",
140        ));
141    }
142    let service = services::ServiceCatalog::new(provider)?;
143    serve_registered(options, None, Some(service), Some(bearer))
144}
145
146fn serve_registered(
147    options: Options,
148    extension: Option<extensions::Catalog>,
149    service: Option<services::ServiceCatalog>,
150    bearer: Option<String>,
151) -> io::Result<ServerHandle> {
152    let dispatcher = service.is_none().then(|| {
153        let dispatcher = Dispatcher::ui_thread();
154        dispatcher.verify_access();
155        dispatcher
156    });
157    if options.dispatch_timeout.is_zero() {
158        return Err(io::Error::new(
159            io::ErrorKind::InvalidInput,
160            "dispatch_timeout must be positive",
161        ));
162    }
163    let stdio = matches!(options.transport, Transport::Stdio);
164    if stdio && STDIO_STARTED.swap(true, Ordering::AcqRel) {
165        return Err(io::Error::new(
166            io::ErrorKind::AlreadyExists,
167            "Only one stdio server may start per process",
168        ));
169    }
170    // A server that fails to start leaves stdio free for another attempt.
171    let result = start(options, dispatcher, extension, service, bearer);
172    if stdio && result.is_err() {
173        STDIO_STARTED.store(false, Ordering::Release);
174    }
175    result
176}
177
178fn start(
179    options: Options,
180    dispatcher: Option<Arc<Dispatcher>>,
181    extension: Option<extensions::Catalog>,
182    service: Option<services::ServiceCatalog>,
183    bearer: Option<String>,
184) -> io::Result<ServerHandle> {
185    let (http, address, token) = match &options.transport {
186        Transport::Http(address) => {
187            if !address.ip().is_loopback() {
188                return Err(io::Error::new(
189                    io::ErrorKind::PermissionDenied,
190                    "MCP HTTP must bind to loopback",
191                ));
192            }
193            let server = std::net::TcpListener::bind(address)?;
194            let address = server.local_addr()?;
195            let mut bytes = [0u8; 32];
196            getrandom::fill(&mut bytes).map_err(|e| io::Error::other(e.to_string()))?;
197            let token = bearer
198                .unwrap_or_else(|| bytes.iter().map(|b| format!("{b:02x}")).collect::<String>());
199            (Some(server), Some(address), Some(token))
200        }
201        Transport::Stdio => (None, None, None),
202    };
203    let stdio_output = if http.is_none() {
204        Some(match options.stdio_output {
205            Some(file) => file,
206            None => reserve_stdout()?,
207        })
208    } else {
209        None
210    };
211    let log_guard = (options.capture_logs && service.is_none()).then(LogCapture::install);
212    let is_service = service.is_some();
213    let client = Client {
214        dispatcher,
215        extension,
216        service,
217        permissions: options.permissions,
218        timeout: options.dispatch_timeout,
219        stop: Arc::new(AtomicBool::new(false)),
220    };
221    let handle = ServerHandle {
222        client: client.clone(),
223        log_guard,
224        address,
225        token: token.clone(),
226    };
227    let task = move || {
228        let stop = client.stop.clone();
229        if let Some(server) = http {
230            transport::http(server, client, token.unwrap());
231        } else if let Err(e) = transport::stdio(client, stdio_output.unwrap()) {
232            eprintln!("FerroUI MCP stdio: {e}");
233        }
234        stop.store(true, Ordering::Release);
235    };
236    std::thread::Builder::new()
237        .name("FerroUI MCP".into())
238        .spawn(task)?;
239    if !is_service {
240        if let (Some(address), Some(token)) = (handle.address, &handle.token) {
241            eprintln!("FerroUI MCP: http://{address}/mcp\nAuthorization: Bearer {token}");
242        }
243    }
244    Ok(handle)
245}
246
247#[derive(Clone)]
248pub(crate) struct Client {
249    dispatcher: Option<Arc<Dispatcher>>,
250    extension: Option<extensions::Catalog>,
251    service: Option<services::ServiceCatalog>,
252    permissions: Permissions,
253    timeout: Duration,
254    stop: Arc<AtomicBool>,
255}
256impl Client {
257    fn tools(&self) -> Vec<Value> {
258        if let Some(service) = &self.service {
259            return service.tools().to_vec();
260        }
261        let mut tools = ferroui_inspect::tools();
262        if let Some(extension) = &self.extension {
263            tools.extend(extension.tools.iter().cloned());
264        }
265        tools
266    }
267    fn validate(&self, name: &str, args: &Value) -> Result<(), String> {
268        if let Some(service) = &self.service {
269            return service.validate(name, args);
270        }
271        if let Some(extension) = self
272            .extension
273            .as_ref()
274            .filter(|extension| extension.contains(name))
275        {
276            extension.validate_envelope(name, args)
277        } else {
278            ferroui_inspect::validate(name, args)
279        }
280    }
281    fn call(&self, name: &str, args: Value, cancel: &Arc<AtomicBool>) -> Result<Value, String> {
282        self.validate(name, &args)?;
283        if let Some(service) = &self.service {
284            let context = RequestContext::new(
285                self.permissions,
286                cancel.clone(),
287                self.stop.clone(),
288                self.timeout,
289            );
290            context.check_cancelled()?;
291            return std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
292                service.execute(name, &args, &context)
293            }))
294            .unwrap_or_else(|_| Err("Service operation panicked".into()));
295        }
296        // Only the built-in wait tool owns timeout_ms. In particular, application tools may
297        // use that argument independently, and adding an arbitrary duration to Instant can
298        // overflow on some platforms. Elapsed-time comparison also keeps wait_for portable.
299        let wait_started = Instant::now();
300        let wait_timeout = Duration::from_millis(args["timeout_ms"].as_u64().unwrap_or(5000));
301        loop {
302            if self.stop.load(Ordering::Acquire) || cancel.load(Ordering::Acquire) {
303                return Err("Request cancelled or server stopped".into());
304            }
305            let (tx, rx) = mpsc::sync_channel(1);
306            let permissions = self.permissions;
307            let extension = self
308                .extension
309                .clone()
310                .filter(|extension| extension.contains(name));
311            let tool = name.to_owned();
312            let args = args.clone();
313            let cancelled = cancel.clone();
314            let stop = self.stop.clone();
315            let pending = Arc::new(AtomicBool::new(true));
316            let queued = pending.clone();
317            self.dispatcher
318                .as_ref()
319                .expect("UI client owns a dispatcher")
320                .post(
321                    move || {
322                        if !queued.swap(false, Ordering::AcqRel)
323                            || cancelled.load(Ordering::Acquire)
324                            || stop.load(Ordering::Acquire)
325                        {
326                            return;
327                        }
328                        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
329                            if let Some(extension) = extension {
330                                extension.execute(&tool, &args, permissions)
331                            } else {
332                                ferroui_inspect::execute(&tool, &args, permissions)
333                            }
334                        }))
335                        .unwrap_or_else(|_| Err("UI operation panicked".into()));
336                        let _ = tx.send(result);
337                    },
338                    DispatcherPriority::BACKGROUND,
339                );
340            let queued_at = Instant::now();
341            let result = loop {
342                if self.stop.load(Ordering::Acquire) || cancel.load(Ordering::Acquire) {
343                    pending.store(false, Ordering::Release);
344                    return Err("Request cancelled or server stopped".into());
345                }
346                let remaining = self.timeout.saturating_sub(queued_at.elapsed());
347                if remaining.is_zero() {
348                    pending.store(false, Ordering::Release);
349                    return Err("UI dispatcher timed out".into());
350                }
351                match rx.recv_timeout(remaining.min(Duration::from_millis(25))) {
352                    Ok(result) => break result?,
353                    Err(mpsc::RecvTimeoutError::Timeout) => continue,
354                    Err(mpsc::RecvTimeoutError::Disconnected) => {
355                        return Err("UI operation cancelled".into())
356                    }
357                }
358            };
359            if name != "wait_for" || result["matched"] == true {
360                return Ok(result);
361            }
362            if wait_started.elapsed() >= wait_timeout {
363                return Err("Condition timed out".into());
364            }
365            std::thread::sleep(Duration::from_millis(25));
366        }
367    }
368}
369
370/// Reserve stdout for MCP before application startup. Ordinary stdout is redirected to stderr
371/// for the rest of the process, including native/library output on Unix (on Windows, output
372/// through the process's standard handle). Pass the returned file in Options.
373pub fn reserve_stdout() -> io::Result<std::fs::File> {
374    use std::io::Write;
375    std::io::stdout().flush()?;
376    let original = filedescriptor::FileDescriptor::redirect_stdio(
377        &std::io::stderr(),
378        filedescriptor::StdioDescriptor::Stdout,
379    )
380    .map_err(io::Error::other)?;
381    // The original stdout changes owner instead of being duplicated: dropping it would close the
382    // handle that the C runtime of Windows still holds as its descriptor 1.
383    #[cfg(unix)]
384    {
385        use std::os::fd::{FromRawFd, IntoRawFd};
386        // SAFETY: `into_raw_fd` hands over the descriptor, which nothing else owns.
387        Ok(unsafe { std::fs::File::from_raw_fd(original.into_raw_fd()) })
388    }
389    #[cfg(windows)]
390    {
391        use std::os::windows::io::{FromRawHandle, IntoRawHandle};
392        // SAFETY: `into_raw_handle` hands over the handle, which nothing else owns.
393        Ok(unsafe { std::fs::File::from_raw_handle(original.into_raw_handle()) })
394    }
395}
396
397#[cfg(test)]
398mod extensions_tests;
399#[cfg(test)]
400mod tests;
ferroui_mcp/lib.rs

ferroui_mcp/
lib.rs
