//! MCP tool registrations: 19 tools on the shared library services.
//! `query`/`execute` go through the gate with elicitation approvals; every
//! other tool maps to a library service with the legacy JSON shape.
use rmcp::handler::server::tool::ToolCallContext;
use rmcp::handler::server::wrapper::Parameters;
use rmcp::model::{CallToolResponse, CallToolResult};
use rmcp::schemars;
use rmcp::tool;
use rmcp::tool_router;
use rmcp::{ServerHandler, tool_handler};
use serde::Deserialize;
use serde_json::json;
use crate::app::gate::{self, GateDeps, RunSqlArgs};
use crate::config::{ConfigStore, Connection, SqliteConnection};
use crate::policy::model::{PartialPolicy, PolicyPresetName, SqlCategory, TableRuleKey};
use super::{SequelServer, error_tool_result, json_tool_result, text_tool_result};
impl SequelServer {
/// Process-shared approval engine: "Allow for session" grants must
/// survive across tool calls (a fresh engine per call silently
/// voided the documented session scope).
fn gate_deps_blocking(&self, sink: Box<dyn gate::ApprovalSink>) -> GateDeps {
GateDeps::with_sink_and_approvals(sink, self.ctx.approvals.clone())
}
}
fn resolve_conn(store: &ConfigStore, name: Option<&str>) -> Result<Option<Connection>, String> {
store
.load()
.map(|cfg| cfg.resolve(name).cloned())
.map_err(|e| e.to_string())
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct EmptyParams {}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct SqlParams {
/// Configured connection name. Omit to use the default connection.
#[serde(default)]
pub connection: Option<String>,
/// Single SQL statement (multi-statement input rejected).
pub sql: String,
/// Override default database/schema.
#[serde(default)]
pub database: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ConnectionRef {
#[serde(default)]
pub connection: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct AddSqliteParams {
pub name: String,
/// SQLite database file path. `~/` is expanded at execution time.
pub path: String,
/// SQLite schema name used for policy scope and metadata lookups.
#[serde(default)]
pub database: Option<String>,
#[serde(default)]
pub policy_preset: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct AddConnectionParams {
pub name: String,
pub host: String,
#[serde(default)]
pub port: Option<u16>,
pub user: String,
#[serde(default)]
pub database: Option<String>,
#[serde(default)]
pub ssl: Option<bool>,
#[serde(default)]
pub policy_preset: Option<String>,
#[serde(default)]
pub ssh_host: Option<String>,
#[serde(default)]
pub ssh_port: Option<u16>,
#[serde(default)]
pub ssh_user: Option<String>,
#[serde(default)]
pub ssh_key_path: Option<String>,
/// Docker container name (validated `^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$`).
#[serde(default)]
pub ssh_docker_container: Option<String>,
/// `nc` (default), `ncat`, or `socat`.
#[serde(default)]
pub ssh_docker_bridge_tool: Option<String>,
/// `lenient` (default) or `strict`.
#[serde(default)]
pub ssh_host_key_policy: Option<String>,
#[serde(default)]
pub ssh_known_hosts_path: Option<String>,
#[serde(default)]
pub ssl_server_name: Option<String>,
/// PEM/DER file of a private CA the server certificate is verified
/// against (merged with the system roots).
#[serde(default)]
pub ssl_ca_path: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct RemoveParams {
pub name: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct SetDefaultParams {
/// Connection name, or empty string to clear the default.
pub name: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct SelectDatabaseParams {
#[serde(default)]
pub connection: Option<String>,
pub database: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct DescribeParams {
#[serde(default)]
pub connection: Option<String>,
#[serde(default)]
pub database: Option<String>,
pub table: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct PolicyParams {
pub name: String,
pub policy: PartialPolicy,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct DatabasePolicyParams {
#[serde(default)]
pub connection: Option<String>,
pub database: String,
pub policy: PartialPolicy,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct DatabaseOnlyParams {
#[serde(default)]
pub connection: Option<String>,
pub database: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct TablePolicyParams {
#[serde(default)]
pub connection: Option<String>,
/// Table key: `database.table` for exact rules or `database.*` wildcard.
pub table: String,
pub policy: PartialPolicy,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct TableOnlyParams {
#[serde(default)]
pub connection: Option<String>,
pub table: String,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ExplainParams {
#[serde(default)]
pub connection: Option<String>,
pub sql: String,
#[serde(default)]
pub database: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct AuditSearchParams {
#[serde(default)]
pub connection: Option<String>,
#[serde(default)]
pub category: Option<String>,
#[serde(default)]
pub outcome: Option<String>,
#[serde(default)]
pub since_iso: Option<String>,
#[serde(default)]
pub until_iso: Option<String>,
#[serde(default)]
pub limit: Option<u32>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct BackupListParams {
#[serde(default)]
pub connection: Option<String>,
#[serde(default)]
pub limit: Option<u32>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct RestoreParams {
/// Backup id from list_backups.
pub backup_id: i64,
/// Default true: inspect the plan without executing anything.
#[serde(default = "default_true")]
pub dry_run: bool,
}
fn default_true() -> bool {
true
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct AuditCleanupParams {
#[serde(default = "default_true")]
pub dry_run: bool,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct RetentionParams {
#[serde(default)]
pub retention_days_by_category: Option<RetentionPartial>,
#[serde(default)]
pub backup_days: Option<u32>,
#[serde(default)]
pub audit_max_mb: Option<u32>,
#[serde(default)]
pub backup_max_mb: Option<u32>,
#[serde(default)]
pub auto_cleanup_hours: Option<u32>,
#[serde(default)]
pub redact_sql_in_log: Option<bool>,
#[serde(default)]
pub tamper_evident_chain: Option<bool>,
}
#[derive(Debug, Default, Deserialize, schemars::JsonSchema)]
pub struct RetentionPartial {
#[serde(default)]
pub read: Option<u32>,
#[serde(default)]
pub write: Option<u32>,
#[serde(default)]
pub ddl: Option<u32>,
#[serde(default)]
pub admin: Option<u32>,
#[serde(default)]
pub tx_ctrl: Option<u32>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct HistorySearchParams {
#[serde(default)]
pub since_iso: Option<String>,
#[serde(default)]
pub until_iso: Option<String>,
#[serde(default)]
pub search: Option<String>,
#[serde(default)]
pub connection: Option<String>,
#[serde(default)]
pub source: Option<String>,
#[serde(default)]
pub limit: Option<u32>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct SequelAceHistoryParams {
#[serde(default)]
pub since_iso: Option<String>,
#[serde(default)]
pub search: Option<String>,
#[serde(default)]
pub limit: Option<u32>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ImportParams {
/// Copy passwords from the Sequel Ace Keychain (macOS may prompt).
#[serde(default = "default_true")]
pub copy_passwords: bool,
}
fn parse_preset(s: &Option<String>) -> Result<PolicyPresetName, String> {
match s.as_deref() {
None => Ok(PolicyPresetName::ReadOnly),
Some(name) => {
PolicyPresetName::parse(name).ok_or_else(|| format!("unknown policy preset {name:?}"))
}
}
}
#[tool_router]
impl SequelServer {
#[tool(
name = "query",
title = "Run a read-only SQL query",
description = "Run a single read-only SQL statement (SELECT/SHOW/DESCRIBE/EXPLAIN, plus read-only SQLite PRAGMA). SQLite opens a read-only file handle.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn query(&self, Parameters(p): Parameters<SqlParams>) -> CallToolResult {
// Hand-routed in call_tool for elicitation; this entry exists for
// discovery and direct dispatch without interactive approvals.
self.run_sql_blocking(
RunSqlArgs {
connection: p.connection,
sql: p.sql,
database: p.database,
expected_ddl_targets: None,
},
true,
)
}
#[tool(
name = "execute",
title = "Execute a write/DDL/admin SQL statement",
description = "Run a non-read SQL statement (INSERT/UPDATE/DELETE/DDL/admin). Subject to the two-layer policy: table-rule elevation still requires user confirmation.",
annotations(destructive_hint = true, open_world_hint = false)
)]
fn execute(&self, Parameters(p): Parameters<SqlParams>) -> CallToolResult {
self.run_sql_blocking(
RunSqlArgs {
connection: p.connection,
sql: p.sql,
database: p.database,
expected_ddl_targets: None,
},
false,
)
}
#[tool(
name = "list_connections",
title = "List configured connections",
description = "Return all connections configured in the local config (no passwords).",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn list_connections(&self, _p: Parameters<EmptyParams>) -> CallToolResult {
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("config load failed: {e}")),
};
let mut items = Vec::new();
for c in &cfg.connections {
let has_password = c.is_mysql()
&& self
.ctx
.secrets
.has_password(c.name(), mysql_user(c).as_deref().unwrap_or(""));
items.push(json!({
"name": c.name(),
"driver": if c.is_mysql() { "mysql" } else { "sqlite" },
"database": c.database(),
"policy": c.policy(),
"tablePolicies": c.table_policies(),
"isDefault": Some(c.name()) == cfg.default_connection.as_deref(),
"hasStoredPassword": has_password,
}));
}
json_tool_result(json!({
"defaultConnection": cfg.default_connection,
"connections": items,
}))
}
#[tool(
name = "add_connection",
title = "Add or update a MySQL/MariaDB connection",
description = "Persist a MySQL/MariaDB connection. The password is captured via elicitation and stored in the macOS Keychain; it never appears in tool arguments or logs.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn add_connection(&self, _p: Parameters<AddConnectionParams>) -> CallToolResult {
// Hand-routed in call_tool: password capture needs the session
// peer for the elicitation round trip.
error_tool_result(
"add_connection is hand-routed for password elicitation; connect through the full tool call path",
)
}
#[tool(
name = "add_sqlite_connection",
title = "Add or update a SQLite connection",
description = "Persist a SQLite database file connection. Stores only the local file path and policy; no password or Keychain entry is used.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn add_sqlite_connection(&self, Parameters(p): Parameters<AddSqliteParams>) -> CallToolResult {
let preset = match parse_preset(&p.policy_preset) {
Ok(v) => v,
Err(e) => return error_tool_result(e),
};
let mut sc = SqliteConnection {
name: p.name.clone(),
path: p.path.clone(),
database: p.database.clone().unwrap_or_else(|| "main".into()),
..SqliteConnection::default()
};
sc.policy = crate::policy::model::policy_from_preset(preset);
let conn = Connection::Sqlite(sc);
if let Err(e) = conn.validate() {
return error_tool_result(format!("{e}"));
}
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
let name = p.name.clone();
let preset_name = preset.as_str().to_string();
match self.ctx.config.update(cfg.revision, move |c| {
upsert(c, conn);
Ok(())
}) {
Ok(()) => text_tool_result(format!(
"Saved SQLite connection \"{name}\" with policy preset \"{preset_name}\". No password was stored."
)),
Err(e) => error_tool_result(format!("{e}")),
}
}
#[tool(
name = "remove_connection",
title = "Remove a connection",
description = "Delete the connection from config and delete any associated MySQL/MariaDB Keychain password.",
annotations(
destructive_hint = true,
idempotent_hint = true,
open_world_hint = false
)
)]
fn remove_connection(&self, Parameters(p): Parameters<RemoveParams>) -> CallToolResult {
let conn = match resolve_conn(&self.ctx.config, Some(&p.name)) {
Ok(Some(c)) => c,
Ok(None) => return error_tool_result(format!("Connection {:?} not found", p.name)),
Err(e) => return error_tool_result(e),
};
if let Connection::Mysql(m) = &conn {
let _ = self.ctx.secrets.delete_password(&m.name, &m.user);
if let Some(ssh) = &m.ssh {
let _ = self
.ctx
.secrets
.delete_password(&format!("{}::ssh", m.name), &ssh.user);
}
}
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
let name = p.name.clone();
match self.ctx.config.update(cfg.revision, move |c| {
c.connections.retain(|x| x.name() != name);
if c.default_connection.as_deref() == Some(name.as_str()) {
c.default_connection = None;
}
Ok(())
}) {
Ok(()) => text_tool_result(format!("Removed connection \"{}\".", p.name)),
Err(e) => error_tool_result(format!("{e}")),
}
}
#[tool(
name = "set_default_connection",
title = "Set the default connection",
description = "Mark a saved connection as the default. Subsequent query/execute/describe_table/list_databases calls without an explicit \"connection\" arg use it. Pass an empty string to clear.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn set_default_connection(
&self,
Parameters(p): Parameters<SetDefaultParams>,
) -> CallToolResult {
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
if p.name.is_empty() {
let _ = self.ctx.config.update(cfg.revision, |c| {
c.default_connection = None;
Ok(())
});
return text_tool_result("Default connection cleared.");
}
let exists = cfg.connections.iter().any(|c| c.name() == p.name);
if !exists {
return error_tool_result(format!("Connection \"{}\" not found", p.name));
}
let name = p.name.clone();
match self.ctx.config.update(cfg.revision, move |c| {
c.default_connection = Some(name.clone());
Ok(())
}) {
Ok(()) => text_tool_result(format!("Default connection is now \"{}\".", p.name)),
Err(e) => error_tool_result(format!("{e}")),
}
}
#[tool(
name = "get_default_connection",
title = "Get the default connection",
description = "Return the connection name currently used when \"connection\" arg is omitted.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn get_default_connection(&self, _p: Parameters<EmptyParams>) -> CallToolResult {
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
json_tool_result(json!({ "defaultConnection": cfg.default_connection }))
}
#[tool(
name = "select_database",
title = "Set the default database on a connection",
description = "Update a saved connection so that subsequent query/execute calls default to this database when no per-call override is supplied.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn select_database(&self, Parameters(p): Parameters<SelectDatabaseParams>) -> CallToolResult {
if !p
.database
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '$')
{
return error_tool_result("database must be identifier-safe");
}
let mut conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
match &mut conn {
Connection::Mysql(m) => m.database = Some(p.database.clone()),
Connection::Sqlite(s) => s.database = p.database.clone(),
}
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
let name = conn.name().to_string();
let db = p.database.clone();
let name_for_msg = name.clone();
match self.ctx.config.update(cfg.revision, move |c| {
upsert(c, conn);
let _ = (&name, &db);
Ok(())
}) {
Ok(()) => text_tool_result(format!(
"Default database for \"{}\" set to \"{}\". Per-call database overrides still take precedence.",
name_for_msg, p.database
)),
Err(e) => error_tool_result(format!("{e}")),
}
}
#[tool(
name = "set_policy",
title = "Update a connection policy",
description = "Change the action set (read|write|ddl|admin|txCtrl → allow|confirm|deny) and limits for an existing connection baseline.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn set_policy(&self, Parameters(p): Parameters<PolicyParams>) -> CallToolResult {
let mut conn = match resolve_conn(&self.ctx.config, Some(&p.name)) {
Ok(Some(c)) => c,
Ok(None) => return error_tool_result(format!("Connection {:?} not found", p.name)),
Err(e) => return error_tool_result(e),
};
match &mut conn {
Connection::Mysql(m) => {
m.policy = m.policy.merged_with(&p.policy);
}
Connection::Sqlite(s) => {
s.policy = s.policy.merged_with(&p.policy);
}
}
if let Err(e) = conn.validate() {
return error_tool_result(format!("{e}"));
}
let policy = match &conn {
Connection::Mysql(m) => m.policy.clone(),
Connection::Sqlite(s) => s.policy.clone(),
};
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
match self.ctx.config.update(cfg.revision, move |c| {
upsert(c, conn);
Ok(())
}) {
Ok(()) => json_tool_result(json!({ "name": p.name, "policy": policy })),
Err(e) => error_tool_result(format!("{e}")),
}
}
#[tool(
name = "set_table_policy",
title = "Set a table policy rule",
description = "Set an exact (`database.table`) or wildcard (`database.*`) table rule. Exact rules take precedence over wildcards. A rule that elevates what the baseline denies still requires confirmation per statement.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn set_table_policy(&self, Parameters(p): Parameters<TablePolicyParams>) -> CallToolResult {
let key = match TableRuleKey::parse(&p.table) {
Some(k) => k,
None => {
return error_tool_result(format!(
"invalid table key {:?} (expected database.table or database.*)",
p.table
));
}
};
// Table rules previously bypassed all bounds validation — a rule
// could widen rowCap/stmtTimeoutMs beyond what any baseline
// policy allows (review finding).
if let Err(e) = p.policy.validate() {
return error_tool_result(format!("table policy invalid: {e}"));
}
let mut conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
let rendered = key.render();
let partial = p.policy.clone();
match &mut conn {
Connection::Mysql(m) => {
m.table_policies.insert(key, partial);
}
Connection::Sqlite(s) => {
s.table_policies.insert(key, partial);
}
}
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
let name = conn.name().to_string();
match self.ctx.config.update(cfg.revision, move |c| {
upsert(c, conn);
Ok(())
}) {
Ok(()) => json_tool_result(json!({
"connection": name,
"table": rendered,
"policy": p.policy,
})),
Err(e) => error_tool_result(format!("{e}")),
}
}
#[tool(
name = "clear_table_policy",
title = "Clear a table policy rule",
description = "Remove an exact or wildcard table rule; the connection baseline applies again.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn clear_table_policy(&self, Parameters(p): Parameters<TableOnlyParams>) -> CallToolResult {
let key = match TableRuleKey::parse(&p.table) {
Some(k) => k,
None => return error_tool_result(format!("invalid table key {:?}", p.table)),
};
let mut conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
let existed = match &mut conn {
Connection::Mysql(m) => m.table_policies.remove(&key).is_some(),
Connection::Sqlite(s) => s.table_policies.remove(&key).is_some(),
};
if !existed {
return text_tool_result(format!("No rule exists for {}.", p.table));
}
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("{e}")),
};
match self.ctx.config.update(cfg.revision, move |c| {
upsert(c, conn);
Ok(())
}) {
Ok(()) => text_tool_result(format!("Cleared table rule {}.", p.table)),
Err(e) => error_tool_result(format!("{e}")),
}
}
#[tool(
name = "list_table_policies",
title = "List table policy rules",
description = "Show the connection baseline plus every exact and wildcard table rule.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn list_table_policies(&self, Parameters(p): Parameters<ConnectionRef>) -> CallToolResult {
let conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
json_tool_result(json!({
"connection": conn.name(),
"baseline": conn.policy(),
"tablePolicies": conn.table_policies(),
}))
}
#[tool(
name = "explain_policy",
title = "Explain effective policy for a statement",
description = "Classify a statement and show the per-table resolution (baseline vs rule, strictest-wins, elevation flags) without executing it.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn explain_policy(&self, Parameters(p): Parameters<ExplainParams>) -> CallToolResult {
let conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
let dialect = if conn.is_mysql() {
crate::policy::classifier::Dialect::MySql
} else {
crate::policy::classifier::Dialect::SQLite
};
let classified = match crate::policy::classifier::classify_statement(&p.sql, dialect) {
Ok(c) => c,
Err(e) => {
return error_tool_result(format!("Cannot classify statement: {}", e.message()));
}
};
let fallback = p
.database
.clone()
.or_else(|| conn.database().map(str::to_string));
let r = crate::policy::resolver::resolve(&conn, &classified, fallback.as_deref());
let contributions: Vec<serde_json::Value> = r
.contributions
.iter()
.map(|c| {
json!({
"table": format!("{}.{}", c.table.database, c.table.table),
"kind": c.kind,
"category": c.category.as_str(),
"action": c.action.as_str(),
"baselineAction": c.baseline_action.as_str(),
"rule": c.rule,
"elevated": c.elevated,
})
})
.collect();
json_tool_result(json!({
"category": classified.category.as_str(),
"astType": classified.ast_type,
"action": r.action.as_str(),
"elevated": r.elevated,
"denyReason": r.deny_reason.as_ref().map(|d| format!("{d:?}")),
"contributions": contributions,
"contributingDatabases": r.contributing_databases,
"flags": {
"lockingRead": classified.locking_read,
"fileIo": classified.file_io,
"executesWrapped": classified.executes_wrapped,
},
}))
}
#[tool(
name = "set_database_policy",
title = "Set per-database policy override (compatibility)",
description = "Compatibility wrapper: operates on the wildcard table rule `<database>.*`. Prefer set_table_policy.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn set_database_policy(
&self,
Parameters(p): Parameters<DatabasePolicyParams>,
) -> CallToolResult {
self.set_table_policy(Parameters(TablePolicyParams {
connection: p.connection.clone(),
table: format!("{}.*", p.database),
policy: p.policy.clone(),
}))
}
#[tool(
name = "clear_database_policy",
title = "Clear per-database policy override (compatibility)",
description = "Compatibility wrapper: clears the wildcard table rule `<database>.*`.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn clear_database_policy(
&self,
Parameters(p): Parameters<DatabaseOnlyParams>,
) -> CallToolResult {
self.clear_table_policy(Parameters(TableOnlyParams {
connection: p.connection.clone(),
table: format!("{}.*", p.database),
}))
}
#[tool(
name = "list_database_policies",
title = "List per-database policy overrides (compatibility)",
description = "Compatibility wrapper: shows baseline + wildcard (`db.*`) rules rendered as per-database overrides.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn list_database_policies(&self, Parameters(p): Parameters<ConnectionRef>) -> CallToolResult {
let conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
let mut overrides = serde_json::Map::new();
for (k, v) in conn.table_policies() {
if let TableRuleKey::Wildcard { database } = k {
overrides.insert(
database.clone(),
serde_json::to_value(v).unwrap_or_default(),
);
}
}
json_tool_result(json!({
"connection": conn.name(),
"baseline": conn.policy(),
"overrides": overrides,
}))
}
#[tool(
name = "describe_table",
title = "Describe a table",
description = "Describe a table. Uses DESCRIBE on MySQL/MariaDB and PRAGMA table_info on SQLite. Always read-only.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn describe_table(&self, Parameters(p): Parameters<DescribeParams>) -> CallToolResult {
let conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
let q = format!("`{}`", p.table.replace('`', "``"));
let sql = if conn.is_mysql() {
match &p.database {
Some(db) => format!(
"DESCRIBE `{}`.`{}`",
db.replace('`', "``"),
p.table.replace('`', "``")
),
None => format!("DESCRIBE {q}"),
}
} else {
let schema = p
.database
.as_deref()
.or(conn.database().map(|_| ""))
.unwrap_or("main");
let _ = schema;
format!(
"PRAGMA `{}`.table_info('{}')",
p.database
.clone()
.unwrap_or_else(|| "main".into())
.replace('`', "``"),
p.table.replace('\'', "''")
)
};
self.run_sql_blocking(
RunSqlArgs {
connection: p.connection.clone(),
sql,
database: p.database.clone(),
expected_ddl_targets: None,
},
true,
)
}
#[tool(
name = "list_databases",
title = "List databases",
description = "SHOW DATABASES on MySQL/MariaDB or PRAGMA database_list on SQLite.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn list_databases(&self, Parameters(p): Parameters<ConnectionRef>) -> CallToolResult {
let conn = match resolve_conn(&self.ctx.config, p.connection.as_deref()) {
Ok(Some(c)) => c,
Ok(None) => {
return error_tool_result(gate::no_connection_message(p.connection.as_deref()));
}
Err(e) => return error_tool_result(e),
};
let sql = if conn.is_mysql() {
"SHOW DATABASES".to_string()
} else {
"PRAGMA database_list".to_string()
};
self.run_sql_blocking(
RunSqlArgs {
connection: p.connection.clone(),
sql,
database: None,
expected_ddl_targets: None,
},
true,
)
}
#[tool(
name = "audit_search",
title = "Search audit log",
description = "Query the local audit-log SQLite. Returns redacted SQL by default. Includes connection, decision, outcome, duration, and backup_id.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn audit_search(&self, Parameters(p): Parameters<AuditSearchParams>) -> CallToolResult {
let filters = crate::audit::AuditSearchFilters {
since: p.since_iso.clone(),
until: p.until_iso.clone(),
connection: p.connection.clone(),
category: p.category.as_deref().and_then(SqlCategory::parse),
outcome: p.outcome.clone(),
limit: p.limit.unwrap_or(200),
};
match crate::audit::search_audit_log(&self.ctx.audit, &filters) {
Ok(rows) => json_tool_result(json!({ "count": rows.len(), "rows": rows })),
Err(e) => error_tool_result(format!("audit search failed: {e}")),
}
}
#[tool(
name = "list_backups",
title = "List recent row/schema backups",
description = "Show recent pre-mutation backups taken before UPDATE/DELETE/TRUNCATE/DROP/ALTER. Each row links to a backup_id.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn list_backups(&self, Parameters(p): Parameters<BackupListParams>) -> CallToolResult {
match crate::backup::list_backups(
&self.ctx.audit,
p.connection.as_deref(),
p.limit.unwrap_or(50),
) {
Ok(rows) => json_tool_result(json!({ "count": rows.len(), "rows": rows })),
Err(e) => error_tool_result(format!("backup list failed: {e}")),
}
}
#[tool(
name = "restore_backup",
title = "Restore from a pre-mutation backup",
description = "Replay backup #N into the originating connection. Generates dialect-specific upserts for row backups and CREATE TABLE for schema backups. Subject to the same policy gate (counts as a write). Pass dryRun=true to inspect the plan first.",
annotations(
destructive_hint = true,
idempotent_hint = false,
open_world_hint = false
)
)]
fn restore_backup(&self, _p: Parameters<RestoreParams>) -> CallToolResult {
// Hand-routed in call_tool for confirm-gated execution; this
// entry exists for discovery and dry-run dispatch without
// interactive approvals.
error_tool_result(
"restore_backup is hand-routed for confirmation; pass dryRun=true for the plan",
)
}
#[tool(
name = "audit_cleanup",
title = "Clean up audit log + old backups",
description = "Prune audit entries older than retention.auditDays and backups older than retention.backupDays. Hard size caps trigger an additional 20% trim. VACUUMs the file. Pass dryRun=true to preview.",
annotations(
destructive_hint = true,
idempotent_hint = true,
open_world_hint = false
)
)]
fn audit_cleanup(&self, Parameters(p): Parameters<AuditCleanupParams>) -> CallToolResult {
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("config load failed: {e}")),
};
let r = crate::audit::retention::cleanup_audit(&self.ctx.audit, &cfg.retention, p.dry_run);
json_tool_result(json!({
"auditDeleted": r.audit_deleted,
"auditDeletedByCategory": serde_json::Map::from_iter(
r.audit_deleted_by_category
.iter()
.map(|(k, v)| (k.to_string(), serde_json::json!(v))),
),
"backupDeleted": r.backup_deleted,
"bytesReclaimed": r.bytes_reclaimed,
"ranAt": r.ran_at,
"dryRun": p.dry_run,
}))
}
#[tool(
name = "set_retention",
title = "Update retention / cleanup config",
description = "Configure per-category retention (read=7, write=30, ddl=90, admin=180, txCtrl=7 by default), backup retention, hard size caps, and how often auto-cleanup runs on server boot. Pass any subset; missing fields keep current values.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn set_retention(&self, Parameters(p): Parameters<RetentionParams>) -> CallToolResult {
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("config load failed: {e}")),
};
let mut next = cfg.retention.clone();
if let Some(days) = p.retention_days_by_category {
let d = &mut next.retention_days_by_category;
if let Some(v) = days.read {
d.read = v;
}
if let Some(v) = days.write {
d.write = v;
}
if let Some(v) = days.ddl {
d.ddl = v;
}
if let Some(v) = days.admin {
d.admin = v;
}
if let Some(v) = days.tx_ctrl {
d.tx_ctrl = v;
}
}
if let Some(v) = p.backup_days {
next.backup_days = v;
}
if let Some(v) = p.audit_max_mb {
next.audit_max_mb = v.max(10);
}
if let Some(v) = p.backup_max_mb {
next.backup_max_mb = v.max(10);
}
if let Some(v) = p.auto_cleanup_hours {
next.auto_cleanup_hours = v.min(720);
}
if let Some(v) = p.redact_sql_in_log {
next.redact_sql_in_log = v;
}
if let Some(v) = p.tamper_evident_chain {
next.tamper_evident_chain = v;
}
let revision = cfg.revision;
let to_persist = next.clone();
match self.ctx.config.update(revision, move |c| {
c.retention = to_persist;
Ok(())
}) {
Ok(()) => json_tool_result(serde_json::to_value(&next).unwrap_or_default()),
Err(e) => error_tool_result(e.to_string()),
}
}
#[tool(
name = "history_search",
title = "Unified history (MCP audit + Sequel Ace)",
description = "Merge our audit_log with Sequel Ace queryHistory.db, sorted by timestamp DESC. Each row has a source field (mcp | sequel-ace). Use source=mcp or source=sequel-ace to filter to one. Useful when you want a single timeline regardless of where a query was run.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn history_search(&self, Parameters(p): Parameters<HistorySearchParams>) -> CallToolResult {
let source = p.source.as_deref().unwrap_or("both");
let limit = p.limit.unwrap_or(200).min(5000);
let mut out: Vec<serde_json::Value> = Vec::new();
if source == "mcp" || source == "both" {
let filters = crate::audit::AuditSearchFilters {
since: p.since_iso.clone(),
until: p.until_iso.clone(),
connection: p.connection.clone(),
category: None,
outcome: None,
limit: limit.saturating_mul(2),
};
if let Ok(rows) = crate::audit::search_audit_log(&self.ctx.audit, &filters) {
for r in rows {
let sql = r.sql_redacted.clone();
if let Some(needle) = &p.search
&& !sql.to_lowercase().contains(&needle.to_lowercase())
{
continue;
}
out.push(json!({
"source": "mcp",
"ts": r.ts,
"sql": sql,
"connection": r.connection,
"category": r.category,
"outcome": r.outcome,
"decision": r.decision,
"databases": r.databases,
"durationMs": r.duration_ms,
"affectedRows": r.affected_rows,
"backupId": r.backup_id,
}));
}
}
}
if source == "sequel-ace" || source == "both" {
let filters = crate::importer::history::SequelAceHistoryFilters {
since_iso: p.since_iso.as_deref(),
search: p.search.as_deref(),
limit: Some(limit.saturating_mul(2)),
};
for r in crate::importer::read_sequel_ace_history(&filters, None) {
if let Some(until) = &p.until_iso
&& r.created_at_iso.as_str() >= until.as_str()
{
continue;
}
out.push(json!({
"source": "sequel-ace",
"ts": r.created_at_iso,
"sql": r.query,
"sequelAceId": r.id,
}));
}
}
out.sort_by(|a, b| b["ts"].as_str().cmp(&a["ts"].as_str()));
out.truncate(limit as usize);
json_tool_result(json!({ "count": out.len(), "rows": out }))
}
#[tool(
name = "sequel_ace_history",
title = "Read Sequel Ace query history",
description = "Read the queryHistory.db that Sequel Ace maintains in its sandbox. Returns distinct queries the user has run in the GUI (deduplicated by Sequel Ace, with latest createdTime). Read-only — no modification. Optional sinceIso, search (LIKE %text%), limit (default 200, max 5000).",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn sequel_ace_history(
&self,
Parameters(p): Parameters<SequelAceHistoryParams>,
) -> CallToolResult {
let stat = crate::importer::stat_sequel_ace_history(None);
if !stat.exists {
return error_tool_result(format!(
"Sequel Ace queryHistory.db not found at {}. Open Sequel Ace and run at least one query first, or check that Sequel Ace is installed.",
stat.path.display()
));
}
let filters = crate::importer::history::SequelAceHistoryFilters {
since_iso: p.since_iso.as_deref(),
search: p.search.as_deref(),
limit: p.limit,
};
let rows: Vec<serde_json::Value> = crate::importer::read_sequel_ace_history(&filters, None)
.into_iter()
.map(|r| {
json!({
"id": r.id,
"query": r.query,
"createdTime": r.created_time,
"createdAtIso": r.created_at_iso,
})
})
.collect();
let returned = rows.len();
json_tool_result(json!({
"source": "sequel-ace",
"path": stat.path.display().to_string(),
"totalAvailable": stat.entry_count,
"returned": returned,
"note": "Sequel Ace dedupes by query text — only the latest createdTime is kept per distinct query.",
"rows": rows,
}))
}
#[tool(
name = "import_from_sequel_ace",
title = "Import connections from Sequel Ace",
description = "Read Sequel Ace Favorites.plist, copy connections (and optionally passwords via /usr/bin/security; macOS will prompt user to allow access) into our config + keychain. Sequel Ace data is never modified.",
annotations(idempotent_hint = true, open_world_hint = false)
)]
fn import_from_sequel_ace(&self, Parameters(p): Parameters<ImportParams>) -> CallToolResult {
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("config load failed: {e}")),
};
let revision = cfg.revision;
let copy = p.copy_passwords;
let secrets = self.ctx.secrets.clone();
let result = match self.ctx.config.update(revision, move |c| {
Ok(crate::importer::import_from_sequel_ace(
c, &*secrets, copy, None, None,
))
}) {
Ok(r) => r,
Err(e) => return error_tool_result(e.to_string()),
};
json_tool_result(json!({
"total": result.total,
"imported": result.imported,
"withPasswords": result.with_passwords,
"skipped": result.skipped.iter()
.map(|(name, reason)| json!({ "name": name, "reason": reason }))
.collect::<Vec<_>>(),
}))
}
#[tool(
name = "doctor",
title = "Diagnostic report",
description = "Print a sanitized JSON diagnostic of the MCP install: runtime versions, config state, every configured connection (host/user/db, policy), password presence. Contains zero passwords and zero secrets — redact hostnames before posting publicly.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
fn doctor(&self, _p: Parameters<EmptyParams>) -> CallToolResult {
let cfg = match self.ctx.config.load() {
Ok(c) => c,
Err(e) => return error_tool_result(format!("config load failed: {e}")),
};
let default = cfg.default_connection.as_deref();
let connections: Vec<serde_json::Value> = cfg
.connections
.iter()
.map(|c| {
json!({
"name": c.name(),
"driver": if c.is_mysql() { "mysql" } else { "sqlite" },
"database": c.database(),
"policy": c.policy(),
"isDefault": Some(c.name()) == default,
})
})
.collect();
json_tool_result(json!({
"app": crate::PACKAGE_NAME,
"version": crate::PACKAGE_VERSION,
"runtime": {
"platform": std::env::consts::OS,
"arch": std::env::consts::ARCH,
},
"defaultConnection": cfg.default_connection,
"connections": connections,
"note": "no passwords or Keychain secrets included; hostnames/usernames/SQLite paths/key paths ARE included — redact before posting publicly.",
}))
}
}
impl SequelServer {
/// Run one statement through the gate with no interactive approvals
/// (read-only helper tools; confirm-gated statements report
/// unavailable — use query/execute for those).
fn run_sql_blocking(&self, args: RunSqlArgs, expect_read_only: bool) -> CallToolResult {
let deps = self.gate_deps_blocking(Box::new(gate::UnavailableSink));
match gate::run_sql(&deps, &args, expect_read_only) {
Ok(out) => json_tool_result(gate::outcome_to_json(&out)),
Err(e) => error_tool_result(e.to_string()),
}
}
/// Modern-era (2026-07-28) query/execute: MRTR approvals. The first
/// confirm-required call returns `input_required` with an
/// elicitation/create input request and an OPAQUE SERVER-SIDE
/// one-shot `requestState` (a random token; every binding — tool,
/// connection, operation digest, policy revision, DDL plan targets,
/// expiry — lives in the in-process pending store, never on the
/// wire). The retry echoes the token plus `inputResponses`; after
/// atomic single-use consumption and revalidation the statement
/// executes through the normal gate with the pre-decided outcome
/// injected. For MySQL DROP statements the plan-time preflight fixes
/// the approved target set BEFORE the approval is issued, and the
/// retry fails closed (`ddl_precondition_changed`) if any target
/// changed existence in between.
async fn call_sql_modern(
&self,
request: rmcp::model::CallToolRequestParams,
p: SqlParams,
expect_read_only: bool,
tool: &'static str,
) -> Result<rmcp::model::CallToolResponse, rmcp::ErrorData> {
use rmcp::model::{
ElicitRequestParams, ElicitationSchema, InputRequest, InputRequiredResult,
};
// Cheap pure recomputation of the operation identity + decision.
let cfg = self
.ctx
.config
.load()
.map_err(|e| rmcp::ErrorData::internal_error(e.to_string(), None))?;
let conn = cfg
.resolve(p.connection.as_deref())
.cloned()
.ok_or_else(|| {
rmcp::ErrorData::invalid_params(
gate::no_connection_message(p.connection.as_deref()),
None,
)
})?;
let dialect = if conn.is_mysql() {
crate::policy::classifier::Dialect::MySql
} else {
crate::policy::classifier::Dialect::SQLite
};
let classified =
crate::policy::classifier::classify_statement(&p.sql, dialect).map_err(|e| {
rmcp::ErrorData::invalid_params(format!("cannot classify: {}", e.message()), None)
})?;
if expect_read_only && classified.category != crate::policy::model::SqlCategory::Read {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"query tool only accepts read statements (got {}). Use the \"execute\" tool for non-read statements.",
classified.category
))));
}
let fallback = p
.database
.clone()
.or_else(|| conn.database().map(str::to_string));
let resolution = crate::policy::resolver::resolve(&conn, &classified, fallback.as_deref());
let run_with_sink =
|outcome: Box<dyn gate::ApprovalSink>, expected_ddl: Option<Vec<(String, String)>>| {
let args = RunSqlArgs {
connection: p.connection.clone(),
sql: p.sql.clone(),
database: p.database.clone(),
expected_ddl_targets: expected_ddl,
};
let deps = self.gate_deps_blocking(outcome);
tokio::task::spawn_blocking(move || gate::run_sql(&deps, &args, expect_read_only))
};
if resolution.action != crate::policy::model::PolicyAction::Confirm {
let out = run_with_sink(Box::new(gate::UnavailableSink), None)
.await
.map_err(|e| rmcp::ErrorData::internal_error(format!("gate join: {e}"), None))?
.map_err(|e| rmcp::ErrorData::internal_error(e.to_string(), None))?;
return Ok(CallToolResponse::Complete(super::json_tool_result(
gate::outcome_to_json(&out),
)));
}
// Plan-time DDL preflight for MySQL DROP statements: fix the
// approved target set BEFORE any approval is issued. The retry
// can then only ever execute targets that existed when the user
// approved — a target created in between fails closed.
let mut approved_ddl: Option<Vec<(String, String)>> = None;
let mut plan_split: Option<(Vec<String>, Vec<String>)> = None;
if conn.is_mysql()
&& classified.category == crate::policy::model::SqlCategory::Ddl
&& classified.ast_type == "drop"
{
let Connection::Mysql(mysql_conn) = &conn else {
unreachable!("checked is_mysql above");
};
let pw = match self
.ctx
.secrets
.get_password(&mysql_conn.name, &mysql_conn.user)
{
Ok(pw) => pw,
Err(_) => {
return Err(rmcp::ErrorData::invalid_params(
format!(
"no password stored for connection {:?}; cannot plan the approval",
mysql_conn.name
),
None,
));
}
};
let preflight = async {
// Tunnel connections plan through the SAME tunnel runtime
// as the gate (a tunnel-only database must be plannable,
// and the plan's pool must carry the tunnel generation).
let (host, port, tunnel_generation) = if let Some(ssh) = &mysql_conn.ssh {
let ssh_pw = self
.ctx
.secrets
.get_password(&format!("{}::ssh", mysql_conn.name), &ssh.user)
.ok();
let lease = crate::sql::ssh::tunnel_endpoint(
&mysql_conn.name,
ssh,
ssh_pw.as_ref().map(|p| p.as_str()),
&mysql_conn.host,
mysql_conn.port,
cfg.revision,
)
.await
.map_err(|e| format!("ssh tunnel: {e}"))?;
(lease.host, lease.port, Some(lease.generation))
} else {
(mysql_conn.host.clone(), mysql_conn.port, None)
};
let pool = crate::sql::mysql::pool_manager()
.verified_pool(
mysql_conn,
&pw,
fallback.as_deref(),
cfg.revision,
Some(&host),
Some(port),
tunnel_generation,
)
.await
.map_err(|e| format!("pool initialization failed: {e}"))?;
let mut pconn = pool
.get_conn()
.await
.map_err(|e| format!("connection failed: {e}"))?;
crate::sql::ddl::preflight_ddl(&mut pconn, &classified, fallback.as_deref())
.await
.map_err(|e| e.to_string())
};
match preflight.await {
Ok(crate::sql::ddl::DdlPreflight::Present) => {
let mut targets = Vec::new();
for t in &classified.mutated_tables {
if let Some(schema) = t.database.clone().or_else(|| fallback.clone()) {
targets.push((schema, t.table.clone()));
}
}
approved_ddl = Some(targets);
}
Ok(crate::sql::ddl::DdlPreflight::Mixed { existing, missing }) => {
plan_split = Some((
existing.iter().map(|(s, t)| format!("{s}.{t}")).collect(),
missing.iter().map(|(s, t)| format!("{s}.{t}")).collect(),
));
approved_ddl = Some(existing);
}
Ok(crate::sql::ddl::DdlPreflight::MissingNoOp(missing)) => {
plan_split = Some((
Vec::new(),
missing.iter().map(|(s, t)| format!("{s}.{t}")).collect(),
));
approved_ddl = Some(Vec::new());
}
Err(e) => {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"cannot plan the approval (DDL preflight failed; nothing executed): {e}"
))));
}
}
}
// Bind the approval to the effective database scope as well:
// replaying the same SQL against a different database arg (or a
// changed connection default) must fail with mrtr_invalid_state.
let digest =
super::mrtr::operation_digest(&p.sql, conn.name(), fallback.as_deref(), cfg.revision);
// Retry: consume + validate the echoed state, parse the response,
// execute with the plan-approved DDL target set.
if let Some(responses) = &request.input_responses {
if serde_json::to_string(responses)
.map(|s| s.len())
.unwrap_or(usize::MAX)
> super::limits::MAX_INPUT_RESPONSES_BYTES
{
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"[argument_too_large] inputResponses exceeds {} bytes",
super::limits::MAX_INPUT_RESPONSES_BYTES
))));
}
let state = request
.request_state
.as_deref()
.ok_or_else(|| rmcp::ErrorData::invalid_params("missing requestState", None))?;
if state.len() > super::limits::MAX_REQUEST_STATE_BYTES {
return Err(rmcp::ErrorData::invalid_params(
format!(
"[mrtr_invalid_state] requestState exceeds {} bytes",
super::limits::MAX_REQUEST_STATE_BYTES
),
None,
));
}
let pending = super::mrtr::take(state, tool, conn.name(), &digest, cfg.revision)
.map_err(|e| {
rmcp::ErrorData::invalid_params(
format!("{}; approval rejected", e.message()),
None,
)
})?;
let value = responses.get("approval").cloned().ok_or_else(|| {
rmcp::ErrorData::invalid_params("missing approval response", None)
})?;
match super::mrtr::parse_response(&value) {
crate::approval::ConfirmOutcome::Chosen(crate::approval::GrantChoice::Decline) => {
Ok(CallToolResponse::Complete(error_tool_result(
"User declined confirmation. Statement not executed.",
)))
}
outcome @ crate::approval::ConfirmOutcome::Chosen(_) => {
let out = run_with_sink(
Box::new(PreDecidedSink(outcome)),
pending.approved_ddl_targets,
)
.await
.map_err(|e| rmcp::ErrorData::internal_error(format!("gate join: {e}"), None))?
.map_err(|e| rmcp::ErrorData::internal_error(e.to_string(), None))?;
Ok(CallToolResponse::Complete(super::json_tool_result(
gate::outcome_to_json(&out),
)))
}
crate::approval::ConfirmOutcome::Unavailable { reason } => {
Ok(CallToolResponse::Complete(error_tool_result(format!(
"confirmation required, but no prompt could be shown: {reason}. Statement not executed - nothing was changed. This is not a refusal."
))))
}
}
} else {
// First call: emit the input request (redacted SQL preview +
// the plan-time target split for mixed DROP statements).
let snippet = if p.sql.len() > 800 {
format!("{}…", &p.sql[..800])
} else {
p.sql.clone()
};
let tables: Vec<String> = resolution
.contributions
.iter()
.map(|c| format!("{}.{}", c.table.database, c.table.table))
.collect();
let plan_note = match &plan_split {
Some((exec, absent)) => format!(
"\n\nConfirmed to exist at plan time (will be affected): {}\nAbsent at plan time (skipped, recorded in audit): {}",
if exec.is_empty() {
"(none — this will be a no-op)".to_string()
} else {
exec.join(", ")
},
absent.join(", ")
),
None => String::new(),
};
let message = format!(
"About to run a {} statement on {}.\n\n--- SQL ---\n{}\n--- end ---\n\nAffected tables: {}{}\n\nPick an authorization scope.",
classified.category,
conn.name(),
snippet,
if tables.is_empty() {
"(statement scope)".to_string()
} else {
tables.join(", ")
},
plan_note,
);
let mut input_requests = std::collections::BTreeMap::new();
input_requests.insert(
"approval".to_string(),
InputRequest::Elicitation(rmcp::model::ElicitRequest::new(
ElicitRequestParams::FormElicitationParams {
meta: None,
message,
requested_schema: ElicitationSchema::from_json_schema(
serde_json::json!({
"type": "object",
"properties": {
"choice": {
"type": "string",
"title": "Authorization",
"enum": ["once", "session", "decline"]
}
},
"required": ["choice"]
})
.as_object()
.cloned()
.unwrap_or_default(),
)
.map_err(|e| {
rmcp::ErrorData::internal_error(
format!("elicitation schema: {e}"),
None,
)
})?,
},
)),
);
let pending = super::mrtr::PendingApproval {
tool,
connection: conn.name().to_string(),
operation_digest: digest,
policy_revision: cfg.revision,
approved_ddl_targets: approved_ddl,
expires_at: std::time::Instant::now() + super::mrtr::TTL,
};
let state = super::mrtr::issue(pending);
Ok(rmcp::model::CallToolResponse::InputRequired(
InputRequiredResult::new(Some(input_requests), Some(state)),
))
}
}
}
impl SequelServer {
/// add_connection: validates arguments, elicits the password through
/// the session peer (never via tool args), stores it in the secret
/// store (macOS Keychain; the in-memory store under test mode), and
/// upserts the connection. Any decline/cancel/unavailable prompt
/// leaves config AND secrets untouched.
async fn call_add_connection(
&self,
p: AddConnectionParams,
peer: rmcp::service::Peer<rmcp::RoleServer>,
) -> Result<rmcp::model::CallToolResponse, rmcp::ErrorData> {
use crate::config::{BridgeTool, MySqlConnection, SshDocker, SshHostKeyPolicy, SshTunnel};
let preset = match parse_preset(&p.policy_preset) {
Ok(v) => v,
Err(e) => {
return Ok(CallToolResponse::Complete(error_tool_result(e)));
}
};
if p.name.trim().is_empty() {
return Ok(CallToolResponse::Complete(error_tool_result(
"connection name must not be empty",
)));
}
if p.host.trim().is_empty() {
return Ok(CallToolResponse::Complete(error_tool_result(
"host must not be empty",
)));
}
if p.user.trim().is_empty() {
return Ok(CallToolResponse::Complete(error_tool_result(
"user must not be empty",
)));
}
let bridge_tool = match p.ssh_docker_bridge_tool.as_deref() {
None => BridgeTool::Nc,
Some("nc") => BridgeTool::Nc,
Some("ncat") => BridgeTool::Ncat,
Some("socat") => BridgeTool::Socat,
Some(other) => {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"unknown bridge tool {other:?} (nc | ncat | socat)"
))));
}
};
let host_key_policy = match p.ssh_host_key_policy.as_deref() {
None => None,
Some("lenient") => Some(SshHostKeyPolicy::Lenient),
Some("strict") => Some(SshHostKeyPolicy::Strict),
Some(other) => {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"unknown host key policy {other:?} (lenient | strict)"
))));
}
};
let ssh = match (p.ssh_host.as_deref(), p.ssh_user.as_deref()) {
(Some(host), Some(user)) if !host.trim().is_empty() && !user.trim().is_empty() => {
Some(SshTunnel {
host: host.trim().to_string(),
port: p.ssh_port.unwrap_or(22),
user: user.trim().to_string(),
auth_method: if p.ssh_key_path.is_some() {
crate::config::SshAuthMethod::Key
} else {
crate::config::SshAuthMethod::Password
},
private_key_path: p.ssh_key_path.clone(),
docker: p
.ssh_docker_container
.as_deref()
.map(|container| SshDocker {
container: container.to_string(),
bridge_tool,
}),
host_key_policy,
host_key_policy_migrated: false,
known_hosts_path: p.ssh_known_hosts_path.clone(),
})
}
(None, None) => None,
_ => {
return Ok(CallToolResponse::Complete(error_tool_result(
"ssh tunnel requires both ssh_host and ssh_user",
)));
}
};
let port = p.port.unwrap_or(3306);
let mut mc = MySqlConnection {
name: p.name.clone(),
host: p.host.trim().to_string(),
port,
user: p.user.trim().to_string(),
database: p.database.clone(),
ssl: p.ssl.unwrap_or(false),
ssl_server_name: p.ssl_server_name.clone(),
ssl_ca_path: p.ssl_ca_path.clone(),
ssh,
..MySqlConnection::default()
};
mc.policy = crate::policy::model::policy_from_preset(preset);
let conn = Connection::Mysql(mc);
if let Err(e) = conn.validate() {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"{e}"
))));
}
// Elicit the password LAST, after every argument check, so a
// prompt only ever appears for a connection that would save.
let message = format!(
"Enter MySQL/MariaDB password for {}@{}:{} (connection \"{}\"). Stored locally in the macOS Keychain; never in tool arguments or logs.",
p.user.trim(),
p.host.trim(),
port,
p.name
);
let password = match super::confirm::run_password_elicitation(&peer, message).await {
Ok(pw) => pw,
Err(reason) => {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"Password capture cancelled. Connection not saved. ({reason})"
))));
}
};
let cfg = self
.ctx
.config
.load()
.map_err(|e| rmcp::ErrorData::internal_error(e.to_string(), None))?;
let (name, user, preset_name) = (
p.name.clone(),
p.user.trim().to_string(),
preset.as_str().to_string(),
);
match self.ctx.config.update(cfg.revision, move |c| {
upsert(c, conn);
Ok(())
}) {
Ok(()) => {
if let Err(e) = self.ctx.secrets.set_password(&name, &user, &password) {
// The connection saved but the secret did not: say so
// plainly rather than pretending all is well.
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"Saved connection \"{name}\", but storing the password failed: {e}. Re-add the connection or store the password manually."
))));
}
Ok(CallToolResponse::Complete(text_tool_result(format!(
"Saved connection \"{name}\" with policy preset \"{preset_name}\". Password stored in the secret store (macOS Keychain)."
))))
}
Err(e) => Ok(CallToolResponse::Complete(error_tool_result(format!(
"{e}"
)))),
}
}
/// restore_backup: dry-run returns the plan; execution is
/// confirmation-gated (MRTR in the modern era, elicitation in the
/// legacy era) and replays the plan on ONE connection inside ONE
/// transaction, after a policy deny-check. Faithful to the legacy
/// tool, which confirmed once and executed directly.
async fn call_restore(
&self,
p: RestoreParams,
modern: bool,
peer: rmcp::service::Peer<rmcp::RoleServer>,
request: rmcp::model::CallToolRequestParams,
) -> Result<rmcp::model::CallToolResponse, rmcp::ErrorData> {
use crate::backup::restore::{RestoreDialect, plan_restore};
let cfg = self
.ctx
.config
.load()
.map_err(|e| rmcp::ErrorData::internal_error(e.to_string(), None))?;
let Some(detail) = crate::backup::restore::get_backup(&self.ctx.audit, p.backup_id) else {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"Backup #{} not found.",
p.backup_id
))));
};
let Some(conn) = cfg.resolve(Some(detail.connection.as_str())).cloned() else {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"Connection {:?} referenced by backup no longer exists.",
detail.connection
))));
};
let dialect = if conn.is_mysql() {
RestoreDialect::MySql
} else {
RestoreDialect::SQLite
};
let plan = match plan_restore(&self.ctx.audit, p.backup_id, dialect) {
Ok(plan) => plan,
Err(e) => {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"Cannot plan restore: {e}"
))));
}
};
if p.dry_run {
// Per-statement policy preview (review P1): show what the
// two-layer gate says about EVERY replayed statement before
// anything executes — the manual-review surface.
let preview = match resolve_restore_plan(&conn, &detail, &plan) {
Ok(res) => res
.iter()
.enumerate()
.map(|(i, (c, r))| {
json!({
"statement": i,
"category": c.category.as_str(),
"action": r.action,
"databases": c.target_databases,
})
})
.collect::<Vec<_>>(),
Err(reason) => vec![json!({ "wouldDeny": true, "reason": reason })],
};
return Ok(CallToolResponse::Complete(json_tool_result(json!({
"backupId": p.backup_id,
"connection": detail.connection,
"rowCount": plan.row_count,
"statementCount": plan.statements.len(),
"warnings": plan.warnings,
"isInsertHintDelete": plan.is_insert_hint_delete,
"firstStatementPreview":
plan.statements.first().map(|s| s.chars().take(240).collect::<String>()),
"policy": preview,
"note": "dry-run; pass dryRun=false to actually execute",
}))));
}
if plan.statements.is_empty() {
return Ok(CallToolResponse::Complete(json_tool_result(json!({
"backupId": p.backup_id,
"executedStatements": 0,
"warnings": plan.warnings,
"note": "nothing to restore",
}))));
}
// Confirmation: modern era = MRTR one-shot state bound to the
// backup id + policy revision; legacy era = elicitation.
let digest = {
use sha2::Digest;
let mut h = sha2::Sha256::new();
h.update(b"sequel-mcp/restore/v1\n");
h.update(detail.connection.as_bytes());
h.update(b"\n");
h.update(p.backup_id.to_le_bytes());
h.update(cfg.revision.to_le_bytes());
d_hex(h)
};
if modern {
if let Some(responses) = &request.input_responses {
let state = request
.request_state
.as_deref()
.ok_or_else(|| rmcp::ErrorData::invalid_params("missing requestState", None))?;
if state.len() > super::limits::MAX_REQUEST_STATE_BYTES {
return Err(rmcp::ErrorData::invalid_params(
format!(
"[mrtr_invalid_state] requestState exceeds {} bytes",
super::limits::MAX_REQUEST_STATE_BYTES
),
None,
));
}
let _pending = super::mrtr::take(
state,
"restore_backup",
&detail.connection,
&digest,
cfg.revision,
)
.map_err(|e| {
rmcp::ErrorData::invalid_params(
format!("{}; approval rejected", e.message()),
None,
)
})?;
let value = responses.get("approval").cloned().ok_or_else(|| {
rmcp::ErrorData::invalid_params("missing approval response", None)
})?;
match super::mrtr::parse_response(&value) {
crate::approval::ConfirmOutcome::Chosen(
crate::approval::GrantChoice::Decline,
) => Ok(CallToolResponse::Complete(error_tool_result(
"Restore declined.",
))),
outcome @ crate::approval::ConfirmOutcome::Chosen(_) => {
let _ = outcome;
self.execute_restore_plan(
&conn,
&detail,
&plan,
cfg.revision,
&crate::audit::WriteOptions {
redact_sql_in_log: cfg.retention.redact_sql_in_log,
tamper_evident_chain: cfg.retention.tamper_evident_chain,
},
)
.await
}
crate::approval::ConfirmOutcome::Unavailable { reason } => {
Ok(CallToolResponse::Complete(error_tool_result(format!(
"Restore needs confirmation, but no prompt could be shown: {reason}. Nothing was restored. This is not a refusal - set an explicit write policy, or restore outside this tool."
))))
}
}
} else {
use rmcp::model::{
ElicitRequestParams, ElicitationSchema, InputRequest, InputRequiredResult,
};
let message = format!(
"Restore backup #{}: {} statement(s) into {}.{}.{}\n\nWarnings: {}\n\nPick an authorization scope.",
p.backup_id,
plan.statements.len(),
detail.connection,
detail.database.as_deref().unwrap_or("<default>"),
detail.table_name,
if plan.warnings.is_empty() {
"(none)".to_string()
} else {
plan.warnings.join("; ")
}
);
let mut input_requests = std::collections::BTreeMap::new();
input_requests.insert(
"approval".to_string(),
InputRequest::Elicitation(rmcp::model::ElicitRequest::new(
ElicitRequestParams::FormElicitationParams {
meta: None,
message,
requested_schema: ElicitationSchema::from_json_schema(
serde_json::json!({
"type": "object",
"properties": {
"choice": {
"type": "string",
"title": "Authorization",
"enum": ["once", "decline"]
}
},
"required": ["choice"]
})
.as_object()
.cloned()
.unwrap_or_default(),
)
.map_err(|e| {
rmcp::ErrorData::internal_error(
format!("elicitation schema: {e}"),
None,
)
})?,
},
)),
);
let pending = super::mrtr::PendingApproval {
tool: "restore_backup",
connection: detail.connection.clone(),
operation_digest: digest,
policy_revision: cfg.revision,
approved_ddl_targets: None,
expires_at: std::time::Instant::now() + super::mrtr::TTL,
};
let state = super::mrtr::issue(pending);
Ok(rmcp::model::CallToolResponse::InputRequired(
InputRequiredResult::new(Some(input_requests), Some(state)),
))
}
} else {
// Legacy era: elicit directly through the session peer.
let message = format!(
"About to RESTORE backup #{}: {} statement(s) into {}.{}.{}",
p.backup_id,
plan.statements.len(),
detail.connection,
detail.database.as_deref().unwrap_or("<default>"),
detail.table_name
);
let outcome = super::confirm::run_elicitation(&peer, message).await;
match outcome {
crate::approval::ConfirmOutcome::Chosen(crate::approval::GrantChoice::Decline) => {
Ok(CallToolResponse::Complete(error_tool_result(
"Restore declined.",
)))
}
crate::approval::ConfirmOutcome::Chosen(_) => {
self.execute_restore_plan(
&conn,
&detail,
&plan,
cfg.revision,
&crate::audit::WriteOptions {
redact_sql_in_log: cfg.retention.redact_sql_in_log,
tamper_evident_chain: cfg.retention.tamper_evident_chain,
},
)
.await
}
crate::approval::ConfirmOutcome::Unavailable { reason } => {
Ok(CallToolResponse::Complete(error_tool_result(format!(
"Restore needs confirmation, but no prompt could be shown: {reason}. Nothing was restored. This is not a refusal."
))))
}
}
}
}
/// Replay the plan on one connection in one transaction, after a
/// policy deny-check (writes must not be denied for the backup's
/// table scope).
async fn execute_restore_plan(
&self,
conn: &crate::config::Connection,
detail: &crate::backup::restore::BackupDetail,
plan: &crate::backup::restore::RestorePlan,
revision: u64,
write_opts: &crate::audit::WriteOptions,
) -> Result<rmcp::model::CallToolResponse, rmcp::ErrorData> {
// Per-statement policy resolution (review P1): EVERY replayed
// statement must pass the two-layer gate, not just the first —
// an outer approval for "restore A" must never silently cover a
// statement the policy denies.
let resolutions = match resolve_restore_plan(conn, detail, plan) {
Ok(res) => res,
Err(reason) => {
// A denied restore still leaves its trace, linked to the
// backup (same discipline as gate denials).
let entry = crate::audit::AuditEntry {
request_id: format!("restore-{}-{}", detail.id, uuid::Uuid::new_v4()),
connection: detail.connection.clone(),
databases: Vec::new(),
category: crate::policy::model::SqlCategory::Write,
ast_type: None,
sql: plan
.statements
.first()
.map(|s| s.chars().take(800).collect::<String>())
.unwrap_or_default(),
decision: crate::policy::model::PolicyAction::Deny,
confirmed: false,
outcome: crate::approval::outcomes::ApprovalOutcome::Denied,
affected_rows: None,
duration_ms: None,
error: Some(reason.clone()),
backup_id: Some(detail.id),
approval_scope: None,
approval_digest: None,
policy_revision: Some(revision),
};
let _ = crate::audit::write_audit_entry(&self.ctx.audit, &entry, write_opts);
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"Restore denied by policy: {reason}"
))));
}
};
let started = std::time::Instant::now();
let result: Result<crate::backup::restore::RestoreOutcome, String> = match conn {
crate::config::Connection::Sqlite(sc) => {
let sc = sc.clone();
let stmts = plan.statements.clone();
tokio::task::spawn_blocking(move || {
let db = crate::sql::sqlite::open_sqlite_database(
&sc,
false,
sc.policy.stmt_timeout_ms,
)
.map_err(|e| e.to_string())?;
db.execute_batch("BEGIN IMMEDIATE")
.map_err(|e| e.to_string())?;
match crate::backup::restore::execute_restore_sqlite(&db, &fake_plan(&stmts)) {
Ok(r) => {
db.execute_batch("COMMIT").map_err(|e| e.to_string())?;
Ok(r)
}
Err(e) => {
let _ = db.execute_batch("ROLLBACK");
Err(e.to_string())
}
}
})
.await
.map_err(|e| format!("join: {e}"))
.and_then(|r| r)
}
crate::config::Connection::Mysql(mc) => {
let Some(password) = self.ctx.secrets.get_password(&mc.name, &mc.user).ok() else {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"No password for {:?}.",
mc.name
))));
};
let (host, port) = match &mc.ssh {
Some(ssh) => {
let ssh_pw = self
.ctx
.secrets
.get_password(&format!("{}::ssh", mc.name), &ssh.user)
.ok();
match crate::sql::ssh::tunnel_endpoint(
&mc.name,
ssh,
ssh_pw.as_ref().map(|p| p.as_str()),
&mc.host,
mc.port,
revision,
)
.await
{
Ok(lease) => (lease.host, lease.port),
Err(e) => {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"ssh tunnel: {e}"
))));
}
}
}
None => (mc.host.clone(), mc.port),
};
let mc = mc.clone();
let db = detail.database.clone();
let stmts = plan.statements.clone();
tokio::task::block_in_place(|| {
tokio::runtime::Handle::current().block_on(async {
let opts = crate::sql::mysql::build_opts(
&mc,
&password,
db.as_deref(),
Some(&host),
Some(port),
);
use mysql_async::prelude::Queryable;
let mut c = mysql_async::Conn::new(opts)
.await
.map_err(|e| format!("connect: {e}"))?;
c.query_drop("START TRANSACTION READ WRITE")
.await
.map_err(|e| format!("start tx: {e}"))?;
let mut affected: u64 = 0;
let mut run = 0usize;
for stmt in &stmts {
match c.query_iter(stmt.as_str()).await {
Ok(_) => {
affected += c.affected_rows();
run += 1;
}
Err(e) => {
let _ = c.query_drop("ROLLBACK").await;
return Err(format!("statement {run} failed: {e}"));
}
}
}
c.query_drop("COMMIT")
.await
.map_err(|e| format!("commit: {e}"))?;
let _ = c.disconnect().await;
Ok(crate::backup::restore::RestoreOutcome {
statements_run: run,
affected,
})
})
})
}
};
match result {
Ok(r) => {
// Per-statement audit rows (review P1): every replayed
// statement leaves its own row, linked to the backup.
// Write failures are SURFACED (never swallowed — the
// lesson from the audit-write blocker) but the restore
// already committed, so they become result warnings, not
// errors that would invite a retry.
let base_request = format!("restore-{}-{}", detail.id, uuid::Uuid::new_v4());
let mut audit_warnings: Vec<String> = Vec::new();
for (i, (classified, resolution)) in resolutions.iter().enumerate() {
let entry = crate::audit::AuditEntry {
request_id: format!("{base_request}-{i}"),
connection: detail.connection.clone(),
databases: classified.target_databases.clone(),
category: classified.category,
ast_type: Some(classified.ast_type.to_string()),
sql: plan.statements[i].clone(),
decision: resolution.action,
confirmed: true,
outcome: crate::approval::outcomes::ApprovalOutcome::Approved,
affected_rows: None,
duration_ms: None,
error: None,
backup_id: Some(detail.id),
approval_scope: Some("restore".to_string()),
approval_digest: None,
policy_revision: Some(revision),
};
if let Err(e) =
crate::audit::write_audit_entry(&self.ctx.audit, &entry, write_opts)
{
eprintln!(
"sequel-mcp: AUDIT WRITE FAILED for restore statement {i} of backup #{}: {e}",
detail.id
);
audit_warnings.push(format!("audit-write-failed (statement {i}): {e}"));
}
}
Ok(CallToolResponse::Complete(json_tool_result(json!({
"backupId": detail.id,
"executedStatements": r.statements_run,
"affected": r.affected,
"warnings": plan.warnings,
"auditWarnings": audit_warnings,
"durationMs": started.elapsed().as_millis() as u64,
}))))
}
Err(e) => {
// Failure still links the attempt to the backup.
let entry = crate::audit::AuditEntry {
request_id: format!("restore-{}-{}", detail.id, uuid::Uuid::new_v4()),
connection: detail.connection.clone(),
databases: Vec::new(),
category: resolutions
.first()
.map(|(c, _)| c.category)
.unwrap_or(crate::policy::model::SqlCategory::Write),
ast_type: None,
sql: plan
.statements
.first()
.map(|s| s.chars().take(800).collect::<String>())
.unwrap_or_default(),
decision: crate::policy::model::PolicyAction::Confirm,
confirmed: true,
outcome: crate::approval::outcomes::ApprovalOutcome::ExecutionError,
affected_rows: None,
duration_ms: Some(started.elapsed().as_millis() as i64),
error: Some(e.clone()),
backup_id: Some(detail.id),
approval_scope: Some("restore".to_string()),
approval_digest: None,
policy_revision: Some(revision),
};
if let Err(audit_err) =
crate::audit::write_audit_entry(&self.ctx.audit, &entry, write_opts)
{
eprintln!(
"sequel-mcp: audit write also failed for the errored restore: {audit_err}"
);
}
Ok(CallToolResponse::Complete(error_tool_result(format!(
"Restore failed: {e}"
))))
}
}
}
}
fn d_hex(h: sha2::Sha256) -> String {
use sha2::Digest;
h.finalize().iter().map(|b| format!("{b:02x}")).collect()
}
/// Per-statement classification + two-layer resolution for a restore
/// plan (review P1): EVERY replayed statement must pass the gate, not
/// just the first. Any Deny is a typed refusal naming the offending
/// statement and its table scope; unclassifiable statements refuse too.
#[allow(clippy::type_complexity)]
fn resolve_restore_plan(
conn: &crate::config::Connection,
detail: &crate::backup::restore::BackupDetail,
plan: &crate::backup::restore::RestorePlan,
) -> Result<
Vec<(
crate::policy::classifier::ClassifiedStatement,
crate::policy::resolver::Resolution,
)>,
String,
> {
let dialect = if conn.is_mysql() {
crate::policy::classifier::Dialect::MySql
} else {
crate::policy::classifier::Dialect::SQLite
};
let fallback = detail
.database
.clone()
.or_else(|| conn.database().map(str::to_string));
let mut out = Vec::with_capacity(plan.statements.len());
for (i, stmt) in plan.statements.iter().enumerate() {
let classified = crate::policy::classifier::classify_statement(stmt, dialect)
.map_err(|e| format!("cannot classify restore statement {i}: {}", e.message()))?;
let resolution = crate::policy::resolver::resolve(conn, &classified, fallback.as_deref());
if resolution.action == crate::policy::model::PolicyAction::Deny {
let tables: Vec<String> = resolution
.contributions
.iter()
.map(|c| format!("{}.{}", c.table.database, c.table.table))
.collect();
return Err(format!(
"statement {i} ({}) is denied by policy{}",
stmt.chars().take(120).collect::<String>(),
if tables.is_empty() {
String::new()
} else {
format!(" (tables: {})", tables.join(", "))
}
));
}
out.push((classified, resolution));
}
Ok(out)
}
fn fake_plan(stmts: &[String]) -> crate::backup::restore::RestorePlan {
crate::backup::restore::RestorePlan {
backup_id: 0,
statements: stmts.to_vec(),
row_count: 0,
warnings: Vec::new(),
is_insert_hint_delete: false,
}
}
/// Approval sink with a pre-decided outcome (MRTR retry path).
struct PreDecidedSink(crate::approval::ConfirmOutcome);
impl gate::ApprovalSink for PreDecidedSink {
fn confirm(&self, _request: gate::ApprovalRequest) -> crate::approval::ConfirmOutcome {
self.0.clone()
}
}
fn mysql_user(c: &Connection) -> Option<String> {
match c {
Connection::Mysql(m) => Some(m.user.clone()),
Connection::Sqlite(_) => None,
}
}
fn upsert(cfg: &mut crate::config::Config, conn: Connection) {
match cfg.connections.iter_mut().find(|c| c.name() == conn.name()) {
Some(slot) => *slot = conn,
None => cfg.connections.push(conn),
}
}
#[tool_handler]
impl ServerHandler for SequelServer {
fn get_info(&self) -> rmcp::model::ServerInfo {
super::build_server_info()
}
// ---- Prompts (legacy `server/prompts.ts` parity) ----
fn list_prompts(
&self,
_request: Option<rmcp::model::PaginatedRequestParams>,
_context: rmcp::service::RequestContext<rmcp::RoleServer>,
) -> impl Future<Output = Result<rmcp::model::ListPromptsResult, rmcp::ErrorData>> + '_ {
use rmcp::model::{Prompt, PromptArgument};
let setup = Prompt::new(
"setup-connection",
Some("Walks through adding either a MySQL/MariaDB connection with Keychain password capture or a SQLite file connection with no password."),
Some(vec![PromptArgument::new("suggestedName")
.with_title("Suggested name")
.with_description("Optional name to prefill")]),
)
.with_title("Set up a new database connection");
let analyze = Prompt::new(
"analyze-table",
Some("Read-only investigation: schema, row count, indexes, sample rows."),
Some(vec![
PromptArgument::new("connection")
.with_title("Connection")
.with_required(true),
PromptArgument::new("database").with_title("Database"),
PromptArgument::new("table")
.with_title("Table")
.with_required(true),
]),
)
.with_title("Analyze a table");
std::future::ready(Ok(rmcp::model::ListPromptsResult::with_all_items(vec![
setup, analyze,
])))
}
async fn get_prompt(
&self,
request: rmcp::model::GetPromptRequestParams,
_context: rmcp::service::RequestContext<rmcp::RoleServer>,
) -> Result<rmcp::model::GetPromptResponse, rmcp::ErrorData> {
use rmcp::model::{GetPromptResult, PromptMessage, Role};
let arg = |k: &str| -> Option<String> {
request
.arguments
.as_ref()
.and_then(|a| a.get(k))
.and_then(|v| v.as_str())
.map(str::to_string)
};
let (text, description): (String, Option<&str>) = match request.name.as_str() {
"setup-connection" => {
let name = arg("suggestedName");
let named = match &name {
Some(n) => format!(" called \"{n}\""),
None => String::new(),
};
(
format!(
"I want to add a new database connection{named}.\n\n\
First ask whether it is MySQL/MariaDB or SQLite. For MySQL/MariaDB, use \"add_connection\" and ask for: name, host, port (default 3306), user, database (optional), ssl (default false), policy preset (read-only | dev | admin), and optional SSH tunnel (host/port/user/keyPath). The tool will then prompt me for the password via elicitation. Do NOT include the password in the tool arguments. For SQLite, use \"add_sqlite_connection\" and ask for name, path, database/schema (usually main), and policy preset; no password is used."
),
Some("Set up a new database connection"),
)
}
"analyze-table" => {
let Some(connection) = arg("connection") else {
return Err(rmcp::ErrorData::invalid_params(
"missing required argument \"connection\"",
None,
));
};
let Some(table) = arg("table") else {
return Err(rmcp::ErrorData::invalid_params(
"missing required argument \"table\"",
None,
));
};
let database = arg("database");
let in_db = match &database {
Some(d) => format!(" in database `{d}`"),
None => String::new(),
};
(
format!(
"Analyze table `{table}`{in_db} on connection \"{connection}\". \
Use only read-only tools: describe_table, list_databases, and query (SELECT/SHOW or read-only SQLite PRAGMA only). Specifically: \
1) describe schema, 2) inspect indexes (SHOW INDEX for MySQL/MariaDB; PRAGMA index_list/index_info for SQLite), 3) SELECT COUNT(*), 4) SELECT * LIMIT 5. Summarize findings."
),
Some("Analyze a table"),
)
}
other => {
return Err(rmcp::ErrorData::invalid_params(
format!("unknown prompt: {other:?}"),
None,
));
}
};
Ok(rmcp::model::GetPromptResponse::from(
GetPromptResult::new(vec![PromptMessage::new_text(Role::User, text)])
.with_description(description.unwrap_or_default()),
))
}
// ---- Resources (legacy `server/resources.ts` parity) ----
fn list_resources(
&self,
_request: Option<rmcp::model::PaginatedRequestParams>,
_context: rmcp::service::RequestContext<rmcp::RoleServer>,
) -> impl Future<Output = Result<rmcp::model::ListResourcesResult, rmcp::ErrorData>> + '_ {
let resource = rmcp::model::Resource::new("sequel-mcp://connections", "connections")
.with_title("Configured connections")
.with_description("JSON listing of saved connections (no secrets).")
.with_mime_type("application/json");
std::future::ready(Ok(rmcp::model::ListResourcesResult::with_all_items(vec![
resource,
])))
}
async fn read_resource(
&self,
request: rmcp::model::ReadResourceRequestParams,
_context: rmcp::service::RequestContext<rmcp::RoleServer>,
) -> Result<rmcp::model::ReadResourceResponse, rmcp::ErrorData> {
use rmcp::model::{ReadResourceResult, ResourceContents};
if request.uri != *"sequel-mcp://connections" {
return Err(rmcp::ErrorData::invalid_params(
format!("unknown resource: {:?}", request.uri),
None,
));
}
let cfg = self
.ctx
.config
.load()
.map_err(|e| rmcp::ErrorData::internal_error(e.to_string(), None))?;
let presets: Vec<&str> = crate::policy::model::POLICY_PRESET_NAMES.to_vec();
let items: Vec<serde_json::Value> = cfg
.connections
.iter()
.map(|c| {
let (host, port, user, ssh_json) = match c {
crate::config::Connection::Mysql(m) => {
let ssh_json = m.ssh.as_ref().map(|ssh| {
json!({
"host": ssh.host,
"user": ssh.user,
"docker": ssh.docker.as_ref().map(|d| json!({
"container": d.container,
"bridgeTool": d.bridge_tool.as_str(),
})),
})
});
(json!(m.host), json!(m.port), json!(m.user), ssh_json)
}
crate::config::Connection::Sqlite(_) => (
serde_json::Value::Null,
serde_json::Value::Null,
serde_json::Value::Null,
None,
),
};
let has_password = c.is_mysql()
&& self
.ctx
.secrets
.has_password(c.name(), mysql_user(c).as_deref().unwrap_or(""));
json!({
"name": c.name(),
"driver": if c.is_mysql() { "mysql" } else { "sqlite" },
"host": host,
"port": port,
"user": user,
"path": if c.is_mysql() { serde_json::Value::Null } else {
match c {
crate::config::Connection::Sqlite(s) => json!(s.path),
_ => serde_json::Value::Null,
}
},
"database": c.database(),
"ssh": ssh_json,
"policy": c.policy(),
"presets": presets,
"hasPassword": has_password,
})
})
.collect();
let text = serde_json::to_string_pretty(&json!({ "connections": items }))
.unwrap_or_else(|_| "{\n \"connections\": []\n}".into());
Ok(rmcp::model::ReadResourceResponse::from(
ReadResourceResult::new(vec![
ResourceContents::text(text, "sequel-mcp://connections")
.with_mime_type("application/json"),
]),
))
}
async fn call_tool(
&self,
request: rmcp::model::CallToolRequestParams,
context: rmcp::service::RequestContext<rmcp::RoleServer>,
) -> Result<rmcp::model::CallToolResponse, rmcp::ErrorData> {
let name = request.name.as_ref();
if name == "add_connection" {
let arg_value =
serde_json::Value::Object(request.arguments.clone().unwrap_or_default());
let Ok(p) = serde_json::from_value::<AddConnectionParams>(arg_value) else {
return Ok(CallToolResponse::Complete(error_tool_result(
"invalid arguments for add_connection",
)));
};
return self.call_add_connection(p, context.peer.clone()).await;
}
if name == "restore_backup" {
let arg_value =
serde_json::Value::Object(request.arguments.clone().unwrap_or_default());
let Ok(p) = serde_json::from_value::<RestoreParams>(arg_value) else {
return Ok(CallToolResponse::Complete(error_tool_result(
"invalid arguments for restore_backup",
)));
};
let modern = context
.meta
.protocol_version()
.map(|v| v >= rmcp::model::ProtocolVersion::V_2026_07_28)
.unwrap_or(false);
return self
.call_restore(p, modern, context.peer.clone(), request)
.await;
}
if name == "query" || name == "execute" {
let expect_read_only = name == "query";
let tool: &'static str = if expect_read_only { "query" } else { "execute" };
let arg_value =
serde_json::Value::Object(request.arguments.clone().unwrap_or_default());
let params: Result<SqlParams, _> = serde_json::from_value(arg_value);
let Ok(p) = params else {
return Ok(CallToolResponse::Complete(error_tool_result(
"invalid arguments for query/execute",
)));
};
if p.sql.len() > super::limits::MAX_TOOL_ARGUMENT_BYTES {
return Ok(CallToolResponse::Complete(error_tool_result(format!(
"[argument_too_large] sql argument exceeds {} bytes",
super::limits::MAX_TOOL_ARGUMENT_BYTES
))));
}
// Modern (2026-07-28) era: rmcp's stdio server does not
// propagate per-request _meta capabilities into the legacy
// elicit path, so approvals go through MRTR here. The wire
// _meta lives in the RequestContext, not the deserialized
// params.
let modern = context
.meta
.protocol_version()
.map(|v| v >= rmcp::model::ProtocolVersion::V_2026_07_28)
.unwrap_or(false);
if modern {
return self
.call_sql_modern(request, p, expect_read_only, tool)
.await;
}
let args = RunSqlArgs {
connection: p.connection,
sql: p.sql,
database: p.database,
expected_ddl_targets: None,
};
// Gate on the blocking pool; elicitation asks flow back over a
// channel and are answered from the async runtime by a pump
// thread (block_on from a blocking thread is legal).
let (ask_tx, ask_rx) = std::sync::mpsc::sync_channel::<super::confirm::ElicitAsk>(4);
let sink = super::confirm::ElicitationSink::new(ask_tx);
let deps = self.gate_deps_blocking(Box::new(sink));
let handle =
tokio::task::spawn_blocking(move || gate::run_sql(&deps, &args, expect_read_only));
let peer = context.peer.clone();
let ipc_hub = self.ctx.approval_ipc.clone();
let pump = tokio::task::spawn_blocking(move || {
while let Ok(ask) = ask_rx.recv() {
match ask {
super::confirm::ElicitAsk::Request { message, reply } => {
let mut outcome = tokio::runtime::Handle::current().block_on(async {
super::confirm::run_elicitation(&peer, message.clone()).await
});
// Elicitation unavailable (client cannot
// prompt): fall back to the authenticated
// companion IPC and fail closed there too.
if matches!(
outcome,
crate::approval::ConfirmOutcome::Unavailable { .. }
) && let Some(hub) = &ipc_hub
{
let request = gate::ApprovalRequest::from_message(&message);
outcome =
tokio::runtime::Handle::current().block_on(hub.ask(request));
}
let _ = reply.send(outcome);
}
}
}
});
let outcome = match handle.await {
Ok(r) => r,
Err(e) => Err(gate::GateError::Execution(format!("gate task failed: {e}"))),
};
// The pump exits once the gate (and its sink) drop the channel.
let _ = pump.await;
let result = match outcome {
Ok(out) => super::json_tool_result(gate::outcome_to_json(&out)),
Err(e) => super::error_tool_result(e.to_string()),
};
return Ok(CallToolResponse::Complete(result));
}
// Everything else: delegate to the macro-generated router.
let tcc = ToolCallContext::new(self, request, context);
Self::tool_router().call(tcc).await
}
}