use std::sync::Arc;
use rmcp::{
handler::server::wrapper::Parameters,
model::{Implementation, ServerCapabilities, ServerInfo},
tool, tool_handler, tool_router, ErrorData as McpError, ServerHandler, ServiceExt,
};
use serde_json::{json, Value};
use khive_db::ConnectionPool;
use khive_request::{parse_request, ArgValue, DslError, ExecutionMode, ParsedOp};
use khive_runtime::{
present, render_format, KhiveRuntime, Namespace, OutputFormat, PackLoadError, PackRegistry,
PresentationMode, RuntimeConfig, RuntimeError, VerbPresentationPolicy, VerbRegistry,
VerbRegistryBuilder,
};
use khive_storage::EdgeRelation;
use crate::coordinator::CoordinatorService;
use crate::tools::request::RequestParams;
pub fn compute_config_id(
config: &RuntimeConfig,
khive_cfg: Option<&khive_runtime::KhiveConfig>,
) -> String {
let mut packs = config.packs.clone();
packs.sort();
let db = config
.db_path
.as_ref()
.map(|p| p.display().to_string())
.unwrap_or_else(|| ":memory:".to_string());
let primary = config
.embedding_model
.as_ref()
.map(|m| format!("{m:?}"))
.unwrap_or_else(|| "none".to_string());
let mut extra: Vec<String> = config
.additional_embedding_models
.iter()
.map(|m| format!("{m:?}"))
.collect();
extra.sort();
let mut visible: Vec<String> = config
.visible_namespaces
.iter()
.map(|ns| ns.as_str().to_owned())
.collect();
visible.sort();
visible.dedup();
let mut outbound: Vec<String> = config
.allowed_outbound_namespaces
.iter()
.map(|ns| ns.as_str().to_owned())
.collect();
outbound.sort();
outbound.dedup();
let base = format!(
"packs=[{}];db={};embed={};extra=[{}];backend={:?};visible=[{}];outbound=[{}]",
packs.join(","),
db,
primary,
extra.join(","),
config.backend_id,
visible.join(","),
outbound.join(","),
);
let topology = khive_cfg
.filter(|cfg| !cfg.backends.is_empty())
.map(|cfg| {
let mut backend_entries: Vec<String> = cfg
.backends
.iter()
.map(|b| {
let path = b
.path
.as_ref()
.map(|p| p.display().to_string())
.unwrap_or_else(|| ":memory:".to_string());
format!("{}:{:?}:{}", b.name, b.kind, path)
})
.collect();
backend_entries.sort();
let mut pack_entries: Vec<String> = cfg
.packs
.iter()
.map(|(pack, pc)| format!("{}={}", pack, pc.backend))
.collect();
pack_entries.sort();
format!(
";backends=[{}];pack_backends=[{}]",
backend_entries.join(","),
pack_entries.join(","),
)
})
.unwrap_or_default();
format!("{base}{topology}")
}
fn build_verb_catalog(verbs: impl IntoIterator<Item = (String, String, String)>) -> String {
let mut by_verb: std::collections::BTreeMap<String, Vec<(String, String)>> =
std::collections::BTreeMap::new();
for (pack_name, verb_name, description) in verbs {
by_verb
.entry(verb_name)
.or_default()
.push((pack_name, description));
}
let mut out = String::new();
for (name, pack_descs) in &by_verb {
if pack_descs.len() > 1 {
let packs: Vec<&str> = pack_descs.iter().map(|(p, _)| p.as_str()).collect();
tracing::warn!(
verb = %name,
packs = ?packs,
"verb registered by multiple packs; all descriptions included in catalog"
);
}
out.push_str(" ");
out.push_str(name);
out.push_str(" — ");
if pack_descs.len() == 1 {
out.push_str(&pack_descs[0].1);
} else {
for (i, (pack, desc)) in pack_descs.iter().enumerate() {
if i > 0 {
out.push_str("\n ");
}
out.push('[');
out.push_str(pack);
out.push_str("] ");
out.push_str(desc);
}
}
out.push('\n');
}
out
}
#[derive(Clone)]
pub struct KhiveMcpServer {
registry: VerbRegistry,
default_namespace: String,
config_id: String,
coordinator: Option<Arc<dyn CoordinatorService>>,
pool: Option<Arc<ConnectionPool>>,
default_output_format: OutputFormat,
}
pub enum PackRegFailure {
UnknownPack(String),
MissingDependency { pack: String, dep: String },
Registry(khive_runtime::RuntimeError),
}
pub struct PackRegError {
pub failure: PackRegFailure,
pub runtime: KhiveRuntime,
}
impl std::fmt::Debug for PackRegError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut dbg = f.debug_struct("PackRegError");
match &self.failure {
PackRegFailure::UnknownPack(unknown) => dbg.field("unknown", unknown),
PackRegFailure::MissingDependency { pack, dep } => {
dbg.field("pack", pack).field("missing_dep", dep)
}
PackRegFailure::Registry(source) => dbg.field("source", source),
}
.finish_non_exhaustive()
}
}
impl std::fmt::Display for PackRegError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match &self.failure {
PackRegFailure::UnknownPack(unknown) => write!(
f,
"unknown pack name {:?} — built-in packs: {}",
unknown,
builtin_pack_names().join(", ")
),
PackRegFailure::MissingDependency { pack, dep } => write!(
f,
"pack {pack:?} requires {dep:?}, which is not in the requested pack list; \
add --pack {dep} before --pack {pack}"
),
PackRegFailure::Registry(source) => write!(f, "pack registry build failed: {source}"),
}
}
}
impl std::error::Error for PackRegError {}
pub fn builtin_pack_names() -> Vec<&'static str> {
PackRegistry::discovered_names()
}
impl KhiveMcpServer {
#[allow(clippy::result_large_err)]
pub fn new(runtime: KhiveRuntime) -> Result<Self, PackRegError> {
let packs: Vec<String> = runtime.config().packs.clone();
Self::with_packs(runtime, &packs)
}
#[allow(clippy::result_large_err)]
pub fn with_packs(runtime: KhiveRuntime, packs: &[String]) -> Result<Self, PackRegError> {
let gate = runtime.config().gate.clone();
let default_namespace = runtime.config().default_namespace.clone();
let config_id = compute_config_id(runtime.config(), None);
let visible_namespaces = runtime.config().visible_namespaces.clone();
let actor_id = runtime.config().actor_id.clone();
let mut builder = VerbRegistryBuilder::new();
builder.with_gate(gate);
builder.with_default_namespace(default_namespace.as_str());
builder.with_visible_namespaces(visible_namespaces);
builder.with_actor_id(actor_id);
if let Ok(tok) = runtime.authorize(khive_runtime::Namespace::local()) {
if let Ok(event_store) = runtime.events(&tok) {
builder.with_event_store(event_store);
}
}
if let Err(load_err) = PackRegistry::register_packs(packs, runtime.clone(), &mut builder) {
let failure = match load_err {
PackLoadError::UnknownPack(name) => PackRegFailure::UnknownPack(name),
PackLoadError::MissingDependency { pack, dep } => {
PackRegFailure::MissingDependency { pack, dep }
}
};
return Err(PackRegError { failure, runtime });
}
let registry = builder.build().map_err(|source| PackRegError {
failure: PackRegFailure::Registry(source),
runtime: runtime.clone(),
})?;
runtime.install_edge_rules(registry.all_edge_rules());
registry.call_register_embedders(&runtime);
registry.call_register_entity_type_validators(&runtime);
registry.apply_schema_plans(runtime.backend());
let pool = if runtime.backend().is_file_backed() {
Some(runtime.backend().pool_arc())
} else {
None
};
Ok(Self {
registry,
default_namespace: default_namespace.as_str().to_string(),
config_id,
coordinator: None,
pool,
default_output_format: OutputFormat::Json,
})
}
#[doc(hidden)]
pub fn from_registry(registry: VerbRegistry) -> Self {
Self {
registry,
default_namespace: "local".to_string(),
config_id: "registry-only".to_string(),
coordinator: None,
pool: None,
default_output_format: OutputFormat::Json,
}
}
pub fn from_registry_with_meta(
registry: VerbRegistry,
default_namespace: &str,
config_id: &str,
) -> Self {
Self {
registry,
default_namespace: default_namespace.to_string(),
config_id: config_id.to_string(),
coordinator: None,
pool: None,
default_output_format: OutputFormat::Json,
}
}
pub fn with_default_output_format(mut self, fmt: OutputFormat) -> Self {
self.default_output_format = fmt;
self
}
pub fn with_coordinator(mut self, coordinator: Arc<dyn CoordinatorService>) -> Self {
self.coordinator = Some(coordinator);
self
}
pub fn with_pool(mut self, pool: Arc<ConnectionPool>) -> Self {
self.pool = Some(pool);
self
}
#[cfg(feature = "channel-email")]
pub(crate) fn verb_registry_clone(&self) -> VerbRegistry {
self.registry.clone()
}
async fn dispatch_via_coordinator(
&self,
tool: &str,
args_value: &Value,
) -> Option<Result<Value, (String, Value)>> {
let coord = self.coordinator.as_ref()?;
if coord.is_single_backend() {
return None;
}
dispatch_via_coordinator_inner(coord.as_ref(), tool, args_value, &self.default_namespace)
.await
}
pub fn default_namespace(&self) -> &str {
&self.default_namespace
}
pub fn config_id(&self) -> &str {
&self.config_id
}
pub fn pool(&self) -> Option<Arc<ConnectionPool>> {
self.pool.clone()
}
pub async fn warm_all(&self) {
self.registry.call_warm_all().await;
}
pub async fn serve_stdio(self) -> anyhow::Result<()> {
use rmcp::transport::stdio;
let service = self.serve(stdio()).await?;
service.waiting().await?;
Ok(())
}
fn verb_catalog(&self) -> String {
let verbs = self
.registry
.all_verbs_with_names()
.into_iter()
.map(|(pack, v)| (pack.to_owned(), v.name.to_owned(), v.description.to_owned()));
build_verb_catalog(verbs)
}
async fn dispatch_op(
&self,
op: ParsedOp,
prev_result: Option<&Value>,
from_wire: bool,
) -> Result<Value, (String, Value)> {
let ParsedOp { tool, args } = op;
let mut resolved: serde_json::Map<String, Value> = serde_json::Map::new();
for (name, arg_val) in args {
let needs_prev = !matches!(&arg_val, ArgValue::Value(_));
let value = if needs_prev {
let prev = prev_result.ok_or_else(|| {
(
tool.clone(),
json!({
"kind": "substitution_error",
"message": format!(
"argument {name:?}: $prev reference in non-chain context"
)
}),
)
})?;
let resolved_val = arg_val.resolve_all(prev).ok_or_else(|| {
let fields_hint = if let Value::Object(map) = prev {
let mut fields: Vec<&str> =
map.keys().map(String::as_str).collect();
fields.sort_unstable();
format!(
" Available top-level fields: [{}]",
fields.join(", ")
)
} else {
String::new()
};
(
tool.clone(),
json!({
"kind": "substitution_error",
"message": format!(
"argument {name:?}: one or more $prev paths not found in prior result.{fields_hint}"
),
}),
)
})?;
if matches!(&arg_val, ArgValue::PrevRef { path } if path.is_empty()) {
match &resolved_val {
Value::Object(map) => {
let fields: Vec<&str> = map.keys().map(String::as_str).collect();
return Err((
tool.clone(),
json!({
"kind": "substitution_error",
"message": format!(
"argument {name:?}: $prev requires a dotted path \
(e.g. $prev.id) when the prior result is a map. \
Available top-level fields: [{}]",
fields.join(", ")
),
}),
));
}
Value::Array(_) => {
return Err((
tool.clone(),
json!({
"kind": "substitution_error",
"message": format!(
"argument {name:?}: $prev requires a dotted path \
(e.g. $prev.0) when the prior result is an array. \
Use $prev.N to select a specific element."
),
}),
));
}
_ => {}
}
}
resolved_val
} else {
match arg_val {
ArgValue::Value(v) => v,
_ => unreachable!(),
}
};
resolved.insert(name, value);
}
let args_value = Value::Object(resolved);
let is_help = args_value
.get("help")
.and_then(Value::as_bool)
.unwrap_or(false);
if from_wire && !is_help && self.registry.is_subhandler_verb(&tool) {
return Err((
tool.clone(),
json!(format!(
"permission denied for verb {tool:?}: verb '{tool}' is an internal \
subhandler and cannot be invoked via the MCP request surface"
)),
));
}
if let Some(coord_result) = self.dispatch_via_coordinator(&tool, &args_value).await {
return coord_result
.map(|result| json!({ "ok": true, "tool": tool, "result": result }));
}
match self.registry.dispatch(&tool, args_value).await {
Ok(result) => Ok(json!({ "ok": true, "tool": tool, "result": result })),
Err(RuntimeError::Khive(k)) => {
let error_payload = serde_json::to_value(&k)
.unwrap_or_else(|_| json!({ "kind": "internal", "message": k.to_string() }));
Err((tool, error_payload))
}
Err(e) => Err((tool, json!(e.to_string()))),
}
}
async fn run_parsed(
&self,
ops: Vec<ParsedOp>,
mode: ExecutionMode,
presentation: PresentationMode,
presentation_per_op: Option<Vec<Option<PresentationMode>>>,
from_wire: bool,
) -> Value {
let now_unix = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0);
let mode_for_op = |i: usize| -> PresentationMode {
presentation_per_op
.as_ref()
.and_then(|v| v.get(i))
.and_then(|o| *o)
.unwrap_or(presentation)
};
match mode {
ExecutionMode::Single | ExecutionMode::Parallel => {
let conflict_indices: std::collections::HashSet<usize> = {
let mut seen: std::collections::HashMap<String, usize> =
std::collections::HashMap::new();
let mut bad: std::collections::HashSet<usize> =
std::collections::HashSet::new();
for (i, op) in ops.iter().enumerate() {
for key in khive_request::write_keys_for_op_pub(op) {
if let Some(&prior) = seen.get(&key) {
bad.insert(prior);
bad.insert(i);
} else {
seen.insert(key, i);
}
}
}
bad
};
let coordinator: Option<Arc<dyn CoordinatorService>> = self.coordinator.clone();
let default_namespace = self.default_namespace.clone();
let futures = ops.into_iter().enumerate().map(|(i, op)| {
let conflict_with: Option<String> = if conflict_indices.contains(&i) {
Some(format!(
"conflict: writes overlap with another op in this batch (op #{})",
i
))
} else {
None
};
let registry = self.registry.clone();
let coord = coordinator.clone();
let ns_str = default_namespace.clone();
let op_mode = mode_for_op(i);
async move {
let tool = op.tool.clone();
if let Some(msg) = conflict_with {
return json!({ "ok": false, "tool": tool, "error": msg });
}
let effective_mode =
if registry.presentation_policy_for(&tool)
== VerbPresentationPolicy::AlwaysVerbose
{
PresentationMode::Verbose
} else {
op_mode
};
let mut resolved: serde_json::Map<String, Value> =
serde_json::Map::new();
let mut prev_error: Option<Value> = None;
for (name, arg_val) in &op.args {
if matches!(arg_val, ArgValue::Value(_)) {
if let ArgValue::Value(v) = arg_val {
resolved.insert(name.clone(), v.clone());
}
} else {
prev_error = Some(json!({
"ok": false,
"tool": tool,
"error": format!(
"argument {name:?}: $prev reference is only valid in chain (|) mode"
)
}));
break;
}
}
if let Some(err) = prev_error {
return err;
}
let args_value = Value::Object(resolved);
let is_help = args_value
.get("help")
.and_then(Value::as_bool)
.unwrap_or(false);
if from_wire && !is_help && registry.is_subhandler_verb(&tool) {
return json!({
"ok": false,
"tool": tool,
"error": format!(
"permission denied for verb {tool:?}: verb '{tool}' is an \
internal subhandler and cannot be invoked via the MCP \
request surface"
)
});
}
if let Some(active_coord) = coord.as_ref() {
if !active_coord.is_single_backend() {
if let Some(coord_result) = dispatch_via_coordinator_inner(
active_coord.as_ref(),
&tool,
&args_value,
&ns_str,
)
.await
{
return match coord_result {
Ok(result) => {
let presented =
present(result, effective_mode, now_unix);
json!({ "ok": true, "tool": tool, "result": presented })
}
Err((_, error_payload)) => {
json!({ "ok": false, "tool": tool, "error": error_payload })
}
};
}
}
}
match registry.dispatch(&tool, args_value).await {
Ok(result) => {
let presented = present(result, effective_mode, now_unix);
json!({ "ok": true, "tool": tool, "result": presented })
}
Err(RuntimeError::Khive(k)) => {
let error_payload = serde_json::to_value(&k).unwrap_or_else(
|_| json!({ "kind": "internal", "message": k.to_string() }),
);
json!({ "ok": false, "tool": tool, "error": error_payload })
}
Err(e) => json!({ "ok": false, "tool": tool, "error": e.to_string() }),
}
}
});
let results: Vec<Value> = futures::future::join_all(futures).await;
let total = results.len();
let succeeded = results
.iter()
.filter(|r| r.get("ok").and_then(Value::as_bool) == Some(true))
.count();
let failed = total - succeeded;
json!({
"results": results,
"summary": { "total": total, "succeeded": succeeded, "failed": failed, "aborted": 0 },
})
}
ExecutionMode::Chain => {
let total = ops.len();
let mut results: Vec<Value> = Vec::with_capacity(total);
let mut prev_result: Option<Value> = None;
let mut aborted_from: Option<usize> = None;
for (i, op) in ops.into_iter().enumerate() {
if aborted_from.is_some() {
results.push(json!({ "ok": false, "tool": op.tool, "aborted": true }));
continue;
}
let op_mode = mode_for_op(i);
let effective_mode = if self.registry.presentation_policy_for(&op.tool)
== VerbPresentationPolicy::AlwaysVerbose
{
PresentationMode::Verbose
} else {
op_mode
};
match self.dispatch_op(op, prev_result.as_ref(), from_wire).await {
Ok(result_obj) => {
prev_result = result_obj.get("result").cloned();
let presented_obj =
apply_presentation_to_result(result_obj, effective_mode, now_unix);
results.push(presented_obj);
}
Err((tool, error_payload)) => {
results
.push(json!({ "ok": false, "tool": tool, "error": error_payload }));
aborted_from = Some(i + 1);
}
}
}
let succeeded = results
.iter()
.filter(|r| r.get("ok").and_then(Value::as_bool) == Some(true))
.count();
let aborted = results
.iter()
.filter(|r| r.get("aborted").and_then(Value::as_bool) == Some(true))
.count();
let failed = total - succeeded - aborted;
json!({
"results": results,
"summary": { "total": total, "succeeded": succeeded, "failed": failed, "aborted": aborted },
})
}
}
}
}
async fn dispatch_via_coordinator_inner(
coord: &dyn CoordinatorService,
tool: &str,
args_value: &Value,
namespace_str: &str,
) -> Option<Result<Value, (String, Value)>> {
let namespace = Namespace::parse(namespace_str).unwrap_or_else(|_| Namespace::local());
match tool {
"link" => {
if args_value.get("links").is_some() {
return None;
}
let source_str = args_value.get("source_id")?.as_str()?;
let target_str = args_value.get("target_id")?.as_str()?;
let relation_str = args_value.get("relation")?.as_str()?;
let source_id: uuid::Uuid = source_str.parse().ok()?;
let target_id: uuid::Uuid = target_str.parse().ok()?;
let relation: EdgeRelation = relation_str.parse().ok()?;
let weight = args_value
.get("weight")
.and_then(Value::as_f64)
.unwrap_or(1.0);
let metadata = args_value.get("metadata").cloned();
let result = coord
.link(&namespace, source_id, target_id, relation, weight, metadata)
.await;
let tool_name = tool.to_string();
Some(match result {
Ok(coord_result) => {
let edge_val = serde_json::to_value(&coord_result.edge)
.unwrap_or_else(|e| json!({"error": format!("serialize edge: {e}")}));
let mut raw = edge_val;
if relation.is_symmetric() {
if let Some(obj) = raw.as_object_mut() {
obj.insert("source_id".to_string(), json!(source_id.to_string()));
obj.insert("target_id".to_string(), json!(target_id.to_string()));
}
}
Ok(raw)
}
Err(e) => {
let re: RuntimeError = e.into();
match re {
RuntimeError::Khive(k) => {
let error_payload = serde_json::to_value(&k).unwrap_or_else(
|_| json!({"kind": "internal", "message": k.to_string()}),
);
Err((tool_name, error_payload))
}
other => Err((tool_name, json!(other.to_string()))),
}
}
})
}
"search" => {
let kind = args_value.get("kind")?.as_str()?;
let query = args_value.get("query")?.as_str()?;
let limit = args_value
.get("limit")
.and_then(Value::as_u64)
.map(|v| v as u32)
.unwrap_or(10)
.min(100);
let score_floor = args_value
.get("min_score")
.and_then(Value::as_f64)
.unwrap_or(0.0)
.max(0.0);
let kind_filter: Option<&str> = match kind {
"entity" | "note" => None,
other => Some(other),
};
let props_filter: Option<&serde_json::Value> =
args_value.get("properties").and_then(|v| {
if v.as_object().is_some_and(|m| !m.is_empty()) {
Some(v)
} else {
None
}
});
let tags_owned: Vec<String> = match args_value.get("tags") {
None | Some(Value::Null) => vec![],
Some(v) => match serde_json::from_value::<Vec<String>>(v.clone()) {
Ok(t) => t,
Err(_) => {
return Some(Err((
"search".to_string(),
json!("tags must be an array of strings"),
)));
}
},
};
let coord_result = coord
.fan_out_search(
kind,
query,
&namespace,
limit,
kind_filter,
props_filter,
&tags_owned,
)
.await;
let result_val = if !coord_result.note_hits.is_empty()
|| (coord_result.entity_hits.is_empty() && coord_result.note_hits.is_empty())
{
let items: Vec<Value> = coord_result
.note_hits
.iter()
.filter(|h| h.score.to_f64() >= score_floor)
.map(|h| {
let note_kind = coord_result.note_kinds.get(&h.note_id);
json!({
"id": h.note_id.to_string(),
"note_kind": note_kind,
"score": h.score.to_f64(),
"title": h.title,
"snippet": h.snippet,
})
})
.collect();
serde_json::to_value(items).unwrap_or_else(|_| json!([]))
} else {
let items: Vec<Value> = coord_result
.entity_hits
.iter()
.filter(|h| h.score.to_f64() >= score_floor)
.map(|h| {
let entity_kind = coord_result.entity_kinds.get(&h.entity_id);
json!({
"id": h.entity_id.to_string(),
"entity_kind": entity_kind,
"score": h.score.to_f64(),
"title": h.title,
"snippet": h.snippet,
})
})
.collect();
serde_json::to_value(items).unwrap_or_else(|_| json!([]))
};
Some(Ok(result_val))
}
_ => None,
}
}
fn apply_presentation_to_result(
mut result_obj: Value,
mode: PresentationMode,
now_unix: i64,
) -> Value {
if result_obj.get("ok").and_then(Value::as_bool) == Some(true) {
if let Some(result_field) = result_obj.get("result").cloned() {
let presented = present(result_field, mode, now_unix);
if let Some(obj) = result_obj.as_object_mut() {
obj.insert("result".to_string(), presented);
}
}
}
result_obj
}
#[tool_router]
impl KhiveMcpServer {
#[tool(description = r#"Run one or more khive verbs in a single MCP call.
ops syntax:
Single op : verb(name=value, name=value)
Batch : [verb(...), verb(...)] — parallel, max 100
Chain : verb1(...) | verb2(id=$prev.id) — sequential, $prev
JSON form : [{"tool":"verb","args":{...}}, ...] — INDEPENDENT ops only
Argument values are JSON literals: strings (double-quoted), numbers, booleans,
null, arrays, objects. Strings may contain commas / parens; escape with \".
Chain-only: $prev resolves to the prior op's result. Path extraction syntax:
$prev — full result
$prev.field — nested object field
$prev.items[0].id — array index
$prev[2] — top-level array index
Quoted strings that contain $prev are promoted to substitutions (e.g. id="$prev.id"
is the same as id=$prev.id). To pass a literal "$prev", escape with backslash:
\"\\$prev\". JSON form is for independent ops only — any $prev string in JSON
form is rejected.
Response shape:
{
"results": [ {"ok": true, "tool": "verb", "result": {...}}, ... ],
"summary": { "total": N, "succeeded": N, "failed": N, "aborted": N }
}
Parallel: a failed op does NOT abort siblings. Chain: failure aborts remaining
ops (reported as {"ok": false, "aborted": true}). Committed ops are not rolled back.
Verb discovery: install the `kg` / `gtd` plugins for usage skills. The verbs
currently registered on this server (pack-derived) are listed below. Argument
schemas live in each pack's docs and SKILL.md files.
Tip: for one-shot calls, the single-op form is the densest. Use batch when
several independent ops can run together; use chain when each op needs the prior
result (e.g. create then link with the new entity's id)."#)]
async fn request(&self, Parameters(p): Parameters<RequestParams>) -> Result<String, McpError> {
#[cfg(unix)]
{
let frame = self.wire_daemon_frame(&p);
if let Some(res) = crate::daemon::forward_or_spawn(&frame).await {
return res;
}
}
self.dispatch_request_wire(p).await
}
}
impl KhiveMcpServer {
#[cfg(unix)]
pub(crate) fn wire_daemon_frame(&self, p: &RequestParams) -> khive_runtime::DaemonRequestFrame {
khive_runtime::DaemonRequestFrame {
ops: p.ops.clone(),
presentation: p.presentation.clone(),
presentation_per_op: p.presentation_per_op.clone(),
namespace: self.default_namespace.clone(),
config_id: self.config_id.clone(),
protocol_version: khive_runtime::daemon::PROTOCOL_VERSION,
probe_only: false,
format: p.format.clone(),
format_per_op: p.format_per_op.clone(),
from_wire: true,
}
}
pub async fn dispatch_request_local(&self, p: RequestParams) -> Result<String, McpError> {
self.dispatch_request_inner(p, false).await
}
pub(crate) async fn dispatch_request_wire(&self, p: RequestParams) -> Result<String, McpError> {
self.dispatch_request_inner(p, true).await
}
pub(crate) async fn dispatch_request_inner(
&self,
p: RequestParams,
from_wire: bool,
) -> Result<String, McpError> {
let save_to = p.save_to.clone();
let parsed = parse_request(&p.ops).map_err(dsl_err_to_mcp)?;
let presentation = parse_presentation_mode(p.presentation.as_deref())
.map_err(|e| McpError::invalid_params(e, None))?;
let presentation_per_op: Option<Vec<Option<PresentationMode>>> =
if let Some(per_op_strs) = p.presentation_per_op {
let mut modes = Vec::with_capacity(per_op_strs.len());
for s in per_op_strs {
let mode = match s.as_deref() {
None => None,
Some(v) => Some(
parse_presentation_mode(Some(v))
.map_err(|e| McpError::invalid_params(e, None))?,
),
};
modes.push(mode);
}
Some(modes)
} else {
None
};
let batch_format = parse_output_format(p.format.as_deref())
.map_err(|e| McpError::invalid_params(e, None))?
.unwrap_or(self.default_output_format);
let format_per_op: Option<Vec<Option<OutputFormat>>> =
if let Some(per_op_strs) = p.format_per_op {
let mut fmts = Vec::with_capacity(per_op_strs.len());
for s in per_op_strs {
let fmt = match s.as_deref() {
None => None,
Some(v) => Some(
parse_output_format(Some(v))
.map_err(|e| McpError::invalid_params(e, None))?
.unwrap_or(batch_format),
),
};
fmts.push(fmt);
}
Some(fmts)
} else {
None
};
let result = self
.run_parsed(
parsed.ops,
parsed.mode,
presentation,
presentation_per_op.clone(),
from_wire,
)
.await;
if let Some(path_str) = save_to {
let path = std::path::Path::new(&path_str);
let manifest = crate::save_sink::write_and_manifest(&result, path)
.map_err(|e| McpError::internal_error(format!("save_to: {e}"), None))?;
return serde_json::to_string(&manifest)
.map_err(|e| McpError::internal_error(format!("serialize manifest: {e}"), None));
}
Ok(render_result(
result,
batch_format,
&format_per_op,
presentation,
&presentation_per_op,
&self.registry,
))
}
}
fn dsl_err_to_mcp(e: DslError) -> McpError {
McpError::invalid_params(e.to_string(), None)
}
fn parse_presentation_mode(s: Option<&str>) -> Result<PresentationMode, String> {
match s {
None | Some("agent") => Ok(PresentationMode::Agent),
Some("verbose") => Ok(PresentationMode::Verbose),
Some("human") => Ok(PresentationMode::Human),
Some(other) => Err(format!(
"unknown presentation mode {other:?}; valid values: \"agent\", \"verbose\", \"human\""
)),
}
}
fn parse_output_format(s: Option<&str>) -> Result<Option<OutputFormat>, String> {
match s {
None => Ok(None),
Some("json") => Ok(Some(OutputFormat::Json)),
Some("auto") => Ok(Some(OutputFormat::Auto)),
Some("table") => Ok(Some(OutputFormat::Table)),
Some(other) => Err(format!(
"unknown output format {other:?}; valid values: \"json\", \"auto\", \"table\""
)),
}
}
fn render_result(
value: serde_json::Value,
batch_format: OutputFormat,
format_per_op: &Option<Vec<Option<OutputFormat>>>,
presentation: PresentationMode,
presentation_per_op: &Option<Vec<Option<PresentationMode>>>,
registry: &VerbRegistry,
) -> String {
if batch_format == OutputFormat::Json && format_per_op.is_none() {
return serde_json::to_string(&value).unwrap_or_else(|_| "null".to_string());
}
if let serde_json::Value::Object(ref map) = value {
if let Some(serde_json::Value::Array(results)) = map.get("results") {
let mut out_results = Vec::with_capacity(results.len());
for (i, entry) in results.iter().enumerate() {
let per_op_fmt = format_per_op
.as_ref()
.and_then(|v| v.get(i))
.and_then(|x| *x)
.unwrap_or(batch_format);
let base_presentation = presentation_per_op
.as_ref()
.and_then(|v| v.get(i))
.and_then(|o| *o)
.unwrap_or(presentation);
let effective_presentation =
match entry.get("tool").and_then(serde_json::Value::as_str) {
Some(tool)
if registry.presentation_policy_for(tool)
== VerbPresentationPolicy::AlwaysVerbose =>
{
PresentationMode::Verbose
}
_ => base_presentation,
};
let is_ok = entry
.get("ok")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
if !is_ok || per_op_fmt == OutputFormat::Json {
out_results.push(entry.clone());
continue;
}
if let Some(result_val) = entry.get("result") {
let rendered =
render_format(result_val.clone(), per_op_fmt, effective_presentation);
let mut new_entry = entry.clone();
if let serde_json::Value::Object(ref mut emap) = new_entry {
emap.insert("result".to_string(), serde_json::Value::String(rendered));
}
out_results.push(new_entry);
} else {
out_results.push(entry.clone());
}
}
let mut out_map = map.clone();
out_map.insert("results".to_string(), serde_json::Value::Array(out_results));
return serde_json::to_string(&serde_json::Value::Object(out_map))
.unwrap_or_else(|_| "null".to_string());
}
}
render_format(value, batch_format, presentation)
}
#[tool_handler]
impl ServerHandler for KhiveMcpServer {
fn get_info(&self) -> ServerInfo {
let catalog = self.verb_catalog();
let builtins = builtin_pack_names().join(", ");
let instructions = format!(
"khive — request-only MCP surface. One tool, `request`, \
dispatches verbs through the loaded pack registry. Configure packs via \
KHIVE_PACKS or --pack (built-ins: {builtins}). Verbs registered on this \
server:\n{catalog}\nFor detailed usage of each verb, see the corresponding \
plugin's SKILL.md files."
);
ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
.with_server_info(Implementation::new(
env!("CARGO_PKG_NAME"),
env!("CARGO_PKG_VERSION"),
))
.with_instructions(instructions)
}
async fn list_tools(
&self,
_request: Option<rmcp::model::PaginatedRequestParams>,
_context: rmcp::service::RequestContext<rmcp::RoleServer>,
) -> Result<rmcp::model::ListToolsResult, McpError> {
let mut tools = Self::tool_router().list_all();
let catalog = self.verb_catalog();
for t in &mut tools {
if t.name == "request" {
let base = t.description.as_deref().unwrap_or("");
t.description = Some(std::borrow::Cow::Owned(format!(
"{base}\n\nVerbs registered on this server:\n{catalog}"
)));
}
}
Ok(rmcp::model::ListToolsResult {
tools,
meta: None,
next_cursor: None,
})
}
}
#[cfg(test)]
mod tests {
use super::build_verb_catalog;
fn t(pack: &str, verb: &str, desc: &str) -> (String, String, String) {
(pack.to_owned(), verb.to_owned(), desc.to_owned())
}
#[test]
fn single_pack_verbs_unchanged() {
let catalog = build_verb_catalog([
t("kg", "create", "Create an entity or note."),
t("kg", "list", "List entities."),
]);
assert_eq!(
catalog,
" create — Create an entity or note.\n list — List entities.\n"
);
}
#[test]
fn duplicate_verb_concatenates_descriptions_with_pack_attribution() {
let catalog = build_verb_catalog([
t("kg", "create", "Create an entity or note."),
t("gtd", "create", "Create a task."),
]);
assert!(catalog.contains("[kg] Create an entity or note."));
assert!(catalog.contains("[gtd] Create a task."));
assert_eq!(catalog.matches(" create — ").count(), 1);
}
#[test]
fn catalog_is_sorted_alphabetically() {
let catalog = build_verb_catalog([
t("kg", "search", "Search."),
t("kg", "assign", "Assign."),
t("kg", "list", "List."),
]);
let names: Vec<&str> = catalog
.lines()
.filter(|l| l.starts_with(" "))
.map(|l| l.trim_start().split(' ').next().unwrap())
.collect();
assert_eq!(names, vec!["assign", "list", "search"]);
}
}