Menu
akurai-tasks
publicLatest change 4b13f3519f4ef1275b7f066ed19a278a822a7260 - fix: accept array, JSON, or CSV repos over MCP by Ólafur Búi Ólafsson
#![forbid(unsafe_code)]
use akurai_json::{parse, Value};
use tasks_core::{field, object, string, CreateItem, Query, Store};
#[derive(Clone)]
pub struct Mcp {
store: Store,
actor: String,
}
impl Mcp {
pub fn new(store: Store, actor: String) -> Self {
Self { store, actor }
}
pub fn handle_line(&self, line: &str) -> String {
let request = match parse(line) {
Ok(request) => request,
Err(error) => return rpc_error(Value::Null, -32700, &format!("parse error: {error}")),
};
let id = request.get("id").cloned();
if id.is_none() {
return String::new();
}
let id = id.unwrap_or(Value::Null);
let method = field(&request, "method").unwrap_or("");
let result = match method {
"initialize" => Ok(object(vec![
("protocolVersion", string("2025-11-25")),
(
"capabilities",
object(vec![
("tools", object(vec![])),
("resources", object(vec![])),
("prompts", object(vec![])),
]),
),
(
"serverInfo",
object(vec![
("name", string("akurai-tasks")),
("version", string(env!("CARGO_PKG_VERSION"))),
]),
),
])),
"ping" => Ok(object(vec![])),
"tools/list" => Ok(object(vec![("tools", Value::Array(tools()))])),
"tools/call" => self.call(request.get("params").unwrap_or(&Value::Null)),
"resources/list" => Ok(object(vec![(
"resources",
Value::Array(vec![object(vec![
("uri", string("tasks://projects")),
("name", string("Projects")),
("mimeType", string("application/json")),
])]),
)])),
"resources/read" => self.read_resource(request.get("params").unwrap_or(&Value::Null)),
"prompts/list" => Ok(object(vec![(
"prompts",
Value::Array(vec![object(vec![
("name", string("next-task")),
(
"description",
string("Select the ready execution frontier for a project"),
),
(
"arguments",
Value::Array(vec![object(vec![
("name", string("project")),
("required", Value::Bool(true)),
])]),
),
])]),
)])),
"prompts/get" => self.get_prompt(request.get("params").unwrap_or(&Value::Null)),
_ => return rpc_error(id, -32601, &format!("method not found: {method}")),
};
match result {
Ok(value) => rpc_result(id, value),
Err(message) => rpc_error(id, -32602, &message),
}
}
fn call(&self, params: &Value) -> Result<Value, String> {
let name = required(params, "name")?;
let arguments = params.get("arguments").unwrap_or(&Value::Null);
let idempotency = field(arguments, "idempotencyKey").unwrap_or("");
let result = match name {
"project_list" | "projects_list" => self.store.list_projects(),
"project_create" => self.store.create_project(
required(arguments, "key")?,
required(arguments, "name")?,
string_array(arguments, "repos")?,
&self.actor,
idempotency,
),
"project_update" => self.store.update_project(
required(arguments, "key")?,
string_array(arguments, "repos")?,
&self.actor,
idempotency,
),
"work_set_repo" | "task_set_repo" => self.store.set_item_repo(
required(arguments, "id")?,
Some(required(arguments, "repo")?),
&self.actor,
idempotency,
),
"task_list" | "work_list" => self.store.list_items(required(arguments, "project")?),
"task_show" | "work_get" => self.store.get_item(required(arguments, "id")?),
"task_create" | "work_create" => self.store.create_item(CreateItem {
project: required(arguments, "project")?,
title: required(arguments, "title")?,
description: field(arguments, "description").unwrap_or(""),
repo: field(arguments, "repo"),
priority: field(arguments, "priority").unwrap_or("normal"),
actor: &self.actor,
idempotency_key: idempotency,
}),
"task_transition" | "work_transition" => self.store.transition(
required(arguments, "id")?,
required(arguments, "state")?,
integer(arguments, "expectedRevision")?,
&self.actor,
idempotency,
),
"task_claim" | "work_claim" => self.store.claim(
required(arguments, "id")?,
&self.actor,
integer(arguments, "expectedRevision")?,
&self.actor,
idempotency,
),
"task_comment" | "task_evidence" => self.store.add_record(
if name == "task_comment" {
"comment"
} else {
"evidence"
},
required(arguments, "id")?,
required(arguments, "body")?,
&self.actor,
idempotency,
),
"task_dependency" => self.store.add_dependency(
required(arguments, "id")?,
required(arguments, "dependsOn")?,
&self.actor,
idempotency,
),
"task_handoff" => self.store.handoff(
required(arguments, "id")?,
required(arguments, "to")?,
required(arguments, "summary")?,
&self.actor,
idempotency,
),
"task_accept_handoff" => {
self.store
.accept_handoff(required(arguments, "id")?, &self.actor, idempotency)
}
"task_query" | "work_query" => self.store.query_items(&query_from(arguments)?),
"work_renew" => self.store.renew_lease(
required(arguments, "id")?,
required(arguments, "leaseToken")?,
integer(arguments, "leaseGeneration")?,
integer(arguments, "expectedRevision")?,
&self.actor,
idempotency,
),
"work_release" => self.store.release_lease(
required(arguments, "id")?,
required(arguments, "leaseToken")?,
integer(arguments, "leaseGeneration")?,
integer(arguments, "expectedRevision")?,
&self.actor,
idempotency,
),
"import_apply" => self.store.apply_import(
arguments.get("bundle").ok_or("bundle is required")?,
&self.actor,
idempotency,
),
"import_verify" => self.store.verify_import(required(arguments, "run")?),
"board_get" => self.store.board(required(arguments, "project")?),
"ready_frontier" => self.store.ready_frontier(required(arguments, "project")?),
_ => return Err(format!("unknown tool: {name}")),
}
.map_err(|error| error.to_string())?;
Ok(object(vec![(
"content",
Value::Array(vec![object(vec![
("type", string("text")),
("text", string(&result.to_json())),
])]),
)]))
}
fn read_resource(&self, params: &Value) -> Result<Value, String> {
match required(params, "uri")? {
"tasks://projects" => {
let projects = self
.store
.list_projects()
.map_err(|error| error.to_string())?;
Ok(object(vec![(
"contents",
Value::Array(vec![object(vec![
("uri", string("tasks://projects")),
("mimeType", string("application/json")),
("text", string(&projects.to_json())),
])]),
)]))
}
uri => Err(format!("unknown resource: {uri}")),
}
}
fn get_prompt(&self, params: &Value) -> Result<Value, String> {
if required(params, "name")? != "next-task" {
return Err("unknown prompt".into());
}
let project = params
.get("arguments")
.and_then(|value| field(value, "project"))
.ok_or_else(|| "project is required".to_string())?;
Ok(object(vec![
("description", string("Pull the next policy-valid leaf task")),
(
"messages",
Value::Array(vec![object(vec![
("role", string("user")),
(
"content",
object(vec![
("type", string("text")),
(
"text",
string(&format!(
"Call ready_frontier for project {project}, select one leaf item, then claim it with its current revision."
)),
),
]),
),
])]),
),
]))
}
}
fn tools() -> Vec<Value> {
[
("projects_list", "List all visible projects", &[][..]),
(
"project_create",
"Create a project with one or more repositories",
&["key", "name", "idempotencyKey"],
),
(
"project_update",
"Replace the repositories registered to a project",
&["key", "repos", "idempotencyKey"],
),
(
"work_set_repo",
"Set the repository a work item belongs to",
&["id", "repo", "idempotencyKey"],
),
(
"work_query",
"Query work across projects and repositories using structured filters",
&[][..],
),
("work_get", "Read one work item", &["id"]),
(
"work_create",
"Create a work item",
&["project", "title", "idempotencyKey"],
),
(
"work_transition",
"Move a work item through its workflow",
&["id", "state", "expectedRevision", "idempotencyKey"],
),
(
"work_claim",
"Claim a Ready work item and receive a fenced lease",
&["id", "expectedRevision", "idempotencyKey"],
),
(
"work_renew",
"Renew the current fenced lease",
&[
"id",
"leaseToken",
"leaseGeneration",
"expectedRevision",
"idempotencyKey",
],
),
(
"work_release",
"Release the current fenced lease",
&[
"id",
"leaseToken",
"leaseGeneration",
"expectedRevision",
"idempotencyKey",
],
),
(
"task_comment",
"Add a comment",
&["id", "body", "idempotencyKey"],
),
(
"task_evidence",
"Attach evidence",
&["id", "body", "idempotencyKey"],
),
(
"task_dependency",
"Add a work dependency",
&["id", "dependsOn", "idempotencyKey"],
),
(
"task_handoff",
"Request ownership handoff",
&["id", "to", "summary", "idempotencyKey"],
),
(
"task_accept_handoff",
"Accept ownership handoff",
&["id", "idempotencyKey"],
),
("board_get", "Read a project Kanban board", &["project"]),
(
"ready_frontier",
"List dependency-free Ready work",
&["project"],
),
(
"import_apply",
"Apply one lossless source import bundle",
&["bundle", "idempotencyKey"],
),
(
"import_verify",
"Verify imported records and exact-byte digests",
&["run"],
),
]
.into_iter()
.map(|(name, description, required)| {
let properties = if name == "work_query" {
query_properties()
} else {
object(
required
.iter()
.map(|field| (*field, schema_for(field)))
.collect(),
)
};
object(vec![
("name", string(name)),
("description", string(description)),
(
"inputSchema",
object(vec![
("type", string("object")),
("properties", properties),
(
"required",
Value::Array(required.iter().map(|name| string(name)).collect()),
),
("additionalProperties", Value::Bool(true)),
]),
),
])
})
.collect()
}
fn query_properties() -> Value {
object(vec![
("projects", array_schema()),
("repos", array_schema()),
("states", array_schema()),
("priorities", array_schema()),
("owners", array_schema()),
("types", array_schema()),
("labels", array_schema()),
("parent", schema_for(&"parent")),
("cycle", schema_for(&"cycle")),
("text", schema_for(&"text")),
("blocked", object(vec![("type", string("boolean"))])),
("createdAfter", integer_schema()),
("createdBefore", integer_schema()),
("updatedAfter", integer_schema()),
("updatedBefore", integer_schema()),
(
"sort",
object(vec![
("type", string("string")),
(
"enum",
Value::Array(
["id", "created", "updated", "priority", "state", "title"]
.iter()
.map(|v| string(v))
.collect(),
),
),
]),
),
("descending", object(vec![("type", string("boolean"))])),
("limit", integer_schema()),
("byteLimit", integer_schema()),
("cursor", schema_for(&"cursor")),
])
}
fn schema_for(name: &&str) -> Value {
match *name {
"expectedRevision" | "leaseGeneration" => integer_schema(),
"repos" => array_schema(),
_ => object(vec![("type", string("string"))]),
}
}
fn integer_schema() -> Value {
object(vec![("type", string("integer"))])
}
fn array_schema() -> Value {
object(vec![
("type", string("array")),
("items", object(vec![("type", string("string"))])),
])
}
fn required<'a>(value: &'a Value, key: &str) -> Result<&'a str, String> {
field(value, key).ok_or_else(|| format!("{key} is required"))
}
fn integer(value: &Value, key: &str) -> Result<i64, String> {
value
.get(key)
.and_then(Value::as_i64)
.ok_or_else(|| format!("{key} must be an integer"))
}
fn string_array(value: &Value, key: &str) -> Result<Vec<String>, String> {
match value.get(key) {
None => Ok(Vec::new()),
Some(Value::Array(values)) => values
.iter()
.map(|value| {
value
.as_str()
.map(str::to_string)
.ok_or_else(|| format!("{key} must contain strings"))
})
.collect(),
Some(other) => {
if other == &Value::Null {
return Ok(Vec::new());
}
let Some(raw) = other.as_str() else {
return Err(format!("{key} must be an array"));
};
let trimmed = raw.trim();
if trimmed.is_empty() {
return Ok(Vec::new());
}
if trimmed.starts_with('[') {
return match parse(trimmed) {
Ok(Value::Array(values)) => values
.iter()
.map(|value| {
value
.as_str()
.map(str::to_string)
.ok_or_else(|| format!("{key} must contain strings"))
})
.collect(),
_ => Err(format!("{key} must be an array")),
};
}
Ok(trimmed
.split(',')
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string)
.collect())
}
}
}
fn query_from(value: &Value) -> Result<Query, String> {
Ok(Query {
projects: string_array(value, "projects")?,
repos: string_array(value, "repos")?,
states: string_array(value, "states")?,
priorities: string_array(value, "priorities")?,
owners: string_array(value, "owners")?,
types: string_array(value, "types")?,
labels: string_array(value, "labels")?,
parent: field(value, "parent").map(str::to_string),
cycle: field(value, "cycle").map(str::to_string),
text: field(value, "text").map(str::to_string),
blocked: value.get("blocked").and_then(Value::as_bool),
created_after: value.get("createdAfter").and_then(Value::as_i64),
created_before: value.get("createdBefore").and_then(Value::as_i64),
updated_after: value.get("updatedAfter").and_then(Value::as_i64),
updated_before: value.get("updatedBefore").and_then(Value::as_i64),
sort: field(value, "sort").unwrap_or("id").to_string(),
descending: value
.get("descending")
.and_then(Value::as_bool)
.unwrap_or(false),
limit: value
.get("limit")
.and_then(Value::as_i64)
.unwrap_or(100)
.try_into()
.map_err(|_| "limit must be positive")?,
byte_limit: value
.get("byteLimit")
.and_then(Value::as_i64)
.unwrap_or(1_048_576)
.try_into()
.map_err(|_| "byteLimit must be positive")?,
cursor: field(value, "cursor").map(str::to_string),
})
}
fn rpc_result(id: Value, result: Value) -> String {
object(vec![
("jsonrpc", string("2.0")),
("id", id),
("result", result),
])
.to_json()
}
fn rpc_error(id: Value, code: i64, message: &str) -> String {
object(vec![
("jsonrpc", string("2.0")),
("id", id),
(
"error",
object(vec![
("code", Value::Int(code)),
("message", string(message)),
]),
),
])
.to_json()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn exposes_and_calls_multi_project_tools() {
let path = std::env::temp_dir().join(format!("tasks-mcp-{}.db", std::process::id()));
let _ = std::fs::remove_file(&path);
let mcp = Mcp::new(Store::open(&path).unwrap(), "agent".into());
let initialize = mcp.handle_line(r#"{"jsonrpc":"2.0","id":1,"method":"initialize"}"#);
assert!(initialize.contains("2025-11-25"));
let created = mcp.handle_line(r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"project_create","arguments":{"key":"OPS","name":"Ops","repos":["owner/repo"],"idempotencyKey":"p1"}}}"#);
assert!(created.contains("OPS"));
let listed = mcp.handle_line(r#"{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"project_list","arguments":{}}}"#);
assert!(listed.contains("owner/repo"));
let _ = std::fs::remove_file(path);
}
}