适用于您的 CMS 的 MCP 服务器:让 AI 代理读写内容
我在三个月内为客户项目构建了 MCP 服务器,现在我确信这就是每个 CMS 在两年内的访问方式。不是通过仪表板。不是通过人类手动调用的 REST API。而是通过理解你的内容模型并能代表你读取、创建和更新内容的 AI 智能体。
模型上下文协议(MCP)是 Anthropic 在 2024 年末发布的开放标准。它为 AI 模型提供了一种结构化的方式来与外部工具和数据源交互。把它想象成 LLM 和任何东西之间的通用适配器。你的数据库。你的文件系统。你的 CMS。
这篇文章介绍了为三个流行的 CMS 后端构建 MCP 服务器的过程:Payload CMS、Supabase(用作 headless CMS)和 WordPress。我们将涵盖架构、实际代码,以及我沿途遇到的陷阱。
目录
- 什么是 MCP 以及为什么你应该关注
- CMS 集成的 MCP 架构
- 为 Payload CMS 构建 MCP 服务器
- 为 Supabase 构建 MCP 服务器
- 为 WordPress 构建 MCP 服务器
- 比较三种方法
- 安全考虑
- 实际用例
- 常见问题
什么是 MCP 以及为什么你应该关注
模型上下文协议本质上是一个基于 JSON-RPC 2.0 的协议,定义了 AI 模型如何与外部系统通信。在 MCP 之前,如果你想让 Claude 或 GPT 与你的 CMS 交互,你需要为每个模型构建一个自定义的函数调用集成。不同的架构、不同的身份验证流程、不同的一切。
MCP 标准化了这一点。你构建一个服务器,任何 MCP 兼容的客户端(Claude Desktop、Cursor、Cline、自定义智能体)都可以连接到它。
以下是为什么这对 CMS 工作特别有趣:
- 内容团队可以要求 AI "查找上个月发布的所有提到 React 的博客文章"并从他们实际的 CMS 获取真实结果
- 开发人员可以让他们的编码助手创建和更新内容条目,无需离开编辑器
- 编辑工作流可以自动化——一个 AI 智能体起草内容、根据你的风格指南检查,并通过你现有的工作流程发布
该协议定义了三个基本元素:
- Tools(工具)—— AI 可以调用的函数(如
create_post、update_entry、search_content) - Resources(资源)—— AI 可以读取的数据(如你的内容架构、最近的文章、媒体库)
- Prompts(提示)—— 用于常见操作的可重用提示模板
对于 CMS 集成,你主要会使用 Tools 和 Resources。
CMS 集成的 MCP 架构
在我们编写代码之前,让我们理解架构。MCP 服务器位于 AI 客户端和你的 CMS 之间:
[AI 客户端] <--MCP 协议--> [MCP 服务器] <--REST/GraphQL/SDK--> [你的 CMS]
(Claude) (stdio/SSE) (Node.js) (HTTP/DB) (Payload/WP/Supabase)
MCP 服务器公开映射到你的 CMS 操作的工具和资源。传输层可以是:
- stdio —— 用于本地开发,AI 客户端将 MCP 服务器生成为子进程
- SSE(服务器发送事件) —— 对于远程服务器,使用带有 SSE 流的 HTTP
- 可流式 HTTP —— 2025 规范修订中的新传输选项
对于 CMS 服务器,我建议从 stdio 开始进行开发,然后过渡到 SSE 或可流式 HTTP 用于生产部署,其中 MCP 服务器与 CMS 一起在你的基础设施上运行。
基本服务器框架
每个 MCP 服务器都遵循相同的模式。以下是使用官方 @modelcontextprotocol/sdk 的 TypeScript 框架:
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const server = new McpServer({
name: "my-cms-mcp",
version: "1.0.0",
});
// 定义一个工具
server.tool(
"search_content",
"Search for content entries by query",
{
query: z.string().describe("Search query"),
collection: z.string().describe("Collection/post type to search"),
limit: z.number().optional().default(10),
},
async ({ query, collection, limit }) => {
// 你的 CMS 特定逻辑在这里
const results = await searchCMS(query, collection, limit);
return {
content: [
{
type: "text",
text: JSON.stringify(results, null, 2),
},
],
};
}
);
// 定义一个资源
server.resource(
"schema",
"cms://schema",
async (uri) => ({
contents: [
{
uri: uri.href,
mimeType: "application/json",
text: JSON.stringify(await getCMSSchema()),
},
],
})
);
const transport = new StdioServerTransport();
await server.connect(transport);
这是基础。现在让我们构建真实的实现。
为 Payload CMS 构建 MCP 服务器
Payload 是我最喜欢的用于 headless 项目的 CMS(我们在我们的 headless CMS 开发工作 中大量使用它)。它是 TypeScript 原生的,具有出色的 API,其基于集合的架构完美映射到 MCP 工具。
设置 Payload MCP 服务器
首先,安装依赖项:
mkdir payload-mcp-server && cd payload-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node tsx
这是完整的实现。Payload 的 REST API 足够干净,使这个过程出奇地简单:
// src/index.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const PAYLOAD_URL = process.env.PAYLOAD_URL || "http://localhost:3000";
const PAYLOAD_API_KEY = process.env.PAYLOAD_API_KEY || "";
const headers = {
"Content-Type": "application/json",
Authorization: `users API-Key ${PAYLOAD_API_KEY}`,
};
async function payloadFetch(path: string, options: RequestInit = {}) {
const res = await fetch(`${PAYLOAD_URL}/api${path}`, {
...options,
headers: { ...headers, ...options.headers },
});
if (!res.ok) {
throw new Error(`Payload API error: ${res.status} ${await res.text()}`);
}
return res.json();
}
const server = new McpServer({
name: "payload-cms-mcp",
version: "1.0.0",
});
// 列出所有集合
server.tool(
"list_collections",
"List all available Payload CMS collections",
{},
async () => {
// Payload 默认没有元端点用于此,
// 所以我们访问配置或硬编码已知集合
const collections = ["posts", "pages", "media", "categories"];
return {
content: [{ type: "text", text: JSON.stringify(collections) }],
};
}
);
// 在集合中搜索/列出文档
server.tool(
"find_documents",
"Find documents in a Payload collection with optional filtering",
{
collection: z.string().describe("Collection slug (e.g., 'posts', 'pages')"),
where: z.string().optional().describe("Payload where query as JSON string"),
limit: z.number().optional().default(10),
page: z.number().optional().default(1),
sort: z.string().optional().describe("Field to sort by, prefix with - for desc"),
},
async ({ collection, where, limit, page, sort }) => {
const params = new URLSearchParams();
params.set("limit", String(limit));
params.set("page", String(page));
if (sort) params.set("sort", sort);
if (where) {
try {
const whereObj = JSON.parse(where);
// 转换为 Payload 查询字符串格式
for (const [key, val] of Object.entries(whereObj)) {
if (typeof val === "object" && val !== null) {
for (const [op, v] of Object.entries(val as Record<string, unknown>)) {
params.set(`where[${key}][${op}]`, String(v));
}
} else {
params.set(`where[${key}][equals]`, String(val));
}
}
} catch {
return { content: [{ type: "text", text: "Invalid where JSON" }] };
}
}
const data = await payloadFetch(`/${collection}?${params}`);
return {
content: [{ type: "text", text: JSON.stringify(data, null, 2) }],
};
}
);
// 获取单个文档
server.tool(
"get_document",
"Get a single document by ID from a Payload collection",
{
collection: z.string(),
id: z.string().describe("Document ID"),
},
async ({ collection, id }) => {
const data = await payloadFetch(`/${collection}/${id}`);
return {
content: [{ type: "text", text: JSON.stringify(data, null, 2) }],
};
}
);
// 创建文档
server.tool(
"create_document",
"Create a new document in a Payload collection",
{
collection: z.string(),
data: z.string().describe("Document data as JSON string"),
},
async ({ collection, data }) => {
const result = await payloadFetch(`/${collection}`, {
method: "POST",
body: data,
});
return {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
};
}
);
// 更新文档
server.tool(
"update_document",
"Update an existing document in a Payload collection",
{
collection: z.string(),
id: z.string(),
data: z.string().describe("Partial document data as JSON string"),
},
async ({ collection, id, data }) => {
const result = await payloadFetch(`/${collection}/${id}`, {
method: "PATCH",
body: data,
});
return {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
};
}
);
const transport = new StdioServerTransport();
await server.connect(transport);
连接到 Claude Desktop
将此添加到你的 Claude Desktop 配置(Mac 上的 ~/Library/Application Support/Claude/claude_desktop_config.json):
{
"mcpServers": {
"payload-cms": {
"command": "npx",
"args": ["tsx", "/path/to/payload-mcp-server/src/index.ts"],
"env": {
"PAYLOAD_URL": "http://localhost:3000",
"PAYLOAD_API_KEY": "your-api-key-here"
}
}
}
}
重新启动 Claude Desktop,你现在可以问它诸如"查找我的 CMS 中的所有草稿文章"或"创建一篇关于 MCP 服务器的新博客文章"之类的问题。
为 Supabase 构建 MCP 服务器
Supabase 很有趣,因为它开箱即用不是 CMS——它是一个具有 REST API 的 Postgres 数据库。但很多团队将其用作一个,特别是使用 Next.js 构建的自定义管理面板(这是我们在 Next.js 开发实践 中经常做的事情)。
Supabase MCP 服务器实际上是三个中最强大的,因为你可以获得直接的 SQL 访问。
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { createClient } from "@supabase/supabase-js";
import { z } from "zod";
const supabase = createClient(
process.env.SUPABASE_URL!,
process.env.SUPABASE_SERVICE_ROLE_KEY! // 使用服务角色以供服务器端使用
);
const server = new McpServer({
name: "supabase-cms-mcp",
version: "1.0.0",
});
// 列出所有表(内容类型)
server.tool(
"list_tables",
"List all tables in the public schema",
{},
async () => {
const { data, error } = await supabase.rpc("get_tables");
// 你需要一个 Postgres 函数,或使用:
const { data: tables } = await supabase
.from("information_schema.tables" as any)
.select("table_name")
.eq("table_schema", "public");
return {
content: [{ type: "text", text: JSON.stringify(tables, null, 2) }],
};
}
);
// 使用过滤器查询内容
server.tool(
"query_content",
"Query rows from a table with optional filters",
{
table: z.string().describe("Table name"),
select: z.string().optional().default("*").describe("Columns to select"),
filters: z.array(z.object({
column: z.string(),
operator: z.enum(["eq", "neq", "gt", "lt", "gte", "lte", "like", "ilike", "is"]),
value: z.string(),
})).optional().describe("Array of filter conditions"),
limit: z.number().optional().default(20),
orderBy: z.string().optional(),
ascending: z.boolean().optional().default(false),
},
async ({ table, select, filters, limit, orderBy, ascending }) => {
let query = supabase.from(table).select(select).limit(limit);
if (filters) {
for (const f of filters) {
query = query.filter(f.column, f.operator, f.value);
}
}
if (orderBy) {
query = query.order(orderBy, { ascending });
}
const { data, error } = await query;
if (error) {
return { content: [{ type: "text", text: `Error: ${error.message}` }] };
}
return {
content: [{ type: "text", text: JSON.stringify(data, null, 2) }],
};
}
);
// 插入内容
server.tool(
"insert_row",
"Insert a new row into a table",
{
table: z.string(),
data: z.string().describe("Row data as JSON string"),
},
async ({ table, data: rowData }) => {
const parsed = JSON.parse(rowData);
const { data, error } = await supabase.from(table).insert(parsed).select();
if (error) {
return { content: [{ type: "text", text: `Error: ${error.message}` }] };
}
return {
content: [{ type: "text", text: JSON.stringify(data, null, 2) }],
};
}
);
// 全文搜索(Supabase 针对 CMS 使用的杀手级功能)
server.tool(
"full_text_search",
"Perform full-text search on a table column",
{
table: z.string(),
column: z.string().describe("Column with tsvector index"),
query: z.string().describe("Search query"),
limit: z.number().optional().default(10),
},
async ({ table, column, query, limit }) => {
const { data, error } = await supabase
.from(table)
.select()
.textSearch(column, query)
.limit(limit);
if (error) {
return { content: [{ type: "text", text: `Error: ${error.message}` }] };
}
return {
content: [{ type: "text", text: JSON.stringify(data, null, 2) }],
};
}
);
const transport = new StdioServerTransport();
await server.connect(transport);
我喜欢 Supabase 方法的一点:因为 Supabase 公开了你的 Postgres 架构,你可以创建一个资源,为 AI 提供你的完整架构定义。AI 然后确切地知道存在哪些字段、它们的类型是什么,以及表如何相互关联。这使其查询更加准确。
为 WordPress 构建 MCP 服务器
WordPress。仍在 2025 年为网络的 43% 提供支持。爱它或讨厌它,你都会遇到它。
WordPress REST API 实际上对 MCP 集成很有效。棘手的部分是身份验证——WordPress 有大约五种不同的身份验证方法,其中没有一种都很好。
对于 MCP 服务器,我建议应用程序密码(自 WordPress 5.6 以来内置):
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const WP_URL = process.env.WP_URL!;
const WP_USER = process.env.WP_USER!;
const WP_APP_PASSWORD = process.env.WP_APP_PASSWORD!;
const authHeader = `Basic ${Buffer.from(`${WP_USER}:${WP_APP_PASSWORD}`).toString("base64")}`;
async function wpFetch(endpoint: string, options: RequestInit = {}) {
const res = await fetch(`${WP_URL}/wp-json/wp/v2${endpoint}`, {
...options,
headers: {
"Content-Type": "application/json",
Authorization: authHeader,
...options.headers,
},
});
if (!res.ok) {
throw new Error(`WP API error: ${res.status} ${await res.text()}`);
}
return res.json();
}
const server = new McpServer({
name: "wordpress-mcp",
version: "1.0.0",
});
server.tool(
"search_posts",
"Search WordPress posts",
{
search: z.string().optional().describe("Search query"),
status: z.enum(["publish", "draft", "pending", "private", "any"]).optional().default("any"),
per_page: z.number().optional().default(10),
categories: z.array(z.number()).optional(),
tags: z.array(z.number()).optional(),
after: z.string().optional().describe("ISO 8601 date -- only posts after this date"),
before: z.string().optional().describe("ISO 8601 date -- only posts before this date"),
},
async (params) => {
const searchParams = new URLSearchParams();
if (params.search) searchParams.set("search", params.search);
if (params.status) searchParams.set("status", params.status);
searchParams.set("per_page", String(params.per_page));
if (params.categories) searchParams.set("categories", params.categories.join(","));
if (params.tags) searchParams.set("tags", params.tags.join(","));
if (params.after) searchParams.set("after", params.after);
if (params.before) searchParams.set("before", params.before);
const posts = await wpFetch(`/posts?${searchParams}`);
// 从内容中剥离 HTML 以获得更清洁的 AI 消费
const cleaned = posts.map((p: any) => ({
id: p.id,
title: p.title.rendered,
slug: p.slug,
status: p.status,
date: p.date,
excerpt: p.excerpt.rendered.replace(/<[^>]*>/g, ""),
content: p.content.rendered.replace(/<[^>]*>/g, ""),
categories: p.categories,
tags: p.tags,
}));
return {
content: [{ type: "text", text: JSON.stringify(cleaned, null, 2) }],
};
}
);
server.tool(
"create_post",
"Create a new WordPress post",
{
title: z.string(),
content: z.string().describe("Post content in HTML"),
status: z.enum(["publish", "draft", "pending"]).optional().default("draft"),
categories: z.array(z.number()).optional(),
tags: z.array(z.number()).optional(),
excerpt: z.string().optional(),
},
async (params) => {
const result = await wpFetch("/posts", {
method: "POST",
body: JSON.stringify(params),
});
return {
content: [{
type: "text",
text: `Created post "${result.title.rendered}" (ID: ${result.id}, Status: ${result.status})`,
}],
};
}
);
server.tool(
"update_post",
"Update an existing WordPress post",
{
id: z.number().describe("Post ID"),
title: z.string().optional(),
content: z.string().optional(),
status: z.enum(["publish", "draft", "pending", "trash"]).optional(),
excerpt: z.string().optional(),
},
async ({ id, ...updates }) => {
const result = await wpFetch(`/posts/${id}`, {
method: "POST",
body: JSON.stringify(updates),
});
return {
content: [{
type: "text",
text: `Updated post "${result.title.rendered}" (ID: ${result.id})`,
}],
};
}
);
const transport = new StdioServerTransport();
await server.connect(transport);
WordPress MCP 最大的陷阱:内容返回为渲染的 HTML。你会想要剥离标签供 AI 的上下文窗口使用,否则你会在阅读五篇博客文章时浪费 50k 个令牌来处理 <p> 标签。我以艰苦的方式学到了这一点。
比较三种方法
| 特性 | Payload CMS | Supabase | WordPress |
|---|---|---|---|
| 身份验证复杂性 | 简单(API 密钥) | 简单(服务角色密钥) | 中等(应用程序密码) |
| 架构意识 | 良好(类型化集合) | 优秀(完整 Postgres 架构) | 差(无架构端点) |
| 内容格式 | 干净 JSON | 干净 JSON | HTML 重,需要清理 |
| 写入支持 | 优秀 | 优秀 | 良好(块有一些怪癖) |
| 媒体处理 | 通过 API 上传 | 存储 API | REST 媒体端点 |
| 查询灵活性 | 良好(Payload 查询) | 优秀(SQL 级过滤) | 有限(WP_Query 参数) |
| 自托管选项 | 是 | 是(或云) | 是 |
| 典型设置时间 | 1-2 小时 | 1-2 小时 | 2-3 小时 |
| 最适合 | 现代 headless 项目 | 自定义内容结构 | 现有 WP 网站 |
安全考虑
这是我几乎忘记写的部分,而且它可能是最重要的部分。
当你给 AI 智能体写入 CMS 的访问权限时,你信任它不会删除所有内容。以下是如何不被烧伤的方法:
最小权限原则
为 MCP 服务器创建一个专用 API 用户/密钥,权限最少。在 Payload 中,创建一个只能访问特定集合的角色。在 Supabase 中,使用行级安全策略。在 WordPress 中,创建一个具有作者角色的用户,而不是管理员。
为破坏性操作添加确认
我总是为 create/update/delete 工具添加一个 dryRun 参数。设置为 true 时,该工具返回会发生什么,而无需实际执行。AI 然后可以在继续之前与用户确认。
server.tool(
"delete_document",
"Delete a document (use dryRun first!)",
{
collection: z.string(),
id: z.string(),
dryRun: z.boolean().default(true).describe("If true, only preview the deletion"),
},
async ({ collection, id, dryRun }) => {
const doc = await payloadFetch(`/${collection}/${id}`);
if (dryRun) {
return {
content: [{
type: "text",
text: `DRY RUN: Would delete "${doc.title}" (${id}) from ${collection}`,
}],
};
}
await payloadFetch(`/${collection}/${id}`, { method: "DELETE" });
return {
content: [{ type: "text", text: `Deleted "${doc.title}" (${id})` }],
};
}
);
速率限制和审计日志
记录每个工具调用。认真对待。当你的内容编辑在凌晨 2 点问为什么有 47 篇关于企鹅的新草稿文章时,你会想知道你的 AI 智能体做了什么。
永远不要直接公开数据库凭据
MCP 服务器是边界。AI 永远看不到你的数据库密码或 API 密钥——它只看到你公开的工具。
实际用例
以下是我在生产环境中看到的效果不错的模式:
内容迁移:一个 AI 智能体,它从遗留 WordPress 网站读取内容,将其转换以匹配你的新 Payload 架构,并在新 CMS 中创建条目。我们为迁移约 3,000 篇文章的客户做过这个。需要数周手动工作的工作在一天半内通过人工审核完成。
SEO 审核:一个 AI 读取所有已发布内容,检查缺失的元描述、内容不足和损坏的内部链接。将其发现写回为 CMS 中的结构化报告。
批量内容更新:「在所有已发布页面中将所有对'我们的旧品牌名称'的引用更改为'我们的新品牌名称'。」该智能体读取、查找、提议更改,并通过人工批准进行更新。
内容日历管理:一个了解你的发布时间表的智能体,检查草稿中的内容,并通过读取 CMS 数据为即将到来的截止日期提醒团队。
如果你正在构建类似的东西并需要帮助,我们在 headless CMS 实践 中对这三个平台都有深刻的经验。你也可以查看我们的 定价页面 以了解这样的项目成本。
常见问题
什么是模型上下文协议(MCP)? MCP 是 Anthropic 创建的开放协议,标准化 AI 模型与外部数据源和工具的交互方式。它使用 JSON-RPC 2.0,定义了 AI 应用程序(客户端)连接到公开工具、资源和提示的服务器的客户端-服务器架构。把它想象成 AI 的 USB-C 端口——一个标准接口连接到许多不同的系统。
我可以与 ChatGPT 一起使用 MCP 还是只有 Claude? MCP 由 Anthropic 创建,所以 Claude 通过 Claude Desktop 和 API 拥有最佳的原生支持。但是该协议是开放的,其他平台也存在 MCP 客户端。OpenAI 在 2025 年 3 月宣布了 MCP 支持。Cursor、Cline、Windsurf 和其他几个开发人员工具也支持 MCP。你的服务器与任何兼容的客户端合作。
给 AI 智能体写入 CMS 的权限是否安全? 通过适当的防护措施可以。使用专用的 API 凭据,权限最少,为破坏性操作实现干运行模式,添加审计日志,并考虑对发布操作要求人工批准。永远不要给 MCP 服务器管理员级访问权限。像对待任何第三方集成一样对待它——具有适当的信任边界。
我需要与我的 CMS 分开托管 MCP 服务器吗? 对于 stdio 传输(本地开发),MCP 服务器作为你本地机器上的子进程运行。对于具有远程访问的生产环境,你需要在某处托管它——这可能在与你的 CMS 相同的服务器上、在与其相邻的 Docker 容器中,或作为单独的服务。MCP 服务器和你的 CMS 之间的延迟很重要,所以并置它们是理想的。
MCP 与仅使用带有函数调用的 CMS REST API 相比如何? MCP 添加了一个标准接口层。使用原始函数调用,你为每个 AI 提供程序定义自定义架构,为每个处理不同的身份验证,并在提供程序更改其 API 时重建。MCP 为你提供一个与任何兼容的客户端合作的服务器。它还以标准化方式处理传输、错误格式和资源发现。对于单个集成,开销可能不值得。对于你将重新使用的任何东西,它绝对值得。
哪个 CMS 最适合 MCP? Supabase 提供了最大的灵活性,因为你拥有直接的数据库级访问和完整的架构内省。Payload CMS 很好,因为其 TypeScript 优先的方法和干净的 REST API 自然映射到 MCP 工具。WordPress 可以很好地工作,但需要更多的 HTML 内容清理和较少灵活的查询 API。"最佳"选择取决于你现有的堆栈。
我可以使用 MCP 在我的 CMS 中管理媒体上传吗?
是的,虽然涉及的比文本内容更多。对于 Payload 和 WordPress,你可以创建接受 base64 编码文件数据或 URL 的工具,并使用相应的上传 API。对于 Supabase,你会使用存储 API。我建议创建单独的 upload_media 和 attach_media 工具,而不是尝试在内容创建工具中处理文件上传。
运行 MCP 服务器需要多少费用? MCP 服务器本身是轻量级的——一个简单的 Node.js 进程,消耗最少的资源。真正的成本在于你的智能体读取和写入内容时消耗的 AI 令牌。典型的内容操作可能使用 2,000-10,000 个令牌,具体取决于处理多少内容。在当前 2025 年定价(Claude Sonnet 在 $3/$15 每百万输入/输出令牌),这是每个操作的几分之一分。托管 MCP 服务器的基础设施成本可以忽略不计——一个小 VPS 或容器实例绰绰有余。