use std::sync::Arc;
use axum::{http::StatusCode, Router};
use regex::Regex;
use rmcp::{
handler::server::{router::tool::ToolRouter, wrapper::Parameters},
model::{Implementation, ServerCapabilities, ServerConfig},
schemars, tool, tool_handler, tool_router,
transport::streamable_http_server::{
session::local::LocalSessionManager, StreamableHttpServerConfig, StreamableHttpService,
},
ServerHandler,
};
use serde::Deserialize;
use tokio_util::sync::CancellationToken;
use crate::{config, db::Db, push, tmux};
pub const PATH: &str = "/mcp";
const LOOPBACK_HOSTS: [&str; 3] = ["localhost", "127.0.0.1", "::1"];
const LOOPBACK_ORIGINS: [&str; 6] = [
"http://localhost:*",
"http://127.0.0.1:*",
"http://[::1]:*",
"https://localhost:*",
"https://127.0.0.1:*",
"https://[::1]:*",
];
#[derive(Clone)]
pub struct Context {
pub session_name: Arc<Regex>,
pub db: Arc<Db>,
pub vapid_contact: String,
}
impl Context {
pub fn new(session_name: Arc<Regex>, db: Arc<Db>, config: &config::Config) -> Self {
Context {
session_name,
db,
vapid_contact: config.push.vapid_contact.clone(),
}
}
}
pub fn router(context: Context, shutdown: CancellationToken) -> Router {
let config = StreamableHttpServerConfig::default()
.with_cancellation_token(shutdown)
.with_allowed_hosts(LOOPBACK_HOSTS)
.with_allowed_origins(LOOPBACK_ORIGINS)
.enforce_origin_validation();
let service = StreamableHttpService::new(
move || Ok(Mobux::new(context.clone())),
Arc::new(LocalSessionManager::default()),
config,
);
Router::new().nest_service(PATH, service)
}
pub fn absent<S: Clone + Send + Sync + 'static>() -> Router<S> {
Router::new().route(
PATH,
axum::routing::any(|| async { (StatusCode::NOT_FOUND, "MCP is served on 127.0.0.1 only") }),
)
}
pub fn link_target(url: &str) -> Result<String, String> {
let url = url.trim();
if url.is_empty() {
return Err("url is required".to_string());
}
let lower = url.to_ascii_lowercase();
if lower.starts_with("http://") || lower.starts_with("https://") {
return Ok(url.to_string());
}
if has_scheme(url) {
return Err(format!("{url}: only http and https URLs open on the phone"));
}
if url.contains('\\') {
return Err(format!("{url}: a path cannot contain a backslash"));
}
Ok(url.trim_start_matches('/').to_string())
}
fn has_scheme(url: &str) -> bool {
let Some((scheme, _)) = url.split_once(':') else {
return false;
};
let mut chars = scheme.chars();
chars.next().is_some_and(|c| c.is_ascii_alphabetic())
&& chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.'))
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ReadScreenArgs {
pub session: String,
pub lines: Option<u32>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct RunCommandArgs {
pub session: String,
pub command: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct SendKeysArgs {
pub session: String,
pub text: String,
pub enter: Option<bool>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct NotifyArgs {
pub title: String,
pub body: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ShowOnPhoneArgs {
pub url: String,
pub title: String,
}
#[derive(Clone)]
pub struct Mobux {
context: Context,
tool_router: ToolRouter<Self>,
}
impl Mobux {
fn new(context: Context) -> Self {
Mobux {
context,
tool_router: Self::tool_router(),
}
}
fn session(&self, name: &str) -> Result<(), String> {
if !crate::is_valid_session_name(&self.context.session_name, name) {
return Err(format!("invalid session name: {name:?}"));
}
Ok(())
}
async fn push(&self, payload: push::Payload) -> Result<String, String> {
let delivery = push::send_to_devices(
self.context.db.clone(),
self.context.vapid_contact.clone(),
payload,
)
.await
.map_err(one_line)?;
Ok(format!(
"sent={} failed={} pruned={}",
delivery.sent, delivery.failed, delivery.pruned
))
}
}
fn one_line(error: anyhow::Error) -> String {
format!("{error:#}").replace('\n', " ")
}
#[tool_router]
impl Mobux {
#[tool(
description = "List the tmux sessions: name, window count, the active window and whether it is on the alternate screen (a full-screen app such as vim or less)."
)]
async fn list_sessions(&self) -> Result<String, String> {
let sessions = tmux::list_sessions(None).await.map_err(one_line)?;
if sessions.is_empty() {
return Ok("no tmux sessions".to_string());
}
let mut lines = Vec::with_capacity(sessions.len());
for session in sessions {
let windows = tmux::list_panes(&session.name, None)
.await
.map_err(one_line)?;
let line = match windows.iter().find(|window| window.active) {
Some(active) => format!(
"{}\twindows={}\tactive={}:{}\talternate_screen={}",
session.name,
session.windows,
active.index,
active.title,
if active.alternate_on { "yes" } else { "no" }
),
None => format!("{}\twindows={}", session.name, session.windows),
};
lines.push(line);
}
Ok(lines.join("\n"))
}
#[tool(
description = "Read the active pane of a session as plain text: the visible screen, plus up to `lines` scrollback lines above it."
)]
async fn read_screen(
&self,
Parameters(args): Parameters<ReadScreenArgs>,
) -> Result<String, String> {
self.session(&args.session)?;
let lines = args.lines.unwrap_or(0).min(crate::HISTORY_MAX_LINES);
let capture = tmux::capture_history(&args.session, lines, tmux::HistoryScope::All, None)
.await
.map_err(one_line)?;
Ok(crate::strip_ansi(&capture.text).trim_end().to_string())
}
#[tool(
description = "Run a tmux command on a session: new-window, kill-window, split-h, split-v, next-window, prev-window, next-pane, prev-pane, kill-pane or zoom-pane."
)]
async fn run_tmux_command(
&self,
Parameters(args): Parameters<RunCommandArgs>,
) -> Result<String, String> {
self.session(&args.session)?;
tmux::list_panes(&args.session, None)
.await
.map_err(one_line)?;
let output = tmux::run_command(&args.session, &args.command, None)
.await
.map_err(one_line)?;
let output = output.trim();
if output.is_empty() {
return Ok(format!("{} on {}: done", args.command, args.session));
}
Ok(format!("{} on {}: {output}", args.command, args.session))
}
#[tool(
description = "Type text into the active pane of a session as literal keystrokes, optionally followed by Enter. A newline in the text runs the line, even with enter off."
)]
async fn send_keys(
&self,
Parameters(args): Parameters<SendKeysArgs>,
) -> Result<String, String> {
self.session(&args.session)?;
let enter = args.enter.unwrap_or(false);
tmux::send_text(&args.session, &args.text, enter, None)
.await
.map_err(one_line)?;
let typed = args.text.chars().count();
Ok(match enter {
true => format!("typed {typed} characters and Enter into {}", args.session),
false => format!("typed {typed} characters into {}", args.session),
})
}
#[tool(description = "Send a push notification to every phone subscribed to this mobux.")]
async fn notify(&self, Parameters(args): Parameters<NotifyArgs>) -> Result<String, String> {
if args.body.trim().is_empty() {
return Err("body is required".to_string());
}
let outcome = self
.push(push::Payload {
title: args.title,
body: args.body,
tag: None,
url: None,
})
.await?;
Ok(format!("notification delivered: {outcome}"))
}
#[tool(
description = "Push a notification that opens a page on the phone when tapped: an http(s) URL, or a path on this mobux such as /files/site/ or /proxy/vite/."
)]
async fn show_on_phone(
&self,
Parameters(args): Parameters<ShowOnPhoneArgs>,
) -> Result<String, String> {
let target = link_target(&args.url)?;
let outcome = self
.push(push::Payload {
title: args.title,
body: target.clone(),
tag: None,
url: Some(target.clone()),
})
.await?;
Ok(format!("{target} delivered: {outcome}"))
}
}
#[tool_handler(router = self.tool_router)]
impl ServerHandler for Mobux {
fn get_info(&self) -> ServerConfig {
ServerConfig::new(ServerCapabilities::builder().enable_tools().build())
.with_server_info(Implementation::new("mobux", env!("CARGO_PKG_VERSION")))
.with_instructions(
"mobux runs the tmux sessions on this host and shows them on the user's phone. \
Read and drive those sessions, and push a notification or a page to the phone.",
)
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::Body;
use axum::http::Request;
use tower::ServiceExt;
const INITIALIZE: &str = r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1"}}}"#;
fn test_router() -> (Router, tempfile::TempDir) {
let dir = tempfile::tempdir().unwrap();
let db = Arc::new(Db::open(&dir.path().join("mobux.db")).unwrap());
let context = Context::new(
Arc::new(Regex::new(r"^[a-zA-Z0-9_-]+$").unwrap()),
db,
&config::Config::default(),
);
(router(context, CancellationToken::new()), dir)
}
async fn post(host: &str, origin: Option<&str>) -> (StatusCode, String) {
let (router, _dir) = test_router();
let mut request = Request::post(PATH)
.header("host", host)
.header("content-type", "application/json")
.header("accept", "application/json, text/event-stream");
if let Some(origin) = origin {
request = request.header("origin", origin);
}
let response = router
.oneshot(request.body(Body::from(INITIALIZE)).unwrap())
.await
.unwrap();
let status = response.status();
let body = axum::body::to_bytes(response.into_body(), 64 * 1024)
.await
.unwrap();
(status, String::from_utf8_lossy(&body).into_owned())
}
#[tokio::test]
async fn a_loopback_host_without_an_origin_is_served() {
for host in ["127.0.0.1:8415", "localhost:8415", "[::1]:8415"] {
let (status, _) = post(host, None).await;
assert_eq!(status, StatusCode::OK, "{host}");
}
}
#[tokio::test]
async fn a_loopback_origin_is_served() {
let (status, _) = post("127.0.0.1:8415", Some("http://localhost:3000")).await;
assert_eq!(status, StatusCode::OK);
}
#[tokio::test]
async fn a_non_loopback_host_is_refused_with_one_line() {
for host in ["evil.example", "evil.example:8415", "192.168.1.5:8415"] {
let (status, body) = post(host, None).await;
assert_eq!(status, StatusCode::FORBIDDEN, "{host}");
assert_eq!(body, "Forbidden: Host header is not allowed");
}
}
#[tokio::test]
async fn a_non_loopback_origin_is_refused_with_one_line() {
for origin in [
"http://evil.example",
"https://127.0.0.1.evil.example",
"null",
] {
let (status, body) = post("127.0.0.1:8415", Some(origin)).await;
assert_eq!(status, StatusCode::FORBIDDEN, "{origin}");
assert_eq!(body, "Forbidden: Origin header is not allowed");
}
}
#[tokio::test]
async fn the_other_listeners_answer_404_at_the_mcp_path() {
let response = absent::<()>()
.oneshot(
Request::post(PATH)
.header("host", "127.0.0.1")
.body(Body::from(INITIALIZE))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[test]
fn a_path_is_sent_relative_to_the_instance_root() {
assert_eq!(link_target("/files/site/").unwrap(), "files/site/");
assert_eq!(link_target("proxy/vite/").unwrap(), "proxy/vite/");
assert_eq!(link_target("//evil.com/x").unwrap(), "evil.com/x");
}
#[test]
fn a_backslash_in_a_path_is_refused() {
assert!(link_target(r"\\evil.com/x").is_err());
assert!(link_target(r"\/evil.com").is_err());
assert!(link_target(r"/files/a\b").is_err());
}
#[test]
fn an_http_url_is_kept_and_other_schemes_are_refused() {
assert_eq!(
link_target("https://example.com/a").unwrap(),
"https://example.com/a"
);
assert!(link_target("javascript:alert(1)").is_err());
assert!(link_target(" ").is_err());
}
}