use std::io::{BufRead, Write};
use std::path::PathBuf;
use crate::command::{self, Arguments, Command, Context, Failed, OpenMode};
use crate::json::{self, Json};
pub const PROTOCOL: &str = "2025-06-18";
pub const MAX_REQUEST_BYTES: usize = 1024 * 1024;
pub const MAX_RESPONSE_BYTES: usize = 8 * 1024 * 1024;
pub const MAX_ROWS: usize = 10_000;
pub struct Settings {
pub database: String,
pub readonly: bool,
pub root: Option<PathBuf>,
pub limit: usize,
pub max_rows: usize,
pub max_time: std::time::Duration,
}
#[derive(Default)]
enum Lifecycle {
#[default]
AwaitingInitialize,
AwaitingInitializedNotification,
Ready,
}
#[derive(Default)]
pub struct Session {
lifecycle: Lifecycle,
}
impl Default for Settings {
fn default() -> Settings {
Settings {
database: ":memory:".to_string(),
readonly: false,
root: None,
limit: 200,
max_rows: MAX_ROWS,
max_time: std::time::Duration::from_secs(60),
}
}
}
pub fn tools() -> Vec<Json> {
command::COMMANDS
.iter()
.filter(|command| command.cli_only.is_none())
.map(tool_of)
.collect()
}
fn tool_of(command: &'static Command) -> Json {
json::object(vec![
("name", json::text(tool_name(command))),
(
"description",
json::text(format!("{}\n\n{}", command.summary, command.detail)),
),
("inputSchema", schema_of(command)),
])
}
pub fn tool_name(command: &Command) -> String {
format!("inillucent_{}", command.name.replace('-', "_"))
}
pub fn schema_of(command: &Command) -> Json {
let properties: Vec<(String, Json)> = command
.params
.iter()
.map(|param| {
let mut member = vec![
("type", json::text(param.kind.schema_type())),
("description", json::text(param.description)),
];
if param.kind == command::Kind::Values {
member.push((
"items",
json::object(vec![(
"type",
Json::Array(vec![
json::text("string"),
json::text("number"),
json::text("boolean"),
json::text("null"),
]),
)]),
));
}
if let Some(allowed) = command.allowed_values(param.name) {
member.push((
"enum",
Json::Array(allowed.iter().map(|value| json::text(*value)).collect()),
));
}
(param.name.to_string(), json::object(member))
})
.collect();
let required: Vec<Json> = command
.params
.iter()
.filter(|param| param.required)
.map(|param| json::text(param.name))
.collect();
json::object(vec![
("type", json::text("object")),
("properties", Json::Object(properties)),
("required", Json::Array(required)),
("additionalProperties", Json::Bool(false)),
])
}
pub fn serve<R: BufRead + Send + 'static>(
settings: Settings,
mut input: R,
output: &mut impl Write,
) -> Result<(), String> {
let mut context = Context::open_for(
&settings.database,
OpenMode::of(settings.readonly),
settings.root.clone(),
false,
)
.map_err(|failure| failure.message)?;
context.refuse_the_world();
context.limit = settings.limit.min(settings.max_rows);
context.set_max_rows(Some(settings.max_rows));
context.set_limits(
inillucent_driver::StatementLimits::served().with_time(Some(settings.max_time)),
);
let mut session = Session::default();
let cancel = context.cancel_flag();
context.preserve_cancellation();
let state: std::sync::Arc<std::sync::Mutex<Cancellation>> =
std::sync::Arc::new(std::sync::Mutex::new(Cancellation::default()));
let noted = std::sync::Arc::clone(&state);
let (lines, arriving) = std::sync::mpsc::channel::<Arrival>();
std::thread::spawn(move || {
let mut line = String::new();
loop {
line.clear();
let arrival = match read_request(&mut input, &mut line) {
Ok(0) => Arrival::Ended,
Ok(_) => {
if let Some(id) = cancellation_target(&line) {
if let Ok(mut held) = noted.lock() {
if held.running.as_deref() == Some(id.as_str()) {
cancel.store(true, std::sync::atomic::Ordering::Relaxed);
} else {
held.cancelled.push(id);
}
}
}
Arrival::Line(line.clone())
}
Err(TooLong) => Arrival::TooLong,
};
let ended = matches!(arrival, Arrival::Ended | Arrival::TooLong);
if lines.send(arrival).is_err() || ended {
return;
}
}
});
let mut line = String::new();
loop {
line.clear();
match arriving.recv() {
Ok(Arrival::Ended) | Err(_) => return Ok(()),
Ok(Arrival::Line(arrived)) => line.push_str(&arrived),
Ok(Arrival::TooLong) => {
let _ = writeln!(
output,
"{}",
error_response(
Json::Null,
-32600,
&format!(
"a request may not be longer than {MAX_REQUEST_BYTES} bytes, and this \
connection has sent one that is."
)
)
);
let _ = output.flush();
return Ok(());
}
}
if line.trim().is_empty() {
continue;
}
match claim(&line, &state, &context) {
Claim::Cancelled(id) => {
let answer = error_response(id, -32800, "this request was cancelled.");
writeln!(output, "{answer}").map_err(|error| error.to_string())?;
output.flush().map_err(|error| error.to_string())?;
continue;
}
Claim::Running => {}
}
let answered = handle_with_session(&mut context, &mut session, &line);
if let Ok(mut held) = state.lock() {
held.running = None;
}
if let Some(answer) = answered {
let answer = enforce_response_budget(answer);
writeln!(output, "{answer}").map_err(|error| error.to_string())?;
output.flush().map_err(|error| error.to_string())?;
}
}
}
fn enforce_response_budget(answer: String) -> String {
if answer.len() <= MAX_RESPONSE_BYTES {
return answer;
}
let id = response_id(&answer);
error_response(
id,
-32603,
&format!(
"this answer would have been {} bytes, past the {MAX_RESPONSE_BYTES} a \
reply may hold. Ask for fewer rows or fewer columns.",
answer.len()
),
)
}
enum Arrival {
Line(String),
Ended,
TooLong,
}
fn cancellation_target(line: &str) -> Option<String> {
let request = json::parse(line).ok()?;
if request.get("method").and_then(Json::text) != Some("notifications/cancelled") {
return None;
}
Some(
request
.get("params")
.and_then(|params| params.get("requestId"))
.map(id_text)
.unwrap_or_default(),
)
}
fn id_text(id: &Json) -> String {
match id {
Json::Text(text) => text.clone(),
Json::Int(number) => number.to_string(),
Json::Real(number) => number.to_string(),
other => format!("{other:?}"),
}
}
#[derive(Default)]
struct Cancellation {
running: Option<String>,
cancelled: Vec<String>,
}
enum Claim {
Cancelled(Json),
Running,
}
fn claim(line: &str, state: &std::sync::Mutex<Cancellation>, context: &Context) -> Claim {
let Ok(request) = json::parse(line) else {
return Claim::Running;
};
let Some(id) = request.get("id") else {
return Claim::Running;
};
let text = id_text(id);
let Ok(mut held) = state.lock() else {
return Claim::Running;
};
if let Some(at) = held.cancelled.iter().position(|named| *named == text) {
held.cancelled.remove(at);
return Claim::Cancelled(id.clone());
}
context
.cancel_flag()
.store(false, std::sync::atomic::Ordering::Relaxed);
held.running = Some(text);
Claim::Running
}
struct TooLong;
fn read_request(input: &mut impl BufRead, line: &mut String) -> Result<usize, TooLong> {
loop {
let mut bytes: Vec<u8> = Vec::new();
let ended = loop {
let mut one = [0u8; 1];
match input.read(&mut one) {
Ok(0) | Err(_) => break true,
Ok(_) => {}
}
let byte = one.first().copied().unwrap_or(b'\n');
if byte == b'\n' {
break false;
}
if bytes.len() >= MAX_REQUEST_BYTES {
return Err(TooLong);
}
bytes.push(byte);
};
let blank = bytes.iter().all(|byte| byte.is_ascii_whitespace());
if blank {
match ended {
true => return Ok(0),
false => continue,
}
}
line.push_str(&String::from_utf8_lossy(&bytes));
return Ok(line.len());
}
}
pub fn handle_with_session(
context: &mut Context,
session: &mut Session,
line: &str,
) -> Option<String> {
let request = match json::parse(line) {
Ok(request) => request,
Err(why) => return Some(error_response(Json::Null, -32700, &why)),
};
let Json::Object(_) = request else {
return Some(error_response(
Json::Null,
-32600,
"a request must be an object.",
));
};
let id = request.get("id").cloned().unwrap_or(Json::Null);
if !valid_request_id(&id) {
return Some(error_response(
Json::Null,
-32600,
"a request id must be a string, number, or null.",
));
}
if request.get("jsonrpc").and_then(Json::text) != Some("2.0") {
return Some(error_response(id, -32600, "'jsonrpc' must be '2.0'."));
}
let Some(method) = request.get("method").and_then(Json::text) else {
return Some(error_response(id, -32600, "a request needs a 'method'."));
};
let is_notification = request.get("id").is_none();
let params = request.get("params").cloned().unwrap_or(Json::Null);
if request.get("params").is_some() && !matches!(params, Json::Object(_) | Json::Array(_)) {
return Some(error_response(
Json::Null,
-32600,
"request params must be an object or array.",
));
}
if method == "initialize" {
if !matches!(session.lifecycle, Lifecycle::AwaitingInitialize) {
return Some(error_response(
id,
-32600,
"initialize was already completed.",
));
}
let result = initialize(¶ms);
if result.is_ok() {
session.lifecycle = Lifecycle::AwaitingInitializedNotification;
}
return response_for(id, is_notification, result);
}
if method == "notifications/initialized" {
if matches!(
session.lifecycle,
Lifecycle::AwaitingInitializedNotification
) {
session.lifecycle = Lifecycle::Ready;
return None;
}
return response_for(
id,
is_notification,
Err(Failed::misuse(
"notifications/initialized must follow initialize.",
)),
);
}
if !matches!(session.lifecycle, Lifecycle::Ready) {
return Some(error_response(
id,
-32002,
"MCP initialization must complete before this method is used.",
));
}
let result = match method {
"tools/list" => Ok(json::object(vec![("tools", Json::Array(tools()))])),
"tools/call" => call(context, ¶ms),
"ping" => Ok(json::object(vec![])),
_ if is_notification => return None,
other => {
return Some(error_response(
id,
-32601,
&format!("this server has no '{other}' method."),
))
}
};
response_for(id, is_notification, result)
}
pub fn handle(context: &mut Context, line: &str) -> Option<String> {
let mut session = Session {
lifecycle: Lifecycle::Ready,
};
handle_with_session(context, &mut session, line)
}
fn valid_request_id(id: &Json) -> bool {
matches!(
id,
Json::Null | Json::Int(_) | Json::Real(_) | Json::Text(_)
)
}
fn response_for(id: Json, is_notification: bool, result: Result<Json, Failed>) -> Option<String> {
if is_notification {
return None;
}
Some(match result {
Ok(value) => json::object(vec![
("jsonrpc", json::text("2.0")),
("id", id),
("result", value),
])
.write(),
Err(failure) => error_response(id, -32602, &failure.message),
})
}
fn initialize(params: &Json) -> Result<Json, Failed> {
let Json::Object(_) = params else {
return Err(Failed::misuse("initialize params must be an object."));
};
for (name, required) in [
("protocolVersion", true),
("capabilities", true),
("clientInfo", true),
] {
if required && params.get(name).is_none() {
return Err(Failed::misuse(format!("initialize needs '{name}'.")));
}
}
if params.get("protocolVersion").and_then(Json::text).is_none() {
return Err(Failed::misuse("'protocolVersion' has to be text."));
}
if !matches!(params.get("capabilities"), Some(Json::Object(_))) {
return Err(Failed::misuse("'capabilities' has to be an object."));
}
if !matches!(params.get("clientInfo"), Some(Json::Object(_))) {
return Err(Failed::misuse("'clientInfo' has to be an object."));
}
Ok(json::object(vec![
("protocolVersion", json::text(PROTOCOL)),
(
"capabilities",
json::object(vec![(
"tools",
json::object(vec![("listChanged", Json::Bool(false))]),
)]),
),
(
"serverInfo",
json::object(vec![
("name", json::text("inillucent")),
("version", json::text(env!("CARGO_PKG_VERSION"))),
]),
),
(
"instructions",
json::text(
"inillucent is an embedded SQL database that speaks SQLite's dialect, with \
full-text and vector search built in. Call inillucent_tables to see what is \
there, inillucent_describe before writing SQL against a table you did not \
create, inillucent_query to read and inillucent_exec to write. If a call comes \
back with status 'unsupported', that construct is not built yet - it is not a \
mistake in your SQL, and rewording it will not help.",
),
),
]))
}
fn call(context: &mut Context, params: &Json) -> Result<Json, Failed> {
let Some(name) = params.get("name").and_then(Json::text) else {
return Err(Failed::misuse("a tools/call needs a 'name'."));
};
let Some(command) = command::find(name) else {
return Ok(tool_error(&format!(
"there is no tool called '{name}'. Call tools/list to see what there is."
)));
};
if command.cli_only.is_some() {
return Ok(tool_error(&format!(
"'{name}' is not served over MCP: {}",
command.cli_only.unwrap_or_default()
)));
}
let arguments = Arguments::from_json(
command,
¶ms.get("arguments").cloned().unwrap_or(Json::Null),
)?;
let wants_json = arguments.text("output") == Some("json");
match command::run(command, context, &arguments) {
Ok(produced) => {
let body = match wants_json {
true => produced.to_json().pretty(0),
false => produced.text.clone(),
};
Ok(content(&body, false))
}
Err(failure) => {
let body = match wants_json {
true => failure.to_json(command.name).pretty(0),
false => failure.to_text(),
};
Ok(content(&body, true))
}
}
}
fn response_id(response: &str) -> Json {
json::parse(response)
.ok()
.and_then(|value| value.get("id").cloned())
.unwrap_or(Json::Null)
}
fn content(text: &str, failed: bool) -> Json {
json::object(vec![
(
"content",
Json::Array(vec![json::object(vec![
("type", json::text("text")),
("text", json::text(text)),
])]),
),
("isError", Json::Bool(failed)),
])
}
fn tool_error(message: &str) -> Json {
content(message, true)
}
fn error_response(id: Json, code: i64, message: &str) -> String {
json::object(vec![
("jsonrpc", json::text("2.0")),
("id", id),
(
"error",
json::object(vec![
("code", Json::Int(code)),
("message", json::text(message)),
]),
),
])
.write()
}
#[cfg(test)]
mod tests {
use super::*;
fn context() -> Context {
Context::open(":memory:", OpenMode::ReadWrite, None).expect("an in-memory database opens")
}
#[test]
fn the_tools_are_the_commands() {
let served: Vec<String> = tools()
.iter()
.filter_map(|tool| tool.get("name").and_then(Json::text).map(str::to_string))
.collect();
let expected: Vec<String> = command::COMMANDS
.iter()
.filter(|command| command.cli_only.is_none())
.map(tool_name)
.collect();
assert_eq!(served, expected);
assert!(served.contains(&"inillucent_query".to_string()));
assert!(!served.contains(&"inillucent_shell".to_string()));
}
#[test]
fn tool_names_have_no_dashes() {
for tool in tools() {
let name = tool
.get("name")
.and_then(Json::text)
.unwrap_or_default()
.to_string();
assert!(!name.contains('-'), "{name} has a dash in it");
}
}
#[test]
fn initialize_selects_the_supported_version() {
let mut session = Session::default();
let answer = handle_with_session(
&mut context(),
&mut session,
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\
\"params\":{\"protocolVersion\":\"2099-01-01\",\"capabilities\":{},\
\"clientInfo\":{\"name\":\"test\"}}}",
)
.unwrap_or_default();
assert!(answer.contains(&format!("\"protocolVersion\":\"{PROTOCOL}\"")));
assert!(answer.contains("\"name\":\"inillucent\""));
}
#[test]
fn invalid_requests_and_initialize_payloads_are_refused() {
for request in [
"{\"id\":41,\"method\":\"ping\"}",
"{\"jsonrpc\":\"1.0\",\"id\":42,\"method\":\"ping\"}",
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\"}",
] {
let answer = handle(&mut context(), request).unwrap_or_default();
assert!(answer.contains("\"error\""), "{answer}");
}
}
#[test]
fn invalid_json_rpc_member_types_are_refused() {
for request in [
"{\"jsonrpc\":\"2.0\",\"id\":31,\"method\":\"ping\",\"params\":\"bad\"}",
"{\"jsonrpc\":\"2.0\",\"id\":true,\"method\":\"ping\"}",
"{\"jsonrpc\":\"2.0\",\"id\":{},\"method\":\"ping\"}",
] {
let answer = handle(&mut context(), request).unwrap_or_default();
assert!(answer.contains("\"code\":-32600"), "{answer}");
assert!(answer.contains("\"id\":null"), "{answer}");
}
}
#[test]
fn initialization_must_complete_before_normal_methods() {
let mut held = context();
let mut session = Session::default();
let before = handle_with_session(
&mut held,
&mut session,
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/list\"}",
)
.unwrap_or_default();
assert!(before.contains("\"code\":-32002"), "{before}");
let initialized = handle_with_session(
&mut held,
&mut session,
"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"initialize\",\"params\":{\"protocolVersion\":\"2025-06-18\",\"capabilities\":{},\"clientInfo\":{\"name\":\"test\"}}}",
)
.unwrap_or_default();
assert!(initialized.contains("\"result\""), "{initialized}");
let waiting = handle_with_session(
&mut held,
&mut session,
"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"tools/list\"}",
)
.unwrap_or_default();
assert!(waiting.contains("\"code\":-32002"), "{waiting}");
assert!(handle_with_session(
&mut held,
&mut session,
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}",
)
.is_none());
let listed = handle_with_session(
&mut held,
&mut session,
"{\"jsonrpc\":\"2.0\",\"id\":4,\"method\":\"tools/list\"}",
)
.unwrap_or_default();
assert!(listed.contains("\"result\""), "{listed}");
}
#[test]
fn a_notification_gets_no_answer() {
assert!(handle(
&mut context(),
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}"
)
.is_none());
}
#[test]
fn a_call_creates_and_reads() {
let mut held = context();
let made = handle(
&mut held,
"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/call\",\"params\":{\
\"name\":\"inillucent_exec\",\"arguments\":{\
\"sql\":\"CREATE TABLE people (id INTEGER PRIMARY KEY, name TEXT)\"}}}",
)
.unwrap_or_default();
assert!(made.contains("\"isError\":false"), "{made}");
handle(
&mut held,
"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"tools/call\",\"params\":{\
\"name\":\"inillucent_exec\",\"arguments\":{\
\"sql\":\"INSERT INTO people VALUES (?1, ?2)\",\"params\":[1,\"Ada\"]}}}",
);
let read = handle(
&mut held,
"{\"jsonrpc\":\"2.0\",\"id\":4,\"method\":\"tools/call\",\"params\":{\
\"name\":\"inillucent_query\",\"arguments\":{\
\"sql\":\"SELECT name FROM people\"}}}",
)
.unwrap_or_default();
assert!(read.contains("Ada"), "{read}");
}
#[test]
fn an_unknown_tool_is_a_tool_error() {
let answer = handle(
&mut context(),
"{\"jsonrpc\":\"2.0\",\"id\":5,\"method\":\"tools/call\",\
\"params\":{\"name\":\"inillucent_nonsense\",\"arguments\":{}}}",
)
.unwrap_or_default();
assert!(answer.contains("\"isError\":true"));
assert!(answer.contains("\"result\""));
assert!(!answer.contains("\"error\""));
}
#[test]
fn a_broken_request_is_refused() {
let answer = handle(&mut context(), "{not json").unwrap_or_default();
assert!(answer.contains("-32700"));
assert!(answer.contains("\"id\":null"));
}
#[test]
fn an_unknown_method_is_refused() {
let answer = handle(
&mut context(),
"{\"jsonrpc\":\"2.0\",\"id\":6,\"method\":\"resources/list\"}",
)
.unwrap_or_default();
assert!(answer.contains("-32601"));
}
#[test]
fn schemas_declare_what_is_required() {
for command in command::COMMANDS {
let schema = schema_of(command);
let required: Vec<String> = schema
.get("required")
.and_then(Json::array)
.unwrap_or_default()
.iter()
.filter_map(|name| name.text().map(str::to_string))
.collect();
let expected: Vec<String> = command
.params
.iter()
.filter(|param| param.required)
.map(|param| param.name.to_string())
.collect();
assert_eq!(
required, expected,
"{} declares the wrong required set",
command.name
);
assert_eq!(schema.get("additionalProperties"), Some(&Json::Bool(false)));
}
}
#[test]
fn tool_arguments_are_checked_against_the_command_schema() {
for arguments in [
"{\"sql\":\"SELECT 1\",\"limit\":\"one\"}",
"{\"sql\":\"SELECT 1\",\"limti\":1}",
"{\"sql\":\"SELECT 1\",\"output\":\"yaml\"}",
] {
let answer = handle(
&mut context(),
&format!("{{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{{\"name\":\"inillucent_query\",\"arguments\":{arguments}}}}}"),
)
.unwrap_or_default();
assert!(answer.contains("\"code\":-32602"), "{answer}");
}
let query_schema = schema_of(command::find("query").unwrap_or(&command::COMMANDS[0]));
let output = query_schema
.get("properties")
.and_then(|value| value.get("output"))
.unwrap_or(&Json::Null)
.write();
assert!(output.contains("\"enum\":[\"text\",\"json\"]"), "{output}");
}
#[test]
fn response_budget_errors_keep_the_request_id() {
let response = format!(
"{{\"jsonrpc\":\"2.0\",\"id\":77,\"result\":\"{}\"}}",
"x".repeat(MAX_RESPONSE_BYTES)
);
let replacement = enforce_response_budget(response);
assert!(replacement.contains("\"id\":77"), "{replacement}");
assert!(replacement.contains("\"code\":-32603"), "{replacement}");
}
#[test]
fn read_only_refuses_a_write() {
let mut held = Context::open(":memory:", OpenMode::ReadOnly, None).expect("opens");
let answer = handle(
&mut held,
"{\"jsonrpc\":\"2.0\",\"id\":7,\"method\":\"tools/call\",\"params\":{\
\"name\":\"inillucent_exec\",\"arguments\":{\"sql\":\"CREATE TABLE t (a)\"}}}",
)
.unwrap_or_default();
assert!(answer.contains("\"isError\":true"), "{answer}");
assert!(answer.contains("read only"), "{answer}");
}
}