Files
dusuniot/mcp/server.ts
T

513 lines
25 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/**
* # MCP 服务器 DusunIoT CMS(stdio)
* 让 ZCode 等 Agent 直接管理站点内容:页面/文章 CRUD(支持 Elementor JSON
* 自定义单页面)、修订恢复、媒体上传、产品/订单/询盘查询与状态流转、看板统计。
*
* 协议:JSON-RPC 2.0 over stdio(MCP 2024-11-05),无 SDK 依赖,Bun 直跑:
* bun mcp/server.ts
* 认证:本地直连 PostgreSQL(与站点同库),仅限受信任的本机环境使用;
* 写操作全部落 audit_logs(actorId 为空表示 MCP 调用)。
*/
import { SQL } from "bun";
import { parseImageSize } from "../src/server/image-size";
/** # 成员变量 数据库连接(与站点同库;单连接足够 MCP 串行调用) **/
const sql = new SQL(process.env.DATABASE_URL || "postgres://dusuniot:dusuniot@127.0.0.1:25432/dusuniot", { max: 2 });
/** # 成员变量 MCP 工具定义(name/description/inputSchema) **/
const TOOLS = [
{
name: "dashboard_stats",
description: "站点运营看板统计:内容/产品/订单/询盘/媒体计数、新询盘数、待处理订单数、营收(按币种)、近 7 天询盘趋势",
inputSchema: { type: "object", properties: {}, additionalProperties: false },
},
{
name: "list_contents",
description: "内容列表(页面 page / 文章 post / 落地页 landing),支持状态与关键字过滤、分页;返回 id/slug/标题/状态/更新时间",
inputSchema: {
type: "object",
properties: {
type: { type: "string", enum: ["page", "post", "landing", "template"], description: "内容类型过滤" },
status: { type: "string", enum: ["draft", "review", "published"], description: "状态过滤" },
q: { type: "string", description: "标题/别名关键字" },
page: { type: "number", description: "页码,默认 1" },
pageSize: { type: "number", description: "每页条数,默认 20,上限 100" },
},
additionalProperties: false,
},
},
{
name: "get_content",
description: "读取单条内容全文:正文 HTML、Elementor JSON(meta.elementor)、SEO 字段、修订数",
inputSchema: { type: "object", properties: { id: { type: "number" } }, required: ["id"], additionalProperties: false },
},
{
name: "create_content",
description: "新建内容(自定义单页面推荐 type=page + elementorJson,前台 /slug/ 通栏渲染;纯 HTML 页用 html 字段)",
inputSchema: {
type: "object",
properties: {
type: { type: "string", enum: ["page", "post", "landing"], description: "内容类型" },
title: { type: "string", description: "标题(必填)" },
slug: { type: "string", description: "URL 别名,留空按标题生成" },
html: { type: "string", description: "正文 HTML" },
elementorJson: { type: "string", description: "Elementor 文档 JSON 字符串(存 meta.elementor,优先于 HTML 渲染)" },
excerpt: { type: "string" },
status: { type: "string", enum: ["draft", "review", "published"] },
seoTitle: { type: "string" },
seoDescription: { type: "string" },
},
required: ["type", "title"],
additionalProperties: false,
},
},
{
name: "update_content",
description: "更新内容(部分字段即可;elementorJson 传 null 表示清除 Elementor 数据)。更新前自动写修订快照",
inputSchema: {
type: "object",
properties: {
id: { type: "number" },
title: { type: "string" },
slug: { type: "string" },
html: { type: "string" },
elementorJson: { type: ["string", "null"], description: "Elementor JSON 字符串或 null(清除)" },
excerpt: { type: "string" },
status: { type: "string", enum: ["draft", "review", "published"] },
seoTitle: { type: ["string", "null"] },
seoDescription: { type: ["string", "null"] },
},
required: ["id"],
additionalProperties: false,
},
},
{
name: "delete_content",
description: "删除内容(物理删除)",
inputSchema: { type: "object", properties: { id: { type: "number" } }, required: ["id"], additionalProperties: false },
},
{
name: "list_revisions",
description: "内容修订快照列表(每次保存自动留档)",
inputSchema: { type: "object", properties: { contentId: { type: "number" } }, required: ["contentId"], additionalProperties: false },
},
{
name: "restore_revision",
description: "恢复内容到指定修订(当前版本先再留档)",
inputSchema: { type: "object", properties: { id: { type: "number", description: "修订 ID" } }, required: ["id"], additionalProperties: false },
},
{
name: "list_media",
description: "媒体库列表(路径/URL/宽高/大小),支持文件名关键字与分页",
inputSchema: {
type: "object",
properties: { q: { type: "string" }, page: { type: "number" }, pageSize: { type: "number" } },
additionalProperties: false,
},
},
{
name: "upload_media",
description: "上传媒体文件(base64),写入 uploads/YYYY/MM/ 并入库,自动解析图片宽高",
inputSchema: {
type: "object",
properties: {
filename: { type: "string" },
mime: { type: "string" },
dataBase64: { type: "string", description: "文件内容 base64(≤20MB)" },
},
required: ["filename", "dataBase64"],
additionalProperties: false,
},
},
{
name: "list_products",
description: "产品列表(SKU/标题/价格/库存/状态),支持关键字与分页",
inputSchema: {
type: "object",
properties: { q: { type: "string" }, page: { type: "number" }, pageSize: { type: "number" } },
additionalProperties: false,
},
},
{
name: "list_orders",
description: "订单列表(单号/状态/金额/客户名),支持状态过滤与分页",
inputSchema: {
type: "object",
properties: { status: { type: "string" }, page: { type: "number" }, pageSize: { type: "number" } },
additionalProperties: false,
},
},
{
name: "update_order_status",
description: "更新订单状态(pending/on-hold/processing/completed/cancelled/refunded)",
inputSchema: {
type: "object",
properties: { id: { type: "number" }, status: { type: "string" } },
required: ["id", "status"],
additionalProperties: false,
},
},
{
name: "list_inquiries",
description: "询盘列表(姓名/邮箱/公司/国家/感兴趣产品/状态),支持状态过滤与分页;CRM 跟进用",
inputSchema: {
type: "object",
properties: { status: { type: "string", enum: ["new", "read", "replied", "spam", "archived"] }, q: { type: "string" }, page: { type: "number" }, pageSize: { type: "number" } },
additionalProperties: false,
},
},
{
name: "update_inquiry_status",
description: "更新询盘状态(跟进流转:new→read→replied,或标记 spam/archived)",
inputSchema: {
type: "object",
properties: { id: { type: "number" }, status: { type: "string", enum: ["new", "read", "replied", "spam", "archived"] } },
required: ["id", "status"],
additionalProperties: false,
},
},
{
name: "get_setting",
description: "读取系统设置键(site/woocommerce/mail/seo 等)",
inputSchema: { type: "object", properties: { key: { type: "string" } }, required: ["key"], additionalProperties: false },
},
{
name: "set_setting",
description: "写入系统设置键(整键覆写,value 须为 JSON 对象)",
inputSchema: {
type: "object",
properties: { key: { type: "string" }, value: { type: "object" } },
required: ["key", "value"],
additionalProperties: false,
},
},
];
/** # 函数 内部工具:把标题转成 URL 别名(slug),与后台 admin.ts 同规则 */
function slugify(title: string): string {
return (
title
.toLowerCase()
.replace(/[^a-z0-9\u4e00-\u9fa5]+/g, "-")
.replace(/^-+|-+$/g, "")
.slice(0, 200) || `item-${Date.now()}`
);
}
/** # 函数 内部工具:写审计日志(MCP 调用 actorId 为空) */
async function audit(action: string, entity: string, entityId: string | number, meta?: unknown) {
await sql`insert into audit_logs (actor_id, action, entity, entity_id, meta) values (null, ${action}, ${entity}, ${String(entityId)}, ${sql.json(meta ?? {})}::jsonb)`;
}
/** # 函数 内部工具:把 SQL 片段用 " and " 连接(Bun SQL 无 sql.join/sql.raw,嵌套模板实现) */
function andJoin(conds: SQL[]): SQL {
return conds.reduce((acc, c) => (acc ? (sql`${acc} and ${c}` as SQL) : c));
}
/** # 函数 内部工具:把 SQL 片段用 ", " 连接(UPDATE SET 列表用) */
function commaJoin(chunks: SQL[]): SQL {
return chunks.reduce((acc, c) => (acc ? (sql`${acc}, ${c}` as SQL) : c));
}
/** # 函数 内部工具:slug 冲突时追加 -2/-3 …(contents_slug_type_idx 唯一约束) */
async function uniqueSlug(base: string, type: string): Promise<string> {
let slug = base;
for (let n = 1; n < 100; n++) {
const hit = await sql`select 1 from contents where slug = ${slug} and type = ${type} limit 1`;
if (hit.length === 0) return slug;
slug = `${base}-${n + 1}`;
}
throw new Error("slug 生成失败:冲突过多");
}
/** # 函数 工具调用实现(tools/call 分发)
* 参数 name:工具名;args:入参对象
* 返回值 任意可 JSON 序列化结果(由调用方包成 MCP text content)
*/
async function callTool(name: string, args: Record<string, unknown>): Promise<unknown> {
const a = args ?? {};
switch (name) {
case "dashboard_stats": {
const [counts] = await sql`
select
(select count(*) from contents) as contents,
(select count(*) from products) as products,
(select count(*) from orders) as orders,
(select count(*) from inquiries) as inquiries,
(select count(*) from media) as media,
(select count(*) from orders where status in ('pending','on-hold')) as pending_orders,
(select count(*) from inquiries where status = 'new') as new_inquiries`;
const revenue = await sql`select currency, coalesce(sum(total),0) as total from orders where status in ('completed','processing') group by currency`;
const trend = await sql`
select to_char(date_trunc('day', created_at), 'YYYY-MM-DD') as day, count(*) as n
from inquiries where created_at > now() - interval '7 days' group by 1 order by 1`;
return { counts, revenue, trend };
}
case "list_contents": {
const page = Math.max(1, Number(a.page) || 1);
const pageSize = Math.min(100, Number(a.pageSize) || 20);
const conds: SQL[] = [];
if (a.type) conds.push(sql`type = ${a.type}`);
if (a.status) conds.push(sql`status = ${a.status}`);
if (a.q) conds.push(sql`(title ilike ${"%" + a.q + "%"} or slug ilike ${"%" + a.q + "%"})`);
const where = conds.length ? sql` where ${andJoin(conds)}` : sql``;
const items = await sql`select id, type, slug, title, status, template, views, published_at, updated_at from contents${where} order by updated_at desc limit ${pageSize} offset ${(page - 1) * pageSize}`;
const [{ total }] = await sql`select count(*) as total from contents${where}`;
return { items, total: Number(total), page, pageSize };
}
case "get_content": {
const [row] = await sql`select * from contents where id = ${Number(a.id)}`;
if (!row) throw new Error(`内容 #${a.id} 不存在`);
const revs = await sql`select count(*) as n from content_revisions where content_id = ${Number(a.id)}`;
return { ...row, revisionCount: Number(revs[0].n) };
}
case "create_content": {
const type = String(a.type || "page");
const title = String(a.title || "").trim();
if (!title) throw new Error("title 必填");
const slug = await uniqueSlug(String(a.slug || "").trim() || slugify(title), type);
// elementorJson 解析进 meta.elementor(非法 JSON 直接报错,防脏数据入渲染器)
const meta: Record<string, unknown> = {};
if (typeof a.elementorJson === "string" && a.elementorJson.trim()) {
try {
meta.elementor = JSON.parse(a.elementorJson);
} catch {
throw new Error("elementorJson 不是合法 JSON");
}
}
const status = String(a.status || "draft");
const [row] = await sql`
insert into contents (type, slug, title, excerpt, html, status, template, seo_title, seo_description, meta, published_at, author_id)
values (${type}, ${slug}, ${title}, ${String(a.excerpt || "")}, ${String(a.html || "")}, ${status}, 'default',
${a.seoTitle ? String(a.seoTitle) : null}, ${a.seoDescription ? String(a.seoDescription) : null},
${sql.json(meta)}::jsonb, ${status === "published" ? new Date() : null}, null)
returning id`;
await audit("content.create", "content", row.id, { title, type, via: "mcp" });
return { ok: true, id: row.id, slug, frontUrl: status === "published" ? `/${slug}/` : null };
}
case "update_content": {
const id = Number(a.id);
const [cur] = await sql`select id, title, html, meta, status, published_at from contents where id = ${id}`;
if (!cur) throw new Error(`内容 #${id} 不存在`);
// 修订快照(与后台保存一致)
await sql`insert into content_revisions (content_id, title, html, author_id) values (${id}, ${cur.title}, ${cur.html}, null)`;
// 逐字段构建 SET(仅更新显式传入的字段)
const sets: SQL[] = [sql`updated_at = ${new Date()}`];
if (a.title !== undefined) sets.push(sql`title = ${String(a.title)}`);
if (a.slug !== undefined) sets.push(sql`slug = ${await uniqueSlug(String(a.slug), String(cur.type ?? "page"))}`);
if (a.html !== undefined) sets.push(sql`html = ${String(a.html)}`);
if (a.excerpt !== undefined) sets.push(sql`excerpt = ${String(a.excerpt)}`);
if (a.status !== undefined) {
const st = String(a.status);
sets.push(sql`status = ${st}`);
if (st === "published" && !cur.published_at) sets.push(sql`published_at = ${new Date()}`);
}
if (a.seoTitle !== undefined) sets.push(sql`seo_title = ${a.seoTitle === null ? null : String(a.seoTitle)}`);
if (a.seoDescription !== undefined) sets.push(sql`seo_description = ${a.seoDescription === null ? null : String(a.seoDescription)}`);
if (a.elementorJson !== undefined) {
const base = (cur.meta ?? {}) as Record<string, unknown>;
if (a.elementorJson === null) delete base.elementor;
else {
try {
base.elementor = JSON.parse(String(a.elementorJson));
} catch {
throw new Error("elementorJson 不是合法 JSON");
}
}
sets.push(sql`meta = ${sql.json(base)}::jsonb`);
}
await sql`update contents set ${commaJoin(sets)} where id = ${id}`;
await audit("content.update", "content", id, { fields: Object.keys(a).filter((k) => k !== "id"), via: "mcp" });
return { ok: true, id };
}
case "delete_content": {
const id = Number(a.id);
await sql`delete from content_terms where content_id = ${id}`;
await sql`delete from content_revisions where content_id = ${id}`;
await sql`delete from contents where id = ${id}`;
await audit("content.delete", "content", id, { via: "mcp" });
return { ok: true };
}
case "list_revisions":
return await sql`select id, content_id, title, author_id, created_at from content_revisions where content_id = ${Number(a.contentId)} order by id desc limit 50`;
case "restore_revision": {
const [rev] = await sql`select * from content_revisions where id = ${Number(a.id)}`;
if (!rev) throw new Error(`修订 #${a.id} 不存在`);
const [cur] = await sql`select title, html from contents where id = ${rev.content_id}`;
if (cur) await sql`insert into content_revisions (content_id, title, html, author_id) values (${rev.content_id}, ${cur.title}, ${cur.html}, null)`;
await sql`update contents set title = ${rev.title}, html = ${rev.html}, updated_at = ${new Date()} where id = ${rev.content_id}`;
await audit("content.revision.restore", "content", rev.content_id, { revisionId: rev.id, via: "mcp" });
return { ok: true, contentId: rev.content_id };
}
case "list_media": {
const page = Math.max(1, Number(a.page) || 1);
const pageSize = Math.min(100, Number(a.pageSize) || 20);
const where = a.q ? sql` where path ilike ${"%" + a.q + "%"}` : sql``;
const items = await sql`select id, path, url, mime_type, width, height, size, alt_text, created_at from media${where} order by id desc limit ${pageSize} offset ${(page - 1) * pageSize}`;
const [{ total }] = await sql`select count(*) as total from media${where}`;
return { items, total: Number(total), page, pageSize };
}
case "upload_media": {
const buf = Buffer.from(String(a.dataBase64 || ""), "base64");
if (!buf.length) throw new Error("空文件");
if (buf.length > 20 * 1024 * 1024) throw new Error("文件超过 20MB 上限");
const now = new Date();
const dir = `${now.getFullYear()}/${String(now.getMonth() + 1).padStart(2, "0")}`;
const clean = String(a.filename || `file-${Date.now()}`).replace(/[\\/]+/g, "-").replace(/[^a-zA-Z0-9._\-\u4e00-\u9fa5]+/g, "-").replace(/^-+|-+$/g, "").slice(0, 160) || `file-${Date.now()}`;
const dot = clean.lastIndexOf(".");
const stem = dot === -1 ? clean : clean.slice(0, dot);
const ext = dot === -1 ? "" : clean.slice(dot).toLowerCase();
const fs = await import("node:fs/promises");
const path = await import("node:path");
const root = process.env.DUSUN_ROOT || path.resolve(import.meta.dir, "..");
const absDir = path.join(root, "public", "uploads", dir);
await fs.mkdir(absDir, { recursive: true });
let rel = `${dir}/${stem}${ext}`;
for (let n = 1; n < 1000; n++) {
try {
await fs.access(path.join(root, "public", "uploads", rel));
rel = `${dir}/${stem}-${n}${ext}`;
} catch {
break;
}
}
await fs.writeFile(path.join(root, "public", "uploads", rel), buf);
const dims = parseImageSize(new Uint8Array(buf.slice(0, Math.min(buf.length, 1024 * 1024))));
const [row] = await sql`
insert into media (path, url, mime_type, width, height, size)
values (${rel}, ${"/uploads/" + rel}, ${String(a.mime || "application/octet-stream")},
${dims?.[0] ?? null}, ${dims?.[1] ?? null}, ${buf.length})
returning id`;
await audit("media.upload", "media", row.id, { path: rel, via: "mcp" });
return { ok: true, id: row.id, url: `/uploads/${rel}`, width: dims?.[0] ?? null, height: dims?.[1] ?? null };
}
case "list_products": {
const page = Math.max(1, Number(a.page) || 1);
const pageSize = Math.min(100, Number(a.pageSize) || 20);
const where = a.q ? sql` where (title ilike ${"%" + a.q + "%"} or sku ilike ${"%" + a.q + "%"} or slug ilike ${"%" + a.q + "%"})` : sql``;
const items = await sql`select id, sku, slug, title, price, regular_price, stock, stock_status, status, featured, updated_at from products${where} order by updated_at desc limit ${pageSize} offset ${(page - 1) * pageSize}`;
const [{ total }] = await sql`select count(*) as total from products${where}`;
return { items, total: Number(total), page, pageSize };
}
case "list_orders": {
const page = Math.max(1, Number(a.page) || 1);
const pageSize = Math.min(100, Number(a.pageSize) || 20);
const where = a.status ? sql` where status = ${a.status}` : sql``;
const items = await sql`select id, order_number, status, currency, total, payment_status, billing, created_at from orders${where} order by created_at desc limit ${pageSize} offset ${(page - 1) * pageSize}`;
const [{ total }] = await sql`select count(*) as total from orders${where}`;
return { items, total: Number(total), page, pageSize };
}
case "update_order_status": {
const id = Number(a.id);
const status = String(a.status);
if (!["pending", "on-hold", "processing", "completed", "cancelled", "refunded"].includes(status)) throw new Error("非法订单状态");
await sql`update orders set status = ${status}, updated_at = ${new Date()} where id = ${id}`;
await audit("order.status", "order", id, { status, via: "mcp" });
return { ok: true };
}
case "list_inquiries": {
const page = Math.max(1, Number(a.page) || 1);
const pageSize = Math.min(100, Number(a.pageSize) || 20);
const conds: SQL[] = [];
if (a.status) conds.push(sql`status = ${a.status}`);
if (a.q) conds.push(sql`(name ilike ${"%" + a.q + "%"} or email ilike ${"%" + a.q + "%"} or company ilike ${"%" + a.q + "%"})`);
const where = conds.length ? sql` where ${andJoin(conds)}` : sql``;
const items = await sql`select id, name, email, company, country, product_interest, status, created_at from inquiries${where} order by created_at desc limit ${pageSize} offset ${(page - 1) * pageSize}`;
const [{ total }] = await sql`select count(*) as total from inquiries${where}`;
return { items, total: Number(total), page, pageSize };
}
case "update_inquiry_status": {
const id = Number(a.id);
const status = String(a.status);
if (!["new", "read", "replied", "spam", "archived"].includes(status)) throw new Error("非法询盘状态");
await sql`update inquiries set status = ${status} where id = ${id}`;
await audit("inquiry.status", "inquiry", id, { status, via: "mcp" });
return { ok: true };
}
case "get_setting": {
const [row] = await sql`select key, value, updated_at from settings where key = ${String(a.key)}`;
return row ?? null;
}
case "set_setting": {
const key = String(a.key);
if (!/^[a-z0-9_-]{1,128}$/i.test(key)) throw new Error("非法设置键名");
await sql`insert into settings (key, value, updated_at) values (${key}, ${sql.json(a.value)}::jsonb, ${new Date()})
on conflict (key) do update set value = ${sql.json(a.value)}::jsonb, updated_at = ${new Date()}`;
await audit("settings.update", "settings", key, { via: "mcp" });
return { ok: true };
}
default:
throw new Error(`未知工具:${name}`);
}
}
/* ============================= MCP stdio 协议层 ============================= */
/** # 函数 写一帧 JSON-RPC 响应到 stdout */
function send(msg: Record<string, unknown>) {
process.stdout.write(JSON.stringify(msg) + "\n");
}
/** # 类型 已初始化标记(initialize 之前拒绝 tools/* 调用) **/
let initialized = false;
/** # 函数 逐行解析 stdin 请求并分发(initialize / tools/list / tools/call / ping) */
for await (const chunk of Bun.stdin.stream()) {
const text = new TextDecoder().decode(chunk);
for (const line of text.split("\n")) {
if (!line.trim()) continue;
let req: { id?: unknown; method?: string; params?: Record<string, unknown> };
try {
req = JSON.parse(line);
} catch {
continue;
}
const id = req.id;
try {
switch (req.method) {
case "initialize":
send({
jsonrpc: "2.0",
id,
result: {
protocolVersion: "2024-11-05",
capabilities: { tools: {} },
serverInfo: { name: "dusuniot-cms", version: process.env.npm_package_version ?? "0.11.0" },
},
});
break;
case "notifications/initialized":
initialized = true;
break;
case "ping":
send({ jsonrpc: "2.0", id, result: {} });
break;
case "tools/list":
send({ jsonrpc: "2.0", id, result: { tools: TOOLS } });
break;
case "tools/call": {
if (!initialized) throw new Error("server not initialized");
const name = String((req.params as { name?: string }).name);
const args = ((req.params as { arguments?: Record<string, unknown> }).arguments ?? {}) as Record<string, unknown>;
try {
const result = await callTool(name, args);
send({ jsonrpc: "2.0", id, result: { content: [{ type: "text", text: JSON.stringify(result, null, 2) }] } });
} catch (e) {
// 工具级错误按 isError 返回(协议要求不中断会话)
send({
jsonrpc: "2.0",
id,
result: { content: [{ type: "text", text: String(e instanceof Error ? e.message : e) }], isError: true },
});
}
break;
}
default:
if (id !== undefined) send({ jsonrpc: "2.0", id, error: { code: -32601, message: `method not found: ${req.method}` } });
}
} catch (e) {
if (id !== undefined) send({ jsonrpc: "2.0", id, error: { code: -32603, message: String(e instanceof Error ? e.message : e) } });
}
}
}