Menu
Public repository / 228b5716ba0b7bcc87cce261ea6df891133a35bb
AkurAI-CRM / crates/crm-mcp/src/tools.rs
Latest change 46f8d3437352b9043c8406756a07d2c52c892d19 - Gate MCP merge tool on IdP role; send app.js 401s through login by Ólafur Búi Ólafsson
use akurai_json::Value;
use akurai_storage::BTree;
use crm_api::graph::{self, LinkKind};
use crm_api::lifecycle;
use crm_api::upper_bound;
use std::sync::{Arc, Mutex};
/// Tools that require an organization owner/admin (or the static credential).
const ADMIN_TOOLS: [&str; 1] = ["merge_account_companies"];
/// Who an authenticated MCP request acts as.
#[derive(Debug, Clone, PartialEq)]
pub struct Caller {
pub tenant: String,
/// Owner/admin IdP role, or the static service credential.
pub admin: bool,
}
pub struct ToolRegistry {
db: Arc<Mutex<BTree>>,
}
impl ToolRegistry {
pub fn new(db: Arc<Mutex<BTree>>) -> Self {
Self { db }
}
/// List all available tools with their schemas
pub fn list_tools(&self) -> Vec<Value> {
vec![
self.tool_def(
"list_people",
"List all CRM contacts",
vec![("limit", "integer", "Maximum number of results", false)],
),
self.tool_def(
"get_person",
"Get a person by ID",
vec![("id", "integer", "Person record ID", true)],
),
self.tool_def(
"create_person",
"Create a new person",
vec![
("firstName", "string", "First name", true),
("lastName", "string", "Last name", true),
("email", "string", "Email address", false),
("jobTitle", "string", "Job title", false),
("companyId", "integer", "Company ID", false),
],
),
self.tool_def(
"create_company",
"Create a new company",
vec![
("name", "string", "Company name", true),
("domain", "string", "Primary email or web domain", false),
("stage", "string", "Lifecycle stage (lead/customer/churned)", false),
],
),
self.tool_def(
"list_companies",
"List all companies",
vec![("limit", "integer", "Maximum number of results", false)],
),
self.tool_def(
"get_company",
"Get a company by ID",
vec![("id", "integer", "Company record ID", true)],
),
self.tool_def(
"list_opportunities",
"List all sales opportunities",
vec![
(
"stage",
"string",
"Filter by stage (new/screening/meeting/proposal/negotiation/won/lost)",
false,
),
("limit", "integer", "Maximum number of results", false),
],
),
self.tool_def(
"get_opportunity",
"Get an opportunity by ID",
vec![("id", "integer", "Opportunity record ID", true)],
),
self.tool_def(
"get_pipeline",
"Get pipeline summary with counts per stage",
vec![],
),
self.tool_def(
"search_crm",
"Search across all CRM entities",
vec![("query", "string", "Search term", true)],
),
self.tool_def(
"list_tasks",
"List tasks",
vec![
(
"status",
"string",
"Filter by status (todo/in_progress/done/cancelled)",
false,
),
("limit", "integer", "Maximum number of results", false),
],
),
self.tool_def(
"get_account_graph",
"Read a bounded tenant-authorized customer account graph",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
(
"limit",
"integer",
"Maximum links and related records (1-100)",
false,
),
],
),
self.tool_def(
"list_account_links",
"List bounded account links, including stale and orphaned status",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
(
"status",
"string",
"Optional active/stale/orphaned/unlinked/merged filter",
false,
),
("limit", "integer", "Maximum links (1-100)", false),
],
),
self.tool_def(
"link_account_resource",
"Link a CRM company to a source-owned opaque resource ID",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
(
"kind",
"string",
"organization/workspace/subscription/opportunity/installation",
true,
),
("externalId", "string", "Opaque source resource ID", true),
("sourceOwner", "string", "idp/platform/crm", false),
],
),
self.tool_def(
"unlink_account_resource",
"Audit and unlink a CRM account resource",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
("linkId", "integer", "Account link ID", true),
],
),
self.tool_def(
"merge_account_companies",
"Merge one tenant CRM company into another while retaining history",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("sourceCompanyId", "integer", "Company being merged", true),
("targetCompanyId", "integer", "Company that remains", true),
],
),
self.tool_def(
"list_orphaned_account_links",
"List bounded orphaned account links for an authorized tenant",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("limit", "integer", "Maximum links (1-100)", false),
],
),
self.tool_def(
"record_lifecycle_touchpoint",
"Record a source-attributed lifecycle touchpoint without message bodies or secrets",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
("source", "string", "Attribution source", true),
("event", "string", "Event name, e.g. first_value or renewal", true),
("stage", "string", "acquisition/activation/retention/expansion", true),
("occurredAt", "integer", "Unix timestamp; defaults to now", false),
("value", "integer", "Optional integer outcome value", false),
],
),
self.tool_def(
"record_email_touchpoint",
"Link an email thread to CRM records with metadata only — message id, participants, subject, summary and a deep link; never message bodies or secrets",
vec![
("messageId", "string", "Source email message or thread ID", true),
("participants", "array", "Participant email addresses", false),
("subject", "string", "Message subject (bounded, no body)", false),
("summary", "string", "Short human summary (bounded, no body)", false),
("threadLink", "string", "Deep link to the source thread", false),
("personId", "integer", "Linked CRM person ID", false),
("companyId", "integer", "Linked CRM company ID", false),
("opportunityId", "integer", "Linked CRM opportunity ID", false),
("occurredAt", "integer", "Unix timestamp; defaults to now", false),
],
),
self.tool_def(
"record_customer_health",
"Record an explainable customer-health snapshot with scored component facts",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
("score", "integer", "Overall health score 0-100", true),
("components", "array", "Named component scores, weights and fact counts", true),
],
),
self.tool_def(
"create_growth_experiment",
"Retain a growth experiment hypothesis, metrics, measurements and decision",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("name", "string", "Experiment name", true),
("hypothesis", "string", "Falsifiable hypothesis", true),
("baselineMetrics", "array", "Metric name/value/sampleSize baseline", true),
("successMetrics", "array", "Metric name/value/sampleSize success criteria", true),
("guardrailMetrics", "array", "Metric name/value/sampleSize guardrails", true),
("measurements", "array", "Optional observed metrics", false),
("decision", "string", "scale/iterate/stop; defaults to iterate", false),
],
),
self.tool_def(
"get_lifecycle_profile",
"Read a bounded company lifecycle timeline with first-value and activation dates",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
("limit", "integer", "Maximum touchpoints (1-100)", false),
],
),
self.tool_def(
"get_customer_health",
"Read an explainable customer-health snapshot and freshness",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("companyId", "integer", "CRM company ID", true),
],
),
self.tool_def(
"list_growth_experiments",
"Read bounded growth experiments with hypotheses, metric evidence and decisions",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("limit", "integer", "Maximum experiments (1-100)", false),
],
),
self.tool_def(
"get_growth_aggregates",
"Read bounded 24h, 7d and 30d funnel, source, cohort and event drilldowns",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("limit", "integer", "Maximum drilldown events per window (1-100)", false),
],
),
self.tool_def(
"recommend_growth_actions",
"Recommend retain, scale, iterate or stop actions from health and experiment evidence",
vec![
("tenantId", "string", "Authorized tenant identifier", true),
("limit", "integer", "Maximum recommendations (1-100)", false),
],
),
]
}
fn tool_def(
&self,
name: &str,
description: &str,
mut params: Vec<(&str, &str, &str, bool)>,
) -> Value {
if !params.iter().any(|(pname, _, _, _)| *pname == "tenantId") {
params.insert(
0,
("tenantId", "string", "Authenticated tenant scope", true),
);
}
let mut properties = Vec::new();
let mut required = Vec::new();
for (pname, ptype, pdesc, is_required) in params {
properties.push((
pname.to_string(),
Value::Object(vec![
("type".into(), Value::Str(ptype.into())),
("description".into(), Value::Str(pdesc.into())),
]),
));
if is_required {
required.push(Value::Str(pname.to_string()));
}
}
Value::Object(vec![
("name".into(), Value::Str(name.into())),
("description".into(), Value::Str(description.into())),
(
"inputSchema".into(),
Value::Object(vec![
("type".into(), Value::Str("object".into())),
("properties".into(), Value::Object(properties)),
("required".into(), Value::Array(required)),
]),
),
])
}
/// Execute a tool for an authenticated caller.
pub fn execute_authenticated(
&self,
tool_name: &str,
args: &[(String, Value)],
caller: &Caller,
) -> Result<Value, String> {
let requested = Self::tenant_arg(args)?;
if requested != caller.tenant {
return Err("tenant argument does not match authenticated tenant".into());
}
if ADMIN_TOOLS.contains(&tool_name) && !caller.admin {
return Err(crm_api::graph::ADMIN_REQUIRED.into());
}
self.execute(tool_name, args)
}
fn execute(&self, tool_name: &str, args: &[(String, Value)]) -> Result<Value, String> {
match tool_name {
"list_people" => self.list_people(args),
"get_person" => self.get_person(args),
"create_person" => self.create_person(args),
"create_company" => self.create_company(args),
"list_companies" => self.list_companies(args),
"get_company" => self.get_company(args),
"list_opportunities" => self.list_opportunities(args),
"get_opportunity" => self.get_opportunity(args),
"get_pipeline" => self.get_pipeline(args),
"search_crm" => self.search_crm(args),
"list_tasks" => self.list_tasks(args),
"get_account_graph" => self.get_account_graph(args),
"list_account_links" => self.list_account_links(args),
"list_orphaned_account_links" => self.list_orphaned_account_links(args),
"link_account_resource" => self.link_account_resource(args),
"unlink_account_resource" => self.unlink_account_resource(args),
"merge_account_companies" => self.merge_account_companies(args),
"record_lifecycle_touchpoint" => self.record_lifecycle_touchpoint(args),
"record_email_touchpoint" => self.record_email_touchpoint(args),
"record_customer_health" => self.record_customer_health(args),
"create_growth_experiment" => self.create_growth_experiment(args),
"get_lifecycle_profile" => self.get_lifecycle_profile(args),
"get_customer_health" => self.get_customer_health(args),
"list_growth_experiments" => self.list_growth_experiments(args),
"get_growth_aggregates" => self.get_growth_aggregates(args),
"recommend_growth_actions" => self.recommend_growth_actions(args),
_ => Err(format!("unknown tool: {tool_name}")),
}
}
pub fn handle_request(&self, body: &str) -> String {
self.handle_request_inner(body, None)
}
pub fn handle_request_authenticated(&self, body: &str, caller: &Caller) -> String {
self.handle_request_inner(body, Some(caller))
}
fn handle_request_inner(&self, body: &str, caller: Option<&Caller>) -> String {
let req = match akurai_json::parse(body) {
Ok(v) => v,
Err(e) => return self.json_rpc_error(None, -32700, &format!("Parse error: {e}")),
};
let method = req.get("method").and_then(|v| v.as_str()).unwrap_or("");
let id = req.get("id").cloned();
match method {
"tools/list" => {
let tools_list = self.list_tools();
self.json_rpc_result(
id,
Value::Object(vec![("tools".into(), Value::Array(tools_list))]),
)
}
"tools/call" => {
let params = match req.get("params") {
Some(Value::Object(pairs)) => pairs.clone(),
_ => vec![],
};
let tool_name = params
.iter()
.find(|(k, _)| k == "name")
.and_then(|(_, v)| v.as_str())
.unwrap_or("");
let arguments: Vec<(String, Value)> = params
.iter()
.find(|(k, _)| k == "arguments")
.and_then(|(_, v)| match v {
Value::Object(pairs) => Some(pairs.clone()),
_ => None,
})
.unwrap_or_default();
let execution = match caller {
Some(caller) => self.execute_authenticated(tool_name, &arguments, caller),
None => Err("authenticated MCP credential required".into()),
};
match execution {
Ok(result) => self.json_rpc_result(
id,
Value::Object(vec![(
"content".into(),
Value::Array(vec![Value::Object(vec![
("type".into(), Value::Str("json".into())),
("text".into(), Value::Str(result.to_json())),
])]),
)]),
),
Err(e) => self.json_rpc_error(id, -32603, &format!("Tool error: {e}")),
}
}
"tools/get" => {
let params = match req.get("params") {
Some(Value::Object(pairs)) => pairs.clone(),
_ => vec![],
};
let tool_name = params
.iter()
.find(|(k, _)| k == "name")
.and_then(|(_, v)| v.as_str())
.unwrap_or("");
let tools_list = self.list_tools();
let tool = tools_list
.into_iter()
.find(|t| t.get("name").and_then(|v| v.as_str()) == Some(tool_name));
match tool {
Some(t) => self.json_rpc_result(id, Value::Object(vec![("tool".into(), t)])),
None => self.json_rpc_error(id, -32602, &format!("Unknown tool: {tool_name}")),
}
}
"initialize" => self.json_rpc_result(
id,
Value::Object(vec![
("protocolVersion".into(), Value::Str("2025-03-26".into())),
(
"capabilities".into(),
Value::Object(vec![("tools".into(), Value::Object(vec![]))]),
),
(
"serverInfo".into(),
Value::Object(vec![
("name".into(), Value::Str("akurai-crm-mcp".into())),
(
"version".into(),
Value::Str(env!("CARGO_PKG_VERSION").into()),
),
]),
),
]),
),
"notifications/initialized" => self.json_rpc_result(id, Value::Null),
_ => self.json_rpc_error(id, -32601, &format!("Method not found: {method}")),
}
}
fn json_rpc_result(&self, id: Option<Value>, result: Value) -> String {
let mut resp = Vec::new();
resp.push(("jsonrpc".into(), Value::Str("2.0".into())));
resp.push(("result".into(), result));
if let Some(id_val) = id {
resp.push(("id".into(), id_val));
}
Value::Object(resp).to_json()
}
fn json_rpc_error(&self, id: Option<Value>, code: i64, message: &str) -> String {
let mut resp = Vec::new();
resp.push(("jsonrpc".into(), Value::Str("2.0".into())));
resp.push((
"error".into(),
Value::Object(vec![
("code".into(), Value::Int(code)),
("message".into(), Value::Str(message.into())),
]),
));
if let Some(id_val) = id {
resp.push(("id".into(), id_val));
}
Value::Object(resp).to_json()
}
fn get_arg<'a>(args: &'a [(String, Value)], name: &str) -> Option<&'a Value> {
args.iter().find(|(k, _)| k == name).map(|(_, v)| v)
}
fn tenant_arg(args: &[(String, Value)]) -> Result<String, String> {
let tenant = Self::get_arg(args, "tenantId")
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|v| !v.is_empty())
.ok_or("missing tenantId")?;
if tenant.len() > 128
|| tenant
.chars()
.any(|c| c.is_control() || c == ':' || c == '/')
{
return Err("invalid tenantId".into());
}
Ok(tenant.to_string())
}
fn bounded_limit(args: &[(String, Value)]) -> usize {
Self::get_arg(args, "limit")
.and_then(|v| v.as_i64())
.and_then(|v| usize::try_from(v).ok())
.unwrap_or(50)
.clamp(1, graph::MAX_GRAPH_LIMIT)
}
fn company_arg(args: &[(String, Value)], name: &str) -> Result<u64, String> {
Self::get_arg(args, name)
.and_then(|v| v.as_i64())
.and_then(|v| u64::try_from(v).ok())
.ok_or_else(|| format!("missing {name}"))
}
fn get_account_graph(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
graph::get_company_graph(&mut db, &tenant, company, Self::bounded_limit(args))
.map_err(|e| format!("{e:?}"))
}
fn list_account_links(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let status = Self::get_arg(args, "status").and_then(|v| v.as_str());
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
graph::list_links(&mut db, &tenant, company, status, Self::bounded_limit(args))
.map_err(|e| format!("{e:?}"))
}
fn list_orphaned_account_links(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
graph::list_orphaned_links(&mut db, &tenant, Self::bounded_limit(args))
.map_err(|e| format!("{e:?}"))
}
fn link_account_resource(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let kind = Self::get_arg(args, "kind")
.and_then(|v| v.as_str())
.and_then(LinkKind::parse)
.ok_or("invalid kind")?;
let external_id = Self::get_arg(args, "externalId")
.and_then(|v| v.as_str())
.ok_or("missing externalId")?;
let owner = Self::get_arg(args, "sourceOwner").and_then(|v| v.as_str());
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
graph::link_company_resource(&mut db, &tenant, company, kind, external_id, owner)
.map_err(|e| format!("{e:?}"))
}
fn unlink_account_resource(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let link_id = Self::company_arg(args, "linkId")?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
graph::unlink_company_resource(&mut db, &tenant, company, link_id)
.map_err(|e| format!("{e:?}"))
}
fn merge_account_companies(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let source = Self::company_arg(args, "sourceCompanyId")?;
let target = Self::company_arg(args, "targetCompanyId")?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
graph::merge_companies(&mut db, &tenant, source, target).map_err(|e| format!("{e:?}"))
}
fn record_lifecycle_touchpoint(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::record_touchpoint(&mut db, &tenant, company, &Value::Object(args.to_vec()))
.map_err(|e| format!("{e:?}"))
}
fn record_customer_health(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::record_health_snapshot(&mut db, &tenant, company, &Value::Object(args.to_vec()))
.map_err(|e| format!("{e:?}"))
}
fn create_growth_experiment(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::create_experiment(&mut db, &tenant, &Value::Object(args.to_vec()))
.map_err(|e| format!("{e:?}"))
}
fn get_lifecycle_profile(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::lifecycle_profile(&mut db, &tenant, company, Self::bounded_limit(args))
.map_err(|e| format!("{e:?}"))
}
fn get_customer_health(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let company = Self::company_arg(args, "companyId")?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::customer_health(&mut db, &tenant, company).map_err(|e| format!("{e:?}"))
}
fn list_growth_experiments(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::list_experiments(&mut db, &tenant, Self::bounded_limit(args))
.map_err(|e| format!("{e:?}"))
}
fn get_growth_aggregates(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::growth_aggregates(&mut db, &tenant, Self::bounded_limit(args))
.map_err(|e| format!("{e:?}"))
}
fn recommend_growth_actions(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
lifecycle::growth_recommendations(&mut db, &tenant, Self::bounded_limit(args))
.map_err(|e| format!("{e:?}"))
}
fn scan_collection(&self, collection: &str, tenant: &str) -> Result<Vec<Value>, String> {
let prefix = format!("{collection}:");
let start = prefix.as_bytes();
let end = upper_bound(start);
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
let mut results = Vec::new();
let entries = db.range(start, &end).map_err(|e| format!("storage: {e}"))?;
for (key, val_bytes) in entries {
if key.ends_with(b":_counter") {
continue;
}
let value = akurai_json::parse(std::str::from_utf8(&val_bytes).unwrap_or(""))
.map_err(|e| format!("parse: {e}"))?;
if value.get("tenantId").and_then(Value::as_str) == Some(tenant) {
results.push(value);
}
}
Ok(results)
}
fn get_record(&self, collection: &str, id: u64, tenant: &str) -> Result<Option<Value>, String> {
let mut key = format!("{collection}:").into_bytes();
key.extend_from_slice(&id.to_be_bytes());
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
match db.get(&key).map_err(|e| format!("storage: {e}"))? {
Some(bytes) => {
let value = akurai_json::parse(std::str::from_utf8(&bytes).unwrap_or(""))
.map_err(|e| format!("parse: {e}"))?;
if value.get("tenantId").and_then(Value::as_str) != Some(tenant) {
return Err("record not found in authenticated tenant".into());
}
Ok(Some(value))
}
None => Ok(None),
}
}
fn page(values: Vec<Value>, args: &[(String, Value)]) -> Value {
let requested = Self::get_arg(args, "limit")
.and_then(Value::as_i64)
.and_then(|value| usize::try_from(value).ok());
let truncated = requested.is_some_and(|limit| values.len() > limit);
let records = match requested {
Some(limit) => values.into_iter().take(limit).collect(),
None => values,
};
Value::Object(vec![
("records".into(), Value::Array(records)),
("truncated".into(), Value::Bool(truncated)),
("complete".into(), Value::Bool(!truncated)),
])
}
fn list_people(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
Ok(Self::page(self.scan_collection("people", &tenant)?, args))
}
fn get_person(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let id = Self::get_arg(args, "id")
.and_then(Value::as_i64)
.and_then(|id| u64::try_from(id).ok())
.ok_or("missing id")?;
self.get_record("people", id, &tenant)
.map(|opt| opt.unwrap_or(Value::Null))
}
fn create_person(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
if let Some(company_id) = Self::get_arg(args, "companyId")
.and_then(Value::as_i64)
.and_then(|id| u64::try_from(id).ok())
{
let company = db
.get(&build_key("companies", company_id))
.map_err(|e| format!("storage: {e}"))?
.ok_or("company not found")?;
let company = akurai_json::parse(std::str::from_utf8(&company).unwrap_or(""))
.map_err(|e| format!("parse: {e}"))?;
if company.get("tenantId").and_then(Value::as_str) != Some(&tenant) {
return Err("company belongs to another tenant".into());
}
}
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs() as i64;
let counter_key = b"people:_counter";
let next_id = db
.get(counter_key)
.map_err(|e| format!("counter read: {e}"))?
.and_then(|bytes| String::from_utf8(bytes).ok())
.and_then(|value| value.parse::<u64>().ok())
.and_then(|value| value.checked_add(1))
.unwrap_or(1);
let first = Self::get_arg(args, "firstName")
.and_then(Value::as_str)
.unwrap_or("");
let last = Self::get_arg(args, "lastName")
.and_then(Value::as_str)
.unwrap_or("");
let mut fields = vec![
("id".into(), Value::Int(next_id as i64)),
("tenantId".into(), Value::Str(tenant)),
("firstName".into(), Value::Str(first.into())),
("lastName".into(), Value::Str(last.into())),
("name".into(), Value::Str(format!("{first} {last}"))),
("createdAt".into(), Value::Int(now)),
("updatedAt".into(), Value::Int(now)),
];
for name in ["email", "jobTitle"] {
if let Some(value) = Self::get_arg(args, name).and_then(Value::as_str) {
fields.push((name.into(), Value::Str(value.into())));
}
}
if let Some(cid) = Self::get_arg(args, "companyId").and_then(Value::as_i64) {
fields.push(("companyId".into(), Value::Int(cid)));
}
let record = Value::Object(fields);
let key = build_key("people", next_id);
db.insert(counter_key, next_id.to_string().as_bytes())
.map_err(|e| format!("counter: {e}"))?;
db.insert(&key, record.to_json().as_bytes())
.map_err(|e| format!("insert: {e}"))?;
db.commit().map_err(|e| format!("commit: {e}"))?;
Ok(record)
}
fn create_company(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let name = Self::get_arg(args, "name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or("missing name")?
.to_string();
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs() as i64;
let counter_key = b"companies:_counter";
let next_id = db
.get(counter_key)
.map_err(|e| format!("counter read: {e}"))?
.and_then(|bytes| String::from_utf8(bytes).ok())
.and_then(|value| value.parse::<u64>().ok())
.and_then(|value| value.checked_add(1))
.unwrap_or(1);
let mut fields = vec![
("id".into(), Value::Int(next_id as i64)),
("tenantId".into(), Value::Str(tenant)),
("name".into(), Value::Str(name)),
("createdAt".into(), Value::Int(now)),
("updatedAt".into(), Value::Int(now)),
];
for field in ["domain", "stage"] {
if let Some(value) = Self::get_arg(args, field)
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
{
fields.push((field.into(), Value::Str(value.into())));
}
}
let record = Value::Object(fields);
let key = build_key("companies", next_id);
db.insert(counter_key, next_id.to_string().as_bytes())
.map_err(|e| format!("counter: {e}"))?;
db.insert(&key, record.to_json().as_bytes())
.map_err(|e| format!("insert: {e}"))?;
db.commit().map_err(|e| format!("commit: {e}"))?;
Ok(record)
}
fn record_email_touchpoint(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let message_id = Self::get_arg(args, "messageId")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or("missing messageId")?;
if message_id.len() > 512 {
return Err("messageId exceeds 512 bytes".into());
}
let message_id = message_id.to_string();
let bounded = |name: &str, max: usize| -> Result<Option<String>, String> {
match Self::get_arg(args, name)
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
{
Some(value) if value.len() > max => Err(format!(
"{name} exceeds {max} bytes — store metadata, not bodies"
)),
Some(value) => Ok(Some(value.to_string())),
None => Ok(None),
}
};
let subject = bounded("subject", 500)?;
let summary = bounded("summary", 1000)?;
let thread_link = bounded("threadLink", 1000)?;
let participants: Vec<Value> = match Self::get_arg(args, "participants") {
Some(Value::Array(items)) => items
.iter()
.filter_map(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.take(50)
.map(|value| Value::Str(value.to_lowercase()))
.collect(),
_ => Vec::new(),
};
let mut db = self.db.lock().map_err(|e| format!("lock: {e}"))?;
let mut links = Vec::new();
for (arg, collection) in [
("personId", "people"),
("companyId", "companies"),
("opportunityId", "opportunities"),
] {
if let Some(id) = Self::get_arg(args, arg)
.and_then(Value::as_i64)
.and_then(|id| u64::try_from(id).ok())
{
let record = db
.get(&build_key(collection, id))
.map_err(|e| format!("storage: {e}"))?
.ok_or_else(|| format!("{arg} not found"))?;
let record = akurai_json::parse(std::str::from_utf8(&record).unwrap_or(""))
.map_err(|e| format!("parse: {e}"))?;
if record.get("tenantId").and_then(Value::as_str) != Some(&tenant) {
return Err(format!("{arg} belongs to another tenant"));
}
links.push((arg.to_string(), Value::Int(id as i64)));
}
}
if links.is_empty() {
return Err("at least one of personId, companyId or opportunityId is required".into());
}
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs() as i64;
let occurred_at = Self::get_arg(args, "occurredAt")
.and_then(Value::as_i64)
.unwrap_or(now);
let counter_key = b"email_touchpoints:_counter";
let next_id = db
.get(counter_key)
.map_err(|e| format!("counter read: {e}"))?
.and_then(|bytes| String::from_utf8(bytes).ok())
.and_then(|value| value.parse::<u64>().ok())
.and_then(|value| value.checked_add(1))
.unwrap_or(1);
let mut fields = vec![
("id".into(), Value::Int(next_id as i64)),
("tenantId".into(), Value::Str(tenant)),
("messageId".into(), Value::Str(message_id)),
("participants".into(), Value::Array(participants)),
("occurredAt".into(), Value::Int(occurred_at)),
("createdAt".into(), Value::Int(now)),
];
fields.extend(links);
if let Some(value) = subject {
fields.push(("subject".into(), Value::Str(value)));
}
if let Some(value) = summary {
fields.push(("summary".into(), Value::Str(value)));
}
if let Some(value) = thread_link {
fields.push(("threadLink".into(), Value::Str(value)));
}
let record = Value::Object(fields);
let key = build_key("email_touchpoints", next_id);
db.insert(counter_key, next_id.to_string().as_bytes())
.map_err(|e| format!("counter: {e}"))?;
db.insert(&key, record.to_json().as_bytes())
.map_err(|e| format!("insert: {e}"))?;
db.commit().map_err(|e| format!("commit: {e}"))?;
Ok(record)
}
fn list_companies(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
Ok(Self::page(
self.scan_collection("companies", &tenant)?,
args,
))
}
fn get_company(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let id = Self::get_arg(args, "id")
.and_then(Value::as_i64)
.and_then(|id| u64::try_from(id).ok())
.ok_or("missing id")?;
self.get_record("companies", id, &tenant)
.map(|opt| opt.unwrap_or(Value::Null))
}
fn list_opportunities(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut opps = self.scan_collection("opportunities", &tenant)?;
if let Some(stage) = Self::get_arg(args, "stage").and_then(Value::as_str) {
opps.retain(|o| o.get("stage").and_then(Value::as_str) == Some(stage));
}
Ok(Self::page(opps, args))
}
fn get_opportunity(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let id = Self::get_arg(args, "id")
.and_then(Value::as_i64)
.and_then(|id| u64::try_from(id).ok())
.ok_or("missing id")?;
self.get_record("opportunities", id, &tenant)
.map(|opt| opt.unwrap_or(Value::Null))
}
fn get_pipeline(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let opps = self.scan_collection("opportunities", &tenant)?;
let stages = [
"new",
"screening",
"meeting",
"proposal",
"negotiation",
"won",
"lost",
];
let stage_counts: Vec<Value> = stages
.iter()
.map(|&stage| {
let matching = opps
.iter()
.filter(|o| o.get("stage").and_then(Value::as_str) == Some(stage));
let list = matching.collect::<Vec<_>>();
let total_amount = list
.iter()
.filter_map(|o| o.get("amount").and_then(Value::as_i64))
.sum();
Value::Object(vec![
("stage".into(), Value::Str(stage.into())),
("count".into(), Value::Int(list.len() as i64)),
("totalAmount".into(), Value::Int(total_amount)),
])
})
.collect();
Ok(Value::Object(vec![
("pipeline".into(), Value::Array(stage_counts)),
("totalDeals".into(), Value::Int(opps.len() as i64)),
("complete".into(), Value::Bool(true)),
]))
}
fn search_crm(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let query = Self::get_arg(args, "query")
.and_then(Value::as_str)
.unwrap_or("")
.to_lowercase();
if query.is_empty() {
return Ok(Value::Object(vec![
("records".into(), Value::Array(vec![])),
("truncated".into(), Value::Bool(false)),
("complete".into(), Value::Bool(true)),
]));
}
let mut results = Vec::new();
for collection in ["people", "companies", "opportunities", "tasks", "notes"] {
for value in self.scan_collection(collection, &tenant)? {
if value.to_json().to_lowercase().contains(&query) {
results.push(Value::Object(vec![
("entity".into(), Value::Str(collection.into())),
("record".into(), value),
]));
}
}
}
Ok(Self::page(results, args))
}
fn list_tasks(&self, args: &[(String, Value)]) -> Result<Value, String> {
let tenant = Self::tenant_arg(args)?;
let mut tasks = self.scan_collection("tasks", &tenant)?;
if let Some(status) = Self::get_arg(args, "status").and_then(Value::as_str) {
tasks.retain(|task| task.get("status").and_then(Value::as_str) == Some(status));
}
Ok(Self::page(tasks, args))
}
}
fn build_key(collection: &str, id: u64) -> Vec<u8> {
let mut key = format!("{collection}:").into_bytes();
key.extend_from_slice(&id.to_be_bytes());
key
}
#[cfg(test)]
mod graph_tests {
use super::*;
#[test]
fn graph_tools_are_tenant_authorized_and_bounded() {
let path = format!("/tmp/akurai-crm-mcp-graph-{}.db", std::process::id());
let _ = std::fs::remove_file(&path);
let mut db = BTree::open(&path).unwrap();
let company = Value::Object(vec![
("id".into(), Value::Int(1)),
("name".into(), Value::Str("Acme".into())),
("tenantId".into(), Value::Str("tenant-a".into())),
]);
db.insert(
b"companies:\x00\x00\x00\x00\x00\x00\x00\x01",
company.to_json().as_bytes(),
)
.unwrap();
db.commit().unwrap();
let registry = ToolRegistry::new(Arc::new(Mutex::new(db)));
let args = vec![
("tenantId".into(), Value::Str("tenant-a".into())),
("companyId".into(), Value::Int(1)),
("kind".into(), Value::Str("organization".into())),
("externalId".into(), Value::Str("org_1".into())),
];
registry.execute("link_account_resource", &args).unwrap();
let graph = registry
.execute(
"get_account_graph",
&[
("tenantId".into(), Value::Str("tenant-a".into())),
("companyId".into(), Value::Int(1)),
("limit".into(), Value::Int(1)),
],
)
.unwrap();
assert_eq!(graph.get("bounded").and_then(Value::as_bool), Some(true));
assert!(registry
.execute(
"get_account_graph",
&[
("tenantId".into(), Value::Str("tenant-b".into())),
("companyId".into(), Value::Int(1)),
],
)
.is_err());
}
}