liminal_cli/mcp/

server.rs

//! The negotiated tools and the session that owns their subscriptions.

use super::framing::{self, INVALID_PARAMS, METHOD_NOT_FOUND};
use super::negotiate::{Era, Refusal, Session, shaped};
use super::tools::{TOOLS_TTL_MS, ToolResult, object, string, tool_table};
use crate::network::{Rail, call, directory, flow};
use serde_json::{Map, Value, json};

/// The ticket's server: one rail, one session and its service subscriptions.
pub struct Server<R> {
    rail: R,
    session: Session,
    pub(super) frame_bytes: usize,
    pub(super) subscriptions: std::sync::Arc<super::subscriptions::Subscriptions>,
    pub(super) warning_only: std::sync::Arc<std::sync::atomic::AtomicBool>,
    pub(super) listens: std::sync::Arc<super::listen::Listens>,
    pub(super) listen_config: super::listen_config::ListenConfig,
    pub(super) channel_source: Option<super::channel_source::Source>,
    pub(super) initial_listens: Vec<String>,
    /// The agent's own inbox when no `[listen]` entry names it: a station
    /// the exchange does not hold leaves the ticket's tools serving.
    pub(super) own_inbox: Option<String>,
}

impl<R> std::fmt::Debug for Server<R> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Server")
            .field("session", &self.session)
            .field("frame_bytes", &self.frame_bytes)
            .field("listens", &self.listens)
            .field("channel_source", &self.channel_source)
            .finish_non_exhaustive()
    }
}

impl<R: Rail> Server<R> {
    /// A server over `rail`, reading lines of at most `frame_bytes`.
    #[must_use]
    pub fn new(rail: R, frame_bytes: usize) -> Self {
        Self {
            rail,
            session: Session::new(),
            frame_bytes,
            subscriptions: std::sync::Arc::default(),
            warning_only: std::sync::Arc::default(),
            listens: std::sync::Arc::default(),
            listen_config: super::listen_config::ListenConfig::default(),
            channel_source: None,
            initial_listens: Vec::new(),
            own_inbox: None,
        }
    }

    /// A ticket on the validated seat, with initial listens opened by `serve`.
    /// No socket or pass request occurs in this constructor.
    ///
    /// # Errors
    /// Returns a named pass-source or listen-configuration refusal.
    pub fn configured(rail: R, config: &crate::config::Config) -> Result<Self, String> {
        config.listen.validate(config.agent.as_deref())?;
        let source = super::channel_source::Source::new(config)?;
        let mut server = Self::new(rail, config.limits.frame_bytes);
        server.listen_config = config.listen.clone();
        server.initial_listens = config.listen.initial_subjects(config.agent.as_deref());
        server.own_inbox = config
            .agent
            .as_deref()
            .map(|agent| format!("agents.{agent}.inbox"))
            .filter(|inbox| !config.listen.subjects.contains(inbox));
        server.channel_source = Some(source);
        Ok(server)
    }

    /// Active channel subjects and all ticket channel threads, respectively.
    /// Each subject owns two streams, a tap reader and a join worker.
    ///
    /// # Errors
    /// Returns a poisoned listen-state lock.
    pub fn listen_counts(&self) -> Result<(usize, usize), String> {
        Ok((
            self.listens.count()?,
            self.listens
                .receivers
                .load(std::sync::atomic::Ordering::SeqCst),
        ))
    }

    pub(super) fn start_listens(&self) -> std::io::Result<()> {
        let Some(source) = &self.channel_source else {
            return Ok(());
        };
        for subject in &self.initial_listens {
            let answer = self.listens.listen(source, &self.listen_config, subject);
            if answer.is_error {
                let words = answer.structured["error"]
                    .as_str()
                    .ok_or_else(|| std::io::Error::other("the listen refusal has no words"))?;
                // The implicit inbox is refused in the server's words and the
                // tools keep serving; a configured subject's refusal is fatal.
                if self.own_inbox.as_deref() == Some(subject.as_str()) {
                    eprintln!("liminal mcp: {subject} is not listened: {words}");
                    continue;
                }
                let closed = self.listens.close_all();
                return Err(std::io::Error::other(match closed {
                    Ok(()) => words.to_owned(),
                    Err(close) => format!("{words}; {close}"),
                }));
            }
        }
        Ok(())
    }

    /// The rail behind the tools.
    #[must_use]
    pub const fn rail(&self) -> &R {
        &self.rail
    }

    /// One request's answer.
    pub fn answer(&mut self, id: &Value, method: &str, params: &Map<String, Value>) -> Value {
        match self.respond(method, params) {
            Ok(result) => framing::result(id, &result),
            Err(refusal) => framing::error(id, refusal.code, &refusal.message, refusal.data),
        }
    }

    fn respond(&mut self, method: &str, params: &Map<String, Value>) -> Result<Value, Refusal> {
        match method {
            "initialize" => return self.session.initialize(params),
            "server/discover" => {
                self.session.era_of(params)?;
                return Ok(Session::discover());
            }
            _ => {}
        }
        let era = self.session.era_of(params)?;
        match method {
            "ping" => Ok(shaped(era, Map::new())),
            "logging/setLevel" => {
                super::notifications::set_level(params, &self.warning_only)?;
                Ok(shaped(era, Map::new()))
            }
            "tools/list" => {
                let mut result = Map::new();
                result.insert("tools".to_owned(), tool_table());
                if era == Era::Modern {
                    // 2026-07-28 list results carry their cache terms. The
                    // table is the same for every pass (what a call may reach
                    // is decided per call), so any client may share it; a
                    // minute lets a replaced binary's tools show promptly.
                    result.insert("ttlMs".to_owned(), json!(TOOLS_TTL_MS));
                    result.insert("cacheScope".to_owned(), json!("public"));
                }
                Ok(shaped(era, result))
            }
            "tools/call" => {
                let name = params
                    .get("name")
                    .and_then(Value::as_str)
                    .ok_or_else(|| Refusal {
                        code: INVALID_PARAMS,
                        message: "the request is missing name".to_owned(),
                        data: None,
                    })?;
                let arguments = params
                    .get("arguments")
                    .and_then(Value::as_object)
                    .cloned()
                    .unwrap_or_default();
                let outcome = self.tool(name, &arguments)?;
                Ok(outcome.into_result(era))
            }
            other => Err(Refusal {
                code: METHOD_NOT_FOUND,
                message: format!("the ticket has no method {other}"),
                data: None,
            }),
        }
    }

    /// One tool's outcome; an unknown tool is method-not-found.
    ///
    /// # Errors
    /// Returns the method-not-found or invalid-params refusal.
    pub fn tool(&self, name: &str, arguments: &Map<String, Value>) -> Result<ToolResult, Refusal> {
        if name == "send" {
            return super::channel_send::send(self, arguments);
        }
        if matches!(name, "listen" | "unlisten") {
            let subject = super::listen::argument(arguments)?;
            if name == "unlisten" {
                return Ok(self.listens.unlisten(subject));
            }
            let Some(source) = &self.channel_source else {
                return Ok(ToolResult::failed(
                    "the ticket has no seat configuration for channel tools",
                    json!({"subject":subject}),
                ));
            };
            return Ok(self.listens.listen(source, &self.listen_config, subject));
        }
        if matches!(name, "subscribe" | "unsubscribe") {
            let service = super::subscriptions::service_argument(arguments)?;
            return Ok(self.subscriptions.tool(&self.rail, name, service));
        }
        if let Some(outcome) = crate::network::lifecycle::tool(&self.rail, name, arguments) {
            return Ok(outcome);
        }
        if let Some(outcome) = super::plan_tools::tool(&self.rail, name, arguments)
            .or_else(|| super::template_tools::tool(&self.rail, name, arguments))
        {
            return Ok(outcome);
        }
        let outcome = match name {
            "list" => directory::list(&self.rail),
            "describe" => {
                let subject = string(arguments, "subject")?;
                directory::describe(&self.rail, subject)
            }
            "call" => {
                let subject = string(arguments, "subject")?;
                let input = object(arguments, "input")?;
                let reply = arguments
                    .get("reply")
                    .and_then(Value::as_bool)
                    .unwrap_or(true);
                let pass_for = match arguments.get("pass_for") {
                    None => None,
                    Some(Value::String(audience)) => Some(audience.as_str()),
                    Some(_) => {
                        return Err(Refusal {
                            code: INVALID_PARAMS,
                            message: "the tool call's pass_for is not a string".to_owned(),
                            data: None,
                        });
                    }
                };
                call::call(&self.rail, subject, input, reply, pass_for)
            }
            "flow" => {
                let steps = arguments
                    .get("steps")
                    .and_then(Value::as_array)
                    .ok_or_else(|| Refusal {
                        code: INVALID_PARAMS,
                        message: "the flow is missing steps".to_owned(),
                        data: None,
                    })?;
                flow::flow(&self.rail, steps)
            }
            other => {
                return Err(Refusal {
                    code: METHOD_NOT_FOUND,
                    message: format!("the ticket has no tool {other}"),
                    data: None,
                });
            }
        };
        Ok(outcome)
    }
}