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};
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>,
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> {
#[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,
}
}
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)
}
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"))?;
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(())
}
#[must_use]
pub const fn rail(&self) -> &R {
&self.rail
}
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 {
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,
}),
}
}
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)
}
}