use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::borrow::Cow;
use std::cell::RefCell;
use std::collections::{BTreeMap, HashMap};
use rmcp::handler::server::common::schema_for_type;
use rmcp::model::{ElicitResult, ElicitationAction};
use schemars::JsonSchema;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use tokio_util::sync::CancellationToken;
use crate::seams::{Feature, Space};
pub type JsonObject = Map<String, Value>;
#[derive(Clone)]
pub struct Ctx {
space: Arc<Space>,
cancel: Option<CancellationToken>,
progress: Option<Arc<dyn Progress>>,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Ask {
pub key: String,
pub message: String,
pub form: Form,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Form {
Fields(JsonObject),
Url(String),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct NeedsInput;
impl std::fmt::Display for NeedsInput {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("the operation needs an answer from the user")
}
}
impl std::error::Error for NeedsInput {}
impl From<NeedsInput> for String {
fn from(needs: NeedsInput) -> Self {
needs.to_string()
}
}
pub(crate) struct Input {
answers: Option<BTreeMap<String, Value>>,
effects: Effects,
}
#[derive(Default)]
pub(crate) struct Effects {
pub asked: Vec<Ask>,
pub updated: Vec<String>,
pub attached: Vec<Content>,
}
tokio::task_local! {
static INPUT: RefCell<Input>;
}
pub(crate) async fn with_input<F: Future>(
answers: Option<BTreeMap<String, Value>>,
work: F,
) -> (F::Output, Effects) {
let input = RefCell::new(Input {
answers,
effects: Effects::default(),
});
INPUT
.scope(input, async {
let output = work.await;
let effects = INPUT.with(|input| std::mem::take(&mut input.borrow_mut().effects));
(output, effects)
})
.await
}
pub(crate) trait Progress: Send + Sync {
fn report(
&self,
done: f64,
total: Option<f64>,
message: Option<String>,
) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;
}
impl Ctx {
pub(crate) fn new(
space: Arc<Space>,
cancel: Option<CancellationToken>,
progress: Option<Arc<dyn Progress>>,
) -> Self {
Self {
space,
cancel,
progress,
}
}
pub fn space(&self) -> &Space {
&self.space
}
pub(crate) fn shared_space(&self) -> Arc<Space> {
self.space.clone()
}
pub fn is_cancelled(&self) -> bool {
self.cancel
.as_ref()
.is_some_and(CancellationToken::is_cancelled)
}
pub async fn cancelled(&self) {
match &self.cancel {
Some(cancel) => cancel.cancelled().await,
None => std::future::pending().await,
}
}
pub fn ask<T: DeserializeOwned + JsonSchema + 'static>(
&self,
key: &str,
message: &str,
) -> Result<Option<T>, NeedsInput> {
let fields = || Form::Fields(JsonObject::clone(&schema_for_type::<T>()));
match self.answer(key, message, fields)? {
Some(content) => match serde_json::from_value(content) {
Ok(answer) => Ok(Some(answer)),
Err(_) => {
self.record(key, message, fields());
Err(NeedsInput)
}
},
None => Ok(None),
}
}
pub fn ask_url(&self, key: &str, message: &str, url: &str) -> Result<bool, NeedsInput> {
let answered = self.answer(key, message, || Form::Url(url.to_owned()))?;
Ok(answered.is_some())
}
fn answer(
&self,
key: &str,
message: &str,
form: impl FnOnce() -> Form,
) -> Result<Option<Value>, NeedsInput> {
let given = INPUT
.try_with(|input| {
let input = input.borrow();
let given = input.answers.as_ref()?.get(key)?;
serde_json::from_value::<ElicitResult>(given.clone()).ok()
})
.map_err(|_| NeedsInput)?;
match given {
Some(result) => Ok(match result.action {
ElicitationAction::Accept => Some(result.content.unwrap_or(Value::Null)),
_ => None,
}),
None => {
self.record(key, message, form());
Err(NeedsInput)
}
}
}
fn record(&self, key: &str, message: &str, form: Form) {
let _ = INPUT.try_with(|input| {
input.borrow_mut().effects.asked.push(Ask {
key: key.to_owned(),
message: message.to_owned(),
form,
});
});
}
pub fn attach(&self, content: Content) {
let _ = INPUT.try_with(|input| input.borrow_mut().effects.attached.push(content));
}
pub fn updated(&self, uri: impl Into<String>) {
let _ = INPUT.try_with(|input| input.borrow_mut().effects.updated.push(uri.into()));
}
pub async fn progress(&self, done: f64, total: Option<f64>, message: Option<&str>) {
if let Some(progress) = &self.progress {
progress
.report(done, total, message.map(str::to_owned))
.await;
}
}
}
pub(crate) type Handler<M, T> = Box<
dyn Fn(Arc<M>, Ctx, Value) -> Pin<Box<dyn Future<Output = Result<T, String>> + Send>>
+ Send
+ Sync,
>;
fn erase<M, A, R, E, F, Fut, T>(handler: F, finish: fn(R) -> Result<T, String>) -> Handler<M, T>
where
M: Send + Sync + 'static,
A: DeserializeOwned,
R: 'static,
E: std::fmt::Display,
F: Fn(Arc<M>, Ctx, A) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<R, E>> + Send + 'static,
T: 'static,
{
Box::new(move |module, ctx, arguments| {
let started = serde_json::from_value(arguments)
.map(|arguments| handler(module, ctx, arguments))
.map_err(|error| format!("invalid arguments: {error}"));
Box::pin(async move {
let result = started?.await.map_err(|error| error.to_string())?;
finish(result)
})
})
}
#[derive(Clone, Debug)]
pub struct Tool {
pub name: &'static str,
pub feature: Feature,
pub description: Option<String>,
pub input: Arc<JsonObject>,
pub output: Option<Arc<JsonObject>>,
pub long: bool,
pub ui: Option<&'static str>,
pub app_only: bool,
pub hints: Option<Hints>,
pub looks: Looks,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Serialize)]
pub struct Looks {
pub title: Option<&'static str>,
pub icons: &'static [Icon],
pub mime: Option<&'static str>,
pub audience: &'static [Role],
pub priority: Option<f32>,
pub about: Option<&'static str>,
pub website: Option<&'static str>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub struct Icon {
pub src: &'static str,
pub mime: Option<&'static str>,
pub sizes: &'static [&'static str],
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub struct Hints {
pub read_only: bool,
pub destructive: bool,
pub idempotent: bool,
pub open_world: bool,
}
impl Default for Hints {
fn default() -> Self {
Self {
read_only: false,
destructive: true,
idempotent: false,
open_world: true,
}
}
}
#[derive(Clone, Debug)]
pub struct Resource {
pub uri: &'static str,
pub name: &'static str,
pub feature: Feature,
pub description: Option<String>,
pub looks: Looks,
}
impl Resource {
pub fn is_template(&self) -> bool {
self.uri.contains('{')
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Body {
Text {
mime: Cow<'static, str>,
text: String,
},
Bytes {
mime: Cow<'static, str>,
#[serde(with = "base64_bytes")]
bytes: Vec<u8>,
},
}
impl Body {
pub fn text(mime: impl Into<Cow<'static, str>>, text: impl Into<String>) -> Self {
Self::Text {
mime: mime.into(),
text: text.into(),
}
}
pub fn bytes(mime: impl Into<Cow<'static, str>>, bytes: impl Into<Vec<u8>>) -> Self {
Self::Bytes {
mime: mime.into(),
bytes: bytes.into(),
}
}
pub fn json<T: Serialize>(value: &T) -> Result<Self, String> {
let text = serde_json::to_string(value).map_err(|error| error.to_string())?;
Ok(Self::text("application/json", text))
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Content {
Text(String),
Image {
mime: Cow<'static, str>,
#[serde(with = "base64_bytes")]
data: Vec<u8>,
},
Audio {
mime: Cow<'static, str>,
#[serde(with = "base64_bytes")]
data: Vec<u8>,
},
Link {
uri: String,
name: String,
mime: Option<Cow<'static, str>>,
},
Resource {
uri: String,
body: Body,
},
}
mod base64_bytes {
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use serde::{Deserialize, Deserializer, Serializer};
pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_str(&STANDARD.encode(bytes))
}
pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
let text = String::deserialize(deserializer)?;
STANDARD.decode(text).map_err(serde::de::Error::custom)
}
}
pub(crate) struct Readable<M> {
pub resource: Resource,
pub read: Handler<M, Body>,
}
#[derive(Clone, Debug)]
pub struct Prompt {
pub name: &'static str,
pub feature: Feature,
pub description: Option<String>,
pub arguments: Vec<PromptArgument>,
pub schema: Arc<JsonObject>,
pub looks: Looks,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct PromptArgument {
pub name: String,
pub description: Option<String>,
pub required: bool,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Role {
User,
Assistant,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Message {
pub role: Role,
pub content: Content,
}
impl Message {
pub fn text(role: Role, text: impl Into<String>) -> Self {
Self {
role,
content: Content::Text(text.into()),
}
}
}
pub(crate) struct Promptable<M> {
pub prompt: Prompt,
pub get: Handler<M, Vec<Message>>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Completing {
pub typed: String,
#[serde(default)]
pub given: HashMap<String, String>,
}
pub(crate) struct Completer<M> {
pub target: &'static str,
pub argument: &'static str,
pub complete: Handler<M, Vec<String>>,
}
pub(crate) const APP_MIME: &str = "text/html;profile=mcp-app";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct App {
pub uri: &'static str,
pub html: &'static str,
}
pub(crate) struct Op<M> {
pub tool: Tool,
pub call: Handler<M, Value>,
}
pub struct Ops<M> {
pub(crate) list: Vec<Op<M>>,
pub(crate) resources: Vec<Readable<M>>,
pub(crate) prompts: Vec<Promptable<M>>,
pub(crate) completers: Vec<Completer<M>>,
pub(crate) apps: Vec<App>,
pub(crate) instructions: Option<&'static str>,
pub(crate) server: Looks,
}
impl<M: Send + Sync + 'static> Ops<M> {
pub(crate) fn new() -> Self {
Self {
list: Vec::new(),
resources: Vec::new(),
prompts: Vec::new(),
completers: Vec::new(),
apps: Vec::new(),
instructions: None,
server: Looks::default(),
}
}
pub fn tool<A, R, E, F, Fut>(
&mut self,
name: &'static str,
feature: Feature,
handler: F,
) -> &mut Self
where
A: DeserializeOwned + JsonSchema + 'static,
R: Serialize + JsonSchema + 'static,
E: std::fmt::Display,
F: Fn(Arc<M>, Ctx, A) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<R, E>> + Send + 'static,
{
self.add(name, feature, false, handler)
}
pub fn task<A, R, E, F, Fut>(
&mut self,
name: &'static str,
feature: Feature,
handler: F,
) -> &mut Self
where
A: DeserializeOwned + JsonSchema + 'static,
R: Serialize + JsonSchema + 'static,
E: std::fmt::Display,
F: Fn(Arc<M>, Ctx, A) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<R, E>> + Send + 'static,
{
self.add(name, feature, true, handler)
}
pub fn app(&mut self, tool: &str, uri: &'static str, html: &'static str) -> &mut Self {
assert!(
uri.starts_with("ui://"),
"the app {uri} must be a ui:// URI"
);
self.registered(tool).ui = Some(uri);
match self.apps.iter().find(|app| app.uri == uri) {
Some(app) => assert_eq!(app.html, html, "{uri} already serves another page"),
None => self.apps.push(App { uri, html }),
}
self
}
pub fn resource<A, E, F, Fut>(
&mut self,
uri: &'static str,
name: &'static str,
feature: Feature,
handler: F,
) -> &mut Self
where
A: DeserializeOwned + JsonSchema + 'static,
E: std::fmt::Display,
F: Fn(Arc<M>, Ctx, A) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Body, E>> + Send + 'static,
{
assert!(
self.resources.iter().all(|r| r.resource.uri != uri),
"resource {uri} is registered twice"
);
let description = take_text(&mut JsonObject::clone(&schema_for_type::<A>()));
self.resources.push(Readable {
resource: Resource {
uri,
name,
feature,
description,
looks: Looks::default(),
},
read: erase(handler, Ok),
});
self
}
pub fn prompt<A, E, F, Fut>(
&mut self,
name: &'static str,
feature: Feature,
handler: F,
) -> &mut Self
where
A: DeserializeOwned + JsonSchema + 'static,
E: std::fmt::Display,
F: Fn(Arc<M>, Ctx, A) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Vec<Message>, E>> + Send + 'static,
{
assert!(
self.prompts.iter().all(|p| p.prompt.name != name),
"prompt {name} is registered twice"
);
let mut schema = JsonObject::clone(&schema_for_type::<A>());
assert!(
is_object(&schema),
"the arguments of {name} must be a struct"
);
let description = take_text(&mut schema);
let arguments = arguments(&schema);
self.prompts.push(Promptable {
prompt: Prompt {
name,
feature,
description,
arguments,
schema: Arc::new(schema),
looks: Looks::default(),
},
get: erase(handler, Ok),
});
self
}
pub fn complete<E, F, Fut>(
&mut self,
target: &'static str,
argument: &'static str,
handler: F,
) -> &mut Self
where
E: std::fmt::Display,
F: Fn(Arc<M>, Ctx, Completing) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<Vec<String>, E>> + Send + 'static,
{
let known = self.prompts.iter().any(|p| p.prompt.name == target)
|| self.resources.iter().any(|r| r.resource.uri == target);
assert!(known, "register {target} before completing it");
self.completers.push(Completer {
target,
argument,
complete: erase(handler, Ok),
});
self
}
pub fn app_only(&mut self, tool: &str) -> &mut Self {
self.registered(tool).app_only = true;
self
}
pub fn hints(&mut self, tool: &str, hints: Hints) -> &mut Self {
self.registered(tool).hints = Some(hints);
self
}
pub fn looks(&mut self, target: &str, looks: Looks) -> &mut Self {
if let Some(op) = self.list.iter_mut().find(|op| op.tool.name == target) {
op.tool.looks = looks;
} else if let Some(p) = self.prompts.iter_mut().find(|p| p.prompt.name == target) {
p.prompt.looks = looks;
} else if let Some(r) = self.resources.iter_mut().find(|r| r.resource.uri == target) {
r.resource.looks = looks;
} else {
panic!("register {target} before describing it");
}
self
}
pub fn server(&mut self, looks: Looks) -> &mut Self {
self.server = looks;
self
}
pub fn instructions(&mut self, text: &'static str) -> &mut Self {
self.instructions = Some(text);
self
}
fn registered(&mut self, name: &str) -> &mut Tool {
self.list
.iter_mut()
.map(|op| &mut op.tool)
.find(|tool| tool.name == name)
.unwrap_or_else(|| panic!("register {name} before describing it"))
}
fn add<A, R, E, F, Fut>(
&mut self,
name: &'static str,
feature: Feature,
long: bool,
handler: F,
) -> &mut Self
where
A: DeserializeOwned + JsonSchema + 'static,
R: Serialize + JsonSchema + 'static,
E: std::fmt::Display,
F: Fn(Arc<M>, Ctx, A) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<R, E>> + Send + 'static,
{
assert!(
self.list.iter().all(|op| op.tool.name != name),
"operation {name} is registered twice"
);
let mut input = JsonObject::clone(&schema_for_type::<A>());
assert!(
is_object(&input),
"the arguments of {name} must be a struct"
);
let description = take_text(&mut input);
let output = Some(JsonObject::clone(&schema_for_type::<R>()))
.filter(is_object)
.map(|mut output| {
take_text(&mut output);
Arc::new(output)
});
let call = erase(handler, |result: R| {
serde_json::to_value(result).map_err(|error| error.to_string())
});
self.list.push(Op {
tool: Tool {
name,
feature,
description,
input: Arc::new(input),
output,
long,
ui: None,
app_only: false,
hints: None,
looks: Looks::default(),
},
call,
});
self
}
}
fn arguments(schema: &JsonObject) -> Vec<PromptArgument> {
let required = |name: &str| {
schema
.get("required")
.and_then(Value::as_array)
.is_some_and(|required| required.iter().any(|r| r == name))
};
let Some(Value::Object(properties)) = schema.get("properties") else {
return Vec::new();
};
properties
.iter()
.map(|(name, property)| PromptArgument {
name: name.clone(),
description: property
.get("description")
.and_then(Value::as_str)
.map(str::to_owned),
required: required(name),
})
.collect()
}
fn is_object(schema: &JsonObject) -> bool {
schema.get("type").and_then(Value::as_str) == Some("object")
}
fn take_text(schema: &mut JsonObject) -> Option<String> {
schema.remove("title");
match schema.remove("description") {
Some(Value::String(text)) => Some(text),
_ => None,
}
}