做内容平台的老开发应该都有同感内容管理、发布、审核、检索、推荐这些模块本身写起来并不难难的是它们会反复出现在不同业务线里而且一旦牵扯到付费解锁、多商户分账代码就开始变得特别容易失控。再加上现在越来越多的 AI 能力要接入内容系统比如自动生成摘要、智能打标、敏感词初筛、私域推荐如果每个能力都单独对接一次后期维护成本会指数级上升。我们在实际项目里换了一条思路用 MCP 协议把内容能力封装成标准化工具再在业务后端统一调用整体架构清晰了很多。这篇文章会把整个方案的架构设计、数据模型、核心代码、常见问题和工程建议完整写出来适合正在开发内容平台、付费阅读系统或多商户 CMS 的同学参考。1. 内容系统为什么需要 MCP背景与核心概念1.1 从内容平台开发痛点说起很多人第一次接触“MCP 内容系统开发”这个说法时会有点困惑MCP 不是用来做大模型工具调用的吗跟内容系统、付费系统、多商户系统有什么关系实际上MCP 在这里解决的是一个非常现实的问题也就是“内容能力复用”。先看一个典型业务场景。假设平台内有多个角色运营可以发布平台公告商户可以发布自己的付费专栏用户可以购买内容并阅读管理员需要对所有内容做合规审核。传统做法是把内容表设计成一张大表用content_type区分栏目用merchant_id区分商户再用一个权限系统控制谁能操作。这套方案初期没问题但到了中期就会面临几个痛点第一AI 辅助写稿、AI 摘要、AI 标签、内容审核这些功能都要写独立的调用逻辑第二商户扩展越来越多每个商户都可能需要自己的内容管理工具第三付费内容和免费内容混在同一套逻辑里订单状态、退款、分账全都耦合在一起。MCP 的切入点就在这里。MCP 全称是 Model Context Protocol也就是模型上下文协议它由 Anthropic 提出目标是让 AI 模型能够通过统一协议去连接外部数据源和工具。你可以把 MCP 理解成一套“AI 世界里的 USB 接口”只要外部系统按照这个协议暴露能力AI 模型就能直接使用不需要为每个系统单独写一套对接代码。放到内容系统里我们就能把“创建内容”“检索内容”“审核内容”“推荐内容”这类能力统一封装成 MCP 工具业务系统和大模型都可以通过同一种方式调用。1.2 什么是 MCP 协议MCP 协议的核心设计是围绕三种原语展开的Tool、Resource、Prompt。Tool一个可执行的动作比如“创建内容”“查询订单”“审核文章”。它需要明确的入参和出参类似函数调用。Resource一份可被读取的数据比如“商户资料”“价格配置”“用户阅读权限”。它对外提供的是数据上下文而不是执行动作。Prompt一段预定义提示词模板方便在 AI 场景里复用固定话术比如“内容摘要生成模板”“合规审核提示词”。在传输方式上MCP 支持本地 stdio标准输入输出和远程 HTTP/SSE 两种模式。本地模式适合把 MCP Server 和业务系统跑在同一个进程环境里调试方便远程模式适合在分布式场景下把 MCP Server 部署成独立服务。和市面上真正的网络代理类工具不同MCP 是一种应用层协议它本身不涉及网络出口代理也不涉及任何网络穿透操作。它解决的是“AI 模型如何调用工具”这个工程问题。所以做 MCP 内容系统开发时完全可以把注意力集中在业务工具设计上不要被相关术语混淆。1.3 付费内容系统与多商户系统的概念边界要设计好这个系统先要把三个概念分开理解。内容系统负责内容的创建、编辑、审核、发布、下架、检索、展示。核心是“内容状态机”和“内容版本管理”。付费内容系统在内容系统之上增加交易能力包括试读、解锁、订单、支付回调、退款、阅读权限校验。核心是“订单状态机”和“用户权益校验”。多商户系统在内容系统之上增加商户租户隔离能力每个商户可以独立管理自己的内容和价格平台方可以按比例抽成。核心是“数据隔离”和“分账结算”。这三个概念是层层叠加的关系。多商户系统不一定有付费功能付费系统不一定是多商户但在这篇文章的实战场景里我们会把三者合并一个多商户内容平台商户发布内容平台允许内容设置免费或付费用户在购买后获得阅读权限平台按比例与商户分账。这个场景对数据一致性要求很高也是最有代表性的综合实战项目。1.4 MCP 为什么适合内容系统开发用传统方式开发这样的多商户付费内容系统通常需要在业务代码里同时维护内容服务、订单服务、商户服务、结算服务并且还要额外处理 AI 功能的接入。而引入 MCP 之后我们可以把内容系统里相对独立、可复用、适合被 AI 或外部系统调用的能力统一封装成 MCP 工具层。这样做有几个明显的好处。第一调用入口统一。不管是运营后台发布内容还是商户批量导入内容或者 AI 自动生成内容并入库最终都是调用同一个“创建内容”工具参数一致、行为一致、校验一致。第二AI 接入成本降低。后续如果要接各种大模型能力只需要让模型具备调用 MCP 工具的能力即可不需要为每个需求写新的服务接口。第三工具与业务解耦。MCP Server 可以独立部署、独立升级业务后端通过标准客户端调用互不影响。当然MCP 也不是银弹。它适合封装“动作清晰、参数明确、返回稳定”的能力不适合承载复杂的事务流程。比如下单支付这种需要强事务和多步骤确认的流程更推荐放在业务服务里做而不是一股脑塞进 MCP 工具。后面在设计工具边界时我们会重点强调这一点。2. 系统总体架构与技术选型2.1 整体架构设计整个系统采用分层架构从上到下可以拆成四层。前端(WEB / 小程序) │ HTTPS ▼ 业务后端 (FastAPI) ├── 商户模块 ├── 内容模块 ├── 订单模块 └── 结算模块 │ MCP 客户端调用 ▼ MCP Server (内容工具 / 检索工具 / 审核工具) │ SQL ▼ MySQL / Redis / 对象存储前端只负责页面展示和交互所有业务规则都通过业务后端接口完成。业务后端内部维护商户、内容、订单、结算等核心模块同时在需要复用内容能力时通过 MCP 客户端调用 MCP Server。MCP Server 负责把内容相关的通用能力封装为工具底层连接 MySQL、Redis 和对象存储。这种架构的好处是如果未来希望大模型直接操作内容系统只需要把 MCP Server 暴露出来让 AI Agent 连接即可如果未来希望其他业务系统也复用内容能力同样可以接入同一套 MCP Server。业务后端不需要改代码。2.2 技术选型和版本说明为了避免版本冲突本文的技术栈尽量选择比较稳定的组合但具体版本仍需要根据你的项目实际情况调整。示例环境以 Python 生态为主。后端框架FastAPI用于实现业务接口。MCP 客户端与服务端MCP Python SDK用于封装和调用内容工具。数据库MySQL 8用于存储商户、内容、订单、结算数据。缓存Redis用于内容热数据和订单幂等键缓存。ORMSQLAlchemy 2.x用于操作 MySQL。对象存储OSS/S3 兼容服务用于存储封面图、正文图片等附件。这里重点说明一下 MCP 的接入方式MCP 协议本身和语言无关官方提供 Python、TypeScript、Java 等多种 SDK。本文以 Python SDK 为例因为 FastAPI 生态写起内容系统来效率很高而且 MCP Python SDK 的 FastMCP 封装非常适合快速暴露工具。2.3 系统模块拆分从工程结构上看系统可以拆成五个核心模块。商户域负责商户注册、商户资料、商户状态、商户密钥管理。内容域负责内容草稿、内容审核、内容发布、内容下架、内容检索。交易域负责付费内容下单、支付回调、订单查询、阅读权限校验。结算域负责平台与商户的分账、提现、对账。MCP 工具域负责把内容域和审核域的能力包装成标准 MCP 工具。其中交易域不建议做成 MCP 工具。原因是支付流程涉及强交互、回调验签、金额计算属于高一致性业务放在业务服务里能更好地控制事务边界。MCP 工具域更适合做内容创建、内容检索、内容推荐、内容审核这类边界清晰的动作。3. 环境准备与工程初始化3.1 环境要求在开始写代码之前建议先准备好以下运行环境。操作系统Windows 10/11、macOS 或 Linux 均可。Python 版本建议 3.10 及以上本文示例使用 Python 3.10 语法。MySQL建议 8.0 及以上需要本地可连接。Redis建议 6.0 及以上用于缓存和幂等控制。Node.js如果后续要跑前端项目建议 18 及以上不是必须项。包管理工具建议使用 pip 和 venv。需要注意MCP 的版本迭代比较快不同版本下 SDK 的 API 会有差异。本文的代码示例以当前主流用法为准如果你的环境中 API 有变化优先查阅你实际安装版本的官方文档。3.2 创建工程目录和虚拟环境我们先创建一个标准工程目录。mkdir -p mcp-content-system/{mcp_server,backend,frontend,docs} cd mcp-content-system python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate激活虚拟环境后创建依赖文件。# requirements.txt fastapi uvicorn[standard] sqlalchemy pymysql redis mcp pydantic然后安装依赖。pip install -r requirements.txt安装完成后我们可以先检查一下关键包是否能正常导入。python -c from mcp.server.fastmcp import FastMCP; print(MCP OK)如果输出MCP OK说明 MCP SDK 已经可用。如果导入失败可能是版本差异问题可以试着调整 MCP SDK 的版本。3.3 数据库和基础配置在 MySQL 中创建系统需要的数据库。CREATE DATABASE IF NOT EXISTS mcp_content_system DEFAULT CHARACTER SET utf8mb4 DEFAULT COLLATE utf8mb4_unicode_ci;在项目根目录下创建配置文件config.py用于保存数据库、Redis、对象存储等连接信息。注意不要把生产环境的密钥直接提交到代码仓库建议通过环境变量注入。# config.py import os class Settings: MYSQL_HOST os.getenv(MYSQL_HOST, 127.0.0.1) MYSQL_PORT int(os.getenv(MYSQL_PORT, 3306)) MYSQL_USER os.getenv(MYSQL_USER, root) MYSQL_PASSWORD os.getenv(MYSQL_PASSWORD, your_password) MYSQL_DB os.getenv(MYSQL_DB, mcp_content_system) REDIS_URL os.getenv(REDIS_URL, redis://127.0.0.1:6379/0) PLATFORM_COMMISSION_RATE float(os.getenv(PLATFORM_COMMISSION_RATE, 0.1)) settings Settings()这里的PLATFORM_COMMISSION_RATE是平台抽成比例示例默认 10%实际项目需要根据自己的商业模式调整。4. MCP Server 核心实现4.1 MCP 工具设计方法论在写 MCP Server 之前先想清楚设计原则。一个合格的 MCP 工具应该满足以下条件。第一一个工具只做一件事。比如“创建内容”和“审核内容”最好拆成两个工具不要让一个工具既落库又审核否则后续很难复用。第二参数必须结构化。所有入参尽可能是明确的字符串、数字、布尔值或 JSON 对象不要用一段自然语言让 AI 自由发挥否则工具调用的稳定性会很差。第三返回结果要有明确结构。建议统一返回 JSON 格式包含status、biz_code、data、message等字段方便上层判断。第四工具描述要写清楚。MCP 是给大模型调用的协议工具描述会直接影响模型是否能正确选择工具所以描述要具体比如“创建一篇内容并返回 articleId”。4.2 实现内容创建工具我们使用 MCP Python SDK 的 FastMCP 封装来编写服务端。这里先创建一个轻量级的内容创建工具它负责校验参数并返回新内容的 ID 和状态。# mcp_server/server.py import time from mcp.server.fastmcp import FastMCP mcp FastMCP( mcp-content-system, instructions面向多商户内容平台提供内容创建、检索、审核等工具能力 ) mcp.tool() def create_article( merchant_id: str, title: str, content: str, is_paid: bool False, price: float 0.0 ) - dict: 创建一篇内容商户ID必传付费内容需要设置价格。 # 实际项目中应写入数据库这里先返回构建结果 article_id ART_ str(int(time.time() * 1000)) return { status: success, article_id: article_id, merchant_id: merchant_id, title: title, content_length: len(content), is_paid: is_paid, price: price, status: draft } if __name__ __main__: mcp.run()在这个示例里create_article接收商户 ID、标题、正文以及是否付费、价格两个可选参数。返回结果里包含内容长度、状态等信息这样调用方可以确认内容是否创建成功。从实际落地的角度create_article内部还应该做好几件事校验商户是否存在并处于可用状态校验正文内容是否为空校验付费内容的价格是否大于 0对内容做敏感词初筛将内容以草稿状态写入数据库。这里为了保持示例简洁用返回结构代替了持久化逻辑读者可以按自己的项目情况补充。4.3 实现内容检索与推荐工具内容平台最常用的能力之一就是检索。我们希望支持按关键词检索、按商户过滤、按付费状态过滤。这个工具适合封装成 MCP 工具因为后续无论是用户端的搜索接口、管理后台的内容列表还是 AI 推荐系统都需要类似的检索能力。# mcp_server/tools/search_tools.py from mcp.server.fastmcp import FastMCP import mcp_server.server as server server.mcp.tool() def search_articles( keyword: str , merchant_id: str , is_paid: bool | None None, limit: int 10 ) - dict: 按关键词检索已发布内容支持按商户和付费状态过滤。 # 实际项目中使用 SQL 查询示例直接返回空结果 return { status: success, total: 0, items: [] }注意这个示例为了调试方便直接返回了空列表。真正落地时这里应该执行 SQL 查询并保证在WHERE条件中强制携带status 2已发布避免把草稿内容检索出来。同时如果当前搜索场景要求商户隔离必须把merchant_id作为必填条件防止串号。检索工具适合做基础实现推荐工具则更复杂。推荐工具通常需要读取用户行为数据、内容标签、热度数据然后组合排序。我们可以把推荐工具单独封装方便后续接入大模型或算法服务。server.mcp.tool() def recommend_articles( user_id: str , limit: int 10 ) - dict: 根据用户行为推荐内容供内容流场景使用。 # 实际项目中可以结合 Redis 热度和用户标签计算 return { status: success, user_id: user_id, items: [] }这个工具的边界很清晰它只负责返回推荐结果不负责用户行为采集。行为采集应该由业务后端在阅读、点赞等事件中完成。4.4 实现内容审核工具内容审核是多商户内容系统里最容易被忽视、但最容易出问题的环节。审核工具可以做两层第一层是本地敏感词过滤第二层是调用专业内容安全服务。专业内容安全服务通常需要申请接口权限本文不展开具体品牌只做一个可以替换实现的占位。# mcp_server/tools/audit_tools.py import mcp_server.server as server server.mcp.tool() def audit_article( article_id: str, title: str, content: str ) - dict: 对标题和正文进行合规初筛返回是否通过。 # 第一层本地敏感词过滤 risk_words [示例违规词A, 示例违规词B] title_hit [w for w in risk_words if w in title] content_hit [w for w in risk_words if w in content] # 第二层专业内容安全服务调用可以写在这里 # external_result call_security_api(title, content) passed len(title_hit) 0 and len(content_hit) 0 return { status: success, article_id: article_id, passed: passed, title_hit: title_hit, content_hit: content_hit }审核工具的返回值非常关键。如果passed为 false业务系统需要决定是把内容打回草稿还是进入人工复审队列。这里的一个最佳实践是机器审核通过不代表完全没问题建议设置“机审通过 人工抽查”的双重机制尤其是涉及付费内容时更要谨慎。4.5 启动 MCP Server 并验证工具注册把上面代码整合到server.py后可以启动服务验证工具是否成功注册。python -m mcp_server.server如果你使用的是 MCP SDK 提供的调试工具可以用 MCP Inspector 或默认的 stdio 调试模式查看工具列表。没有 Inspector 时也可以在代码里打印注册结果或通过客户端调用测试。启动后MCP Server 会等待客户端连接。在实际业务系统中MCP Server 通常作为子进程被业务后端拉起而不是一直占用终端。这样可以减少人工干预也更适合部署到服务器。5. 付费内容与多商户系统数据模型设计5.1 核心表结构设计MCP 工具层负责提供能力真正存储业务数据还是要靠数据库。我们设计四张核心表商户表、内容表、订单表、结算表。商户表保存多商户的基本信息CREATE TABLE merchant ( id BIGINT PRIMARY KEY AUTO_INCREMENT, merchant_no VARCHAR(32) NOT NULL UNIQUE COMMENT 商户编号, name VARCHAR(128) NOT NULL COMMENT 商户名称, status TINYINT NOT NULL DEFAULT 1 COMMENT 1可用 0禁用, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) COMMENT 多商户表;内容表保存商户发布的内容并且用is_paid和price区分免费与付费内容。CREATE TABLE content ( id BIGINT PRIMARY KEY AUTO_INCREMENT, merchant_id BIGINT NOT NULL, title VARCHAR(255) NOT NULL, summary VARCHAR(512) DEFAULT , body LONGTEXT NOT NULL, cover_url VARCHAR(512) DEFAULT , is_paid TINYINT NOT NULL DEFAULT 0 COMMENT 0免费 1付费, price DECIMAL(10,2) NOT NULL DEFAULT 0.00 COMMENT 单价, status TINYINT NOT NULL DEFAULT 0 COMMENT 0草稿 1待审 2已发布 3下架, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_merchant_status (merchant_id, status) ) COMMENT 内容表;这里的status字段是内容状态机的核心。从草稿到待审到已发布再到下架每个状态变化都应该在业务层做校验。免费内容可以直接发布付费内容必须经过审核、绑定收款账户后才会允许上架。订单表保存用户购买付费内容的记录CREATE TABLE orders ( id BIGINT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(64) NOT NULL UNIQUE, merchant_id BIGINT NOT NULL, content_id BIGINT NOT NULL, buyer_id BIGINT NOT NULL, pay_amount DECIMAL(10,2) NOT NULL, pay_status TINYINT NOT NULL DEFAULT 0 COMMENT 0待支付 1已支付 2已退款, pay_channel VARCHAR(32) DEFAULT , paid_at DATETIME DEFAULT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_content_buyer (content_id, buyer_id) ) COMMENT 付费内容订单表;订单表最需要关注的是order_no的唯一性。每次下单必须使用全局唯一的订单号避免支付回调时无法定位订单。为了支持多商户分账还需要一张结算表按日和商户口径汇总CREATE TABLE merchant_settlement ( id BIGINT PRIMARY KEY AUTO_INCREMENT, merchant_no VARCHAR(32) NOT NULL, settle_date DATE NOT NULL, total_amount DECIMAL(12,2) NOT NULL DEFAULT 0, platform_income DECIMAL(12,2) NOT NULL DEFAULT 0, merchant_income DECIMAL(12,2) NOT NULL DEFAULT 0, status TINYINT NOT NULL DEFAULT 0 COMMENT 0未结算 1已结算, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_merchant_date (merchant_no, settle_date) ) COMMENT 商户分账表;结算表用merchant_no settle_date做唯一键防止同一天重复结算。5.2 多商户数据隔离原则数据表建好后最容易踩的坑就是多商户数据串号。比如查询内容时漏掉了merchant_id条件商户 A 的运营人员就可能看到并修改商户 B 的内容。这是多商户系统的红线问题。要避免这个问题可以遵守以下原则。第一所有内容查询必须强制携带merchant_id除非是对平台管理员开放的特殊查询接口。第二在 DAO 层或 ORM 层做统一拦截而不是靠每个开发人员自己记得加条件。第三前端菜单和按钮权限也要按商户角色隔离。第四在测试环节专门设计“商户 A 登录访问商户 B 数据”的用例确保隔离有效。在实际项目中我比较推荐在 ORM 查询基类中内置一个tenant_scope()方法所有查询都通过它来追加租户条件。这样即使某个查询漏写了merchant_id框架层也能兜底。5.3 付费解锁状态机设计付费内容的阅读权限和订单状态强相关。一个用户能否阅读某篇付费内容取决于是否有一笔状态为“已支付”的订单。这里有一个常用的状态机。待支付用户点击解锁后创建订单此时订单等待支付。已支付支付渠道回调成功用户可以阅读全量内容。已退款订单发生退款后阅读权限应立即收回或在该用户下一次访问时校验失效。设计时要注意两个关键点第一支付回调必须做签名验证第二用户阅读权限的校验不能只依赖前端按钮显示后端接口也要在返回正文内容前校验订单状态。否则用户绕过前端直接请求接口就能拿到未购买的付费内容。5.4 分账和结算设计分账是付费内容系统里比较敏感的部分。常见做法是“T1 结算”今天产生的订单明天生成结算单平台按抽成比例扣除佣金后剩余金额进入商户可提现余额。平台抽成比例建议做成可配置项按商户维度设置。不同商户的议价能力可能不同有的抽成 10%有的抽成 5%全平台写死一个比例并不合适。在本文示例中我们在配置里预留了PLATFORM_COMMISSION_RATE实际项目可以把它扩展成商户表字段commission_rate。对账也是一个重要环节。每天定时任务统计前一天的付费订单总额和支付渠道的结算账单进行比对金额不一致时触发告警。这里特别提醒不要在正式的支付环境里随便修改订单金额所有涉及金额变更的操作都应该有审计日志并且在测试环境充分验证后再上线。6. 后端业务接口与 MCP 联动6.1 后端调用 MCP 的统一封装业务后端通过 MCP 客户端调用 MCP Server。我们封装一个统一的调用函数避免每个接口都重复写连接逻辑。# backend/app/mcp_client.py from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def call_mcp_tool(tool_name: str, arguments: dict): server_params StdioServerParameters( commandpython, args[mcp_server/server.py] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(tool_name, arguments) return result这里需要注意每次调用都会启动一个子进程频繁调用时性能不太好。实际项目中推荐启动一个常驻的 MCP 客户端连接或者把 MCP Server 部署成远程服务通过 HTTP/SSE 方式调用。6.2 内容发布接口有了 MCP 客户端封装业务接口就可以直接调用内容创建工具。这里以 FastAPI 为例实现一个内容发布接口。# backend/app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .mcp_client import call_mcp_tool app FastAPI(titleMCP 内容系统 API) class ArticleCreate(BaseModel): merchant_id: str title: str content: str is_paid: bool False price: float 0 app.post(/api/articles) async def create_article(req: ArticleCreate): result await call_mcp_tool(create_article, req.model_dump()) return {code: 0, data: result}在上面的代码中ArticleCreate是请求体模型model_dump()是 Pydantic v2 的方法。如果你还在使用 Pydantic v1需要改成dict()。这里还需要补充分层逻辑接口层负责参数校验业务层负责调用 MCP 工具和落库不要把数据库操作全堆在路由函数里。6.3 付费解锁流程付费解锁是交易域的核心接口我建议把它写在业务服务中而不是做成 MCP 工具。整个流程可以拆成五步。# backend/app/order_service.py import hashlib import time from datetime import datetime from decimal import Decimal def generate_order_no(merchant_id: int, content_id: int) - str: ts int(time.time() * 1000) raw f{merchant_id}-{content_id}-{ts} return ORD hashlib.sha1(raw.encode()).hexdigest()[:20].upper() async def unlock_content(buyer_id: int, content_id: int): # 1. 查询内容确认是付费内容且状态为已发布 # 2. 判断用户是否已经购买存在已支付订单则直接返回阅读权限 # 3. 查询内容价格创建待支付订单 # 4. 调用支付渠道创建支付单返回支付参数 # 5. 异步等待支付回调回调成功后将订单状态改为已支付 pass在实现时要注意创建订单和调用支付渠道之间可能发生重复请求。建议在创建订单前先查一次 Redis用“内容ID 用户ID 商品ID”作为幂等键防止用户重复点击解锁时生成多笔订单。更关键的是支付回调。支付渠道回调通知到达后后端必须验证签名、校验金额、更新订单状态。更新订单时最好加上状态条件比如UPDATE orders SET pay_status 1 WHERE order_no ? AND pay_status 0避免回调重复通知导致状态覆盖。6.4 运行和验证完成以上代码后启动 MCP Server再启动后端接口。uvicorn backend.app.main:app --reload --port 8000然后通过 curl 测试内容创建接口。curl -X POST http://127.0.0.1:8000/api/articles \ -H Content-Type: application/json \ -d {merchant_id:M001,title:支付系统实战,content:正文内容,is_paid:true,price:9.9}如果一切正常接口会返回类似下面的 JSON{ code: 0, data: { status: success, article_id: ART_1700000000000, merchant_id: M001, title: 支付系统实战, content_length: 6, is_paid: true, price: 9.9, status: draft } }这说明 MCP 工具已经被成功调用内容创建流程跑通了。接下来你可以继续把内容写入 MySQL、把草稿状态切换为待审核然后接入审核工具。7. 常见问题与排查思路问题现象常见原因解决思路MCP Server 启动失败Python 版本或 SDK 版本不兼容检查 Python 版本升级或降级 MCP SDK工具调用超时stdio 模式每个请求都新建子进程改为常驻连接或部署远程 MCP Server工具注册不上装饰器导入路径不一致确认所有 MCP 工具统一注册到同一个 FastMCP 实例支付回调重复通知回调接口没有做幂等使用订单号做唯一约束加状态条件更新多商户数据串号查询条件漏传 merchant_idORM 层统一做租户隔离测试用例覆盖付费内容可以被未购买用户访问后端未校验订单状态返回正文前校验支付订单不能只靠前端隐藏AI 内容审核误判敏感词库过于简单接入专业内容安全服务增加人工复核下面展开几个高频问题。第一个是“MCP 工具注册不上”。这通常发生在多文件工程中比如server.py里创建了mcp实例但search_tools.py里又新建了一个FastMCP实例导致工具注册到了不同服务上。解决方法是把mcp实例单独拆到一个mcp_instance.py文件所有工具文件都从该文件导入同一个实例。第二个是“调用超时”。MCP 的 stdio 模式适合本地调试但如果是高并发业务建议部署远程 MCP Server通过 HTTP 或 SSE 方式连接客户端与服务端保持长连接避免频繁创建子进程。第三个是“支付回调重复通知”。绝大多数支付渠道为了保证通知可靠都会提供重试机制。处理办法是回调接口必须幂等先按订单号查库如果订单已经是已支付状态直接返回成功不再重复处理。第四个是“付费内容越权访问”。这个问题在多商户场景里尤其严重。用户 A 购买了商户 X 的内容如果接口返回正文时只校验用户是否登录不校验该用户是否购买了当前内容就会造成越权。正确做法是每次返回内容正文之前都通过订单表查询当前用户对当前内容是否拥有有效已支付订单。8. 最佳实践与工程建议8.1 MCP 工具设计的最佳实践工具命名建议采用“动词 名词”的方式比如create_article、audit_article、search_articles。这样模型在选择工具时更容易理解。工具描述要写清楚入参含义和返回结构因为描述本身会影响模型对工具的理解。工具内部要做好异常捕获。MCP 工具返回给大模型的信息会直接影响模型的下一步判断如果直接把数据库异常堆栈抛给模型既不安全也容易误导。建议统一捕获异常返回结构化的错误信息比如{status: error, message: article not found}。在业务落库时记住一条原则MCP 工具适合无状态的计算和查询不适合承载强事务。如果你需要在单个流程里同时更新内容表和订单表建议在业务后端使用本地事务而不是通过 MCP 工具分步调用。8.2 付费系统安全边界付费内容系统直接涉及资金安全需要守住几条安全底线。第一金额计算必须使用 Decimal不能使用浮点数相加否则会出现精度误差。第二用户下单时后端要以数据库价格为准不能信任前端传过来的金额防止篡改。第三支付回调验签是强制要求验签失败直接拒绝更新订单。第四生产环境数据库账号使用最小权限应用账号不应该有 DROP、TRUNCATE 权限。第五所有退款操作要经过人工审核流程并且保留完整操作日志。数据库变更也是一个不可忽视的风险点。涉及生产环境的内容表、订单表结构变更时必须先备份并在测试环境完整验证后再执行。删除数据时一定记得加 WHERE 条件并且先 SELECT 确认影响范围。8.3 多商户隔离与灰度发布多商户内容系统的开发难点往往不在功能实现而在数据边界控制。建议从以下角度设计隔离能力。数据隔离所有内容、订单、结算数据都带merchant_id查询强制过滤。配置隔离每个商户可以配置不同的抽成比例、审核策略、内容封面规范。发布隔离商户的内容默认先到草稿状态平台可配置是否需要人工审核。灰度隔离在平台升级或模板变更时可以先在部分商户灰度观察数据后再全量放开。内容发布还可以设计定时上架能力。商户提前编辑好内容定时到某个时间点自动发布这样可以减少人工操作。定时任务建议在业务后端实现不要在 MCP 工具里做复杂调度。8.4 从 MCP 到 Agent Skill 的演进MCP 解决的是“模型如何调用工具”的问题它属于通信协议层。而 Agent Skill 更多是智能体侧的高层技能封装可以把多个 MCP 工具调用、提示词模板和业务规则组合成一个完整的技能。在内容系统场景里MCP 负责提供能力通道比如create_article、search_articles、audit_articleAgent Skill 则负责编排逻辑比如“写一篇文章并自动审核”这个技能先调用内容生成能力再调用审核工具最后把结果返回给用户。两者完全可以共存建议团队在初期先专注把 MCP 工具设计好后续再考虑 Agent Skill 的编排。另外MCP 生态里已经有很多成熟的第三方服务比如设计稿场景的 Figma MCP、蓝湖 MCP它们把设计工具能力标准化后提供给 AI 模型调用。做内容系统时也可以参考这些服务的工具拆分方式先定义清晰的资源再定义可执行的动作
做内容平台的老开发应该都有同感内容管理、发布、审核、检索、推荐这些模块本身写起来并不难难的是它们会反复出现在不同业务线里而且一旦牵扯到付费解锁、多商户分账代码就开始变得特别容易失控。再加上现在越来越多的 AI 能力要接入内容系统比如自动生成摘要、智能打标、敏感词初筛、私域推荐如果每个能力都单独对接一次后期维护成本会指数级上升。我们在实际项目里换了一条思路用 MCP 协议把内容能力封装成标准化工具再在业务后端统一调用整体架构清晰了很多。这篇文章会把整个方案的架构设计、数据模型、核心代码、常见问题和工程建议完整写出来适合正在开发内容平台、付费阅读系统或多商户 CMS 的同学参考。1. 内容系统为什么需要 MCP背景与核心概念1.1 从内容平台开发痛点说起很多人第一次接触“MCP 内容系统开发”这个说法时会有点困惑MCP 不是用来做大模型工具调用的吗跟内容系统、付费系统、多商户系统有什么关系实际上MCP 在这里解决的是一个非常现实的问题也就是“内容能力复用”。先看一个典型业务场景。假设平台内有多个角色运营可以发布平台公告商户可以发布自己的付费专栏用户可以购买内容并阅读管理员需要对所有内容做合规审核。传统做法是把内容表设计成一张大表用content_type区分栏目用merchant_id区分商户再用一个权限系统控制谁能操作。这套方案初期没问题但到了中期就会面临几个痛点第一AI 辅助写稿、AI 摘要、AI 标签、内容审核这些功能都要写独立的调用逻辑第二商户扩展越来越多每个商户都可能需要自己的内容管理工具第三付费内容和免费内容混在同一套逻辑里订单状态、退款、分账全都耦合在一起。MCP 的切入点就在这里。MCP 全称是 Model Context Protocol也就是模型上下文协议它由 Anthropic 提出目标是让 AI 模型能够通过统一协议去连接外部数据源和工具。你可以把 MCP 理解成一套“AI 世界里的 USB 接口”只要外部系统按照这个协议暴露能力AI 模型就能直接使用不需要为每个系统单独写一套对接代码。放到内容系统里我们就能把“创建内容”“检索内容”“审核内容”“推荐内容”这类能力统一封装成 MCP 工具业务系统和大模型都可以通过同一种方式调用。1.2 什么是 MCP 协议MCP 协议的核心设计是围绕三种原语展开的Tool、Resource、Prompt。Tool一个可执行的动作比如“创建内容”“查询订单”“审核文章”。它需要明确的入参和出参类似函数调用。Resource一份可被读取的数据比如“商户资料”“价格配置”“用户阅读权限”。它对外提供的是数据上下文而不是执行动作。Prompt一段预定义提示词模板方便在 AI 场景里复用固定话术比如“内容摘要生成模板”“合规审核提示词”。在传输方式上MCP 支持本地 stdio标准输入输出和远程 HTTP/SSE 两种模式。本地模式适合把 MCP Server 和业务系统跑在同一个进程环境里调试方便远程模式适合在分布式场景下把 MCP Server 部署成独立服务。和市面上真正的网络代理类工具不同MCP 是一种应用层协议它本身不涉及网络出口代理也不涉及任何网络穿透操作。它解决的是“AI 模型如何调用工具”这个工程问题。所以做 MCP 内容系统开发时完全可以把注意力集中在业务工具设计上不要被相关术语混淆。1.3 付费内容系统与多商户系统的概念边界要设计好这个系统先要把三个概念分开理解。内容系统负责内容的创建、编辑、审核、发布、下架、检索、展示。核心是“内容状态机”和“内容版本管理”。付费内容系统在内容系统之上增加交易能力包括试读、解锁、订单、支付回调、退款、阅读权限校验。核心是“订单状态机”和“用户权益校验”。多商户系统在内容系统之上增加商户租户隔离能力每个商户可以独立管理自己的内容和价格平台方可以按比例抽成。核心是“数据隔离”和“分账结算”。这三个概念是层层叠加的关系。多商户系统不一定有付费功能付费系统不一定是多商户但在这篇文章的实战场景里我们会把三者合并一个多商户内容平台商户发布内容平台允许内容设置免费或付费用户在购买后获得阅读权限平台按比例与商户分账。这个场景对数据一致性要求很高也是最有代表性的综合实战项目。1.4 MCP 为什么适合内容系统开发用传统方式开发这样的多商户付费内容系统通常需要在业务代码里同时维护内容服务、订单服务、商户服务、结算服务并且还要额外处理 AI 功能的接入。而引入 MCP 之后我们可以把内容系统里相对独立、可复用、适合被 AI 或外部系统调用的能力统一封装成 MCP 工具层。这样做有几个明显的好处。第一调用入口统一。不管是运营后台发布内容还是商户批量导入内容或者 AI 自动生成内容并入库最终都是调用同一个“创建内容”工具参数一致、行为一致、校验一致。第二AI 接入成本降低。后续如果要接各种大模型能力只需要让模型具备调用 MCP 工具的能力即可不需要为每个需求写新的服务接口。第三工具与业务解耦。MCP Server 可以独立部署、独立升级业务后端通过标准客户端调用互不影响。当然MCP 也不是银弹。它适合封装“动作清晰、参数明确、返回稳定”的能力不适合承载复杂的事务流程。比如下单支付这种需要强事务和多步骤确认的流程更推荐放在业务服务里做而不是一股脑塞进 MCP 工具。后面在设计工具边界时我们会重点强调这一点。2. 系统总体架构与技术选型2.1 整体架构设计整个系统采用分层架构从上到下可以拆成四层。前端(WEB / 小程序) │ HTTPS ▼ 业务后端 (FastAPI) ├── 商户模块 ├── 内容模块 ├── 订单模块 └── 结算模块 │ MCP 客户端调用 ▼ MCP Server (内容工具 / 检索工具 / 审核工具) │ SQL ▼ MySQL / Redis / 对象存储前端只负责页面展示和交互所有业务规则都通过业务后端接口完成。业务后端内部维护商户、内容、订单、结算等核心模块同时在需要复用内容能力时通过 MCP 客户端调用 MCP Server。MCP Server 负责把内容相关的通用能力封装为工具底层连接 MySQL、Redis 和对象存储。这种架构的好处是如果未来希望大模型直接操作内容系统只需要把 MCP Server 暴露出来让 AI Agent 连接即可如果未来希望其他业务系统也复用内容能力同样可以接入同一套 MCP Server。业务后端不需要改代码。2.2 技术选型和版本说明为了避免版本冲突本文的技术栈尽量选择比较稳定的组合但具体版本仍需要根据你的项目实际情况调整。示例环境以 Python 生态为主。后端框架FastAPI用于实现业务接口。MCP 客户端与服务端MCP Python SDK用于封装和调用内容工具。数据库MySQL 8用于存储商户、内容、订单、结算数据。缓存Redis用于内容热数据和订单幂等键缓存。ORMSQLAlchemy 2.x用于操作 MySQL。对象存储OSS/S3 兼容服务用于存储封面图、正文图片等附件。这里重点说明一下 MCP 的接入方式MCP 协议本身和语言无关官方提供 Python、TypeScript、Java 等多种 SDK。本文以 Python SDK 为例因为 FastAPI 生态写起内容系统来效率很高而且 MCP Python SDK 的 FastMCP 封装非常适合快速暴露工具。2.3 系统模块拆分从工程结构上看系统可以拆成五个核心模块。商户域负责商户注册、商户资料、商户状态、商户密钥管理。内容域负责内容草稿、内容审核、内容发布、内容下架、内容检索。交易域负责付费内容下单、支付回调、订单查询、阅读权限校验。结算域负责平台与商户的分账、提现、对账。MCP 工具域负责把内容域和审核域的能力包装成标准 MCP 工具。其中交易域不建议做成 MCP 工具。原因是支付流程涉及强交互、回调验签、金额计算属于高一致性业务放在业务服务里能更好地控制事务边界。MCP 工具域更适合做内容创建、内容检索、内容推荐、内容审核这类边界清晰的动作。3. 环境准备与工程初始化3.1 环境要求在开始写代码之前建议先准备好以下运行环境。操作系统Windows 10/11、macOS 或 Linux 均可。Python 版本建议 3.10 及以上本文示例使用 Python 3.10 语法。MySQL建议 8.0 及以上需要本地可连接。Redis建议 6.0 及以上用于缓存和幂等控制。Node.js如果后续要跑前端项目建议 18 及以上不是必须项。包管理工具建议使用 pip 和 venv。需要注意MCP 的版本迭代比较快不同版本下 SDK 的 API 会有差异。本文的代码示例以当前主流用法为准如果你的环境中 API 有变化优先查阅你实际安装版本的官方文档。3.2 创建工程目录和虚拟环境我们先创建一个标准工程目录。mkdir -p mcp-content-system/{mcp_server,backend,frontend,docs} cd mcp-content-system python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate激活虚拟环境后创建依赖文件。# requirements.txt fastapi uvicorn[standard] sqlalchemy pymysql redis mcp pydantic然后安装依赖。pip install -r requirements.txt安装完成后我们可以先检查一下关键包是否能正常导入。python -c from mcp.server.fastmcp import FastMCP; print(MCP OK)如果输出MCP OK说明 MCP SDK 已经可用。如果导入失败可能是版本差异问题可以试着调整 MCP SDK 的版本。3.3 数据库和基础配置在 MySQL 中创建系统需要的数据库。CREATE DATABASE IF NOT EXISTS mcp_content_system DEFAULT CHARACTER SET utf8mb4 DEFAULT COLLATE utf8mb4_unicode_ci;在项目根目录下创建配置文件config.py用于保存数据库、Redis、对象存储等连接信息。注意不要把生产环境的密钥直接提交到代码仓库建议通过环境变量注入。# config.py import os class Settings: MYSQL_HOST os.getenv(MYSQL_HOST, 127.0.0.1) MYSQL_PORT int(os.getenv(MYSQL_PORT, 3306)) MYSQL_USER os.getenv(MYSQL_USER, root) MYSQL_PASSWORD os.getenv(MYSQL_PASSWORD, your_password) MYSQL_DB os.getenv(MYSQL_DB, mcp_content_system) REDIS_URL os.getenv(REDIS_URL, redis://127.0.0.1:6379/0) PLATFORM_COMMISSION_RATE float(os.getenv(PLATFORM_COMMISSION_RATE, 0.1)) settings Settings()这里的PLATFORM_COMMISSION_RATE是平台抽成比例示例默认 10%实际项目需要根据自己的商业模式调整。4. MCP Server 核心实现4.1 MCP 工具设计方法论在写 MCP Server 之前先想清楚设计原则。一个合格的 MCP 工具应该满足以下条件。第一一个工具只做一件事。比如“创建内容”和“审核内容”最好拆成两个工具不要让一个工具既落库又审核否则后续很难复用。第二参数必须结构化。所有入参尽可能是明确的字符串、数字、布尔值或 JSON 对象不要用一段自然语言让 AI 自由发挥否则工具调用的稳定性会很差。第三返回结果要有明确结构。建议统一返回 JSON 格式包含status、biz_code、data、message等字段方便上层判断。第四工具描述要写清楚。MCP 是给大模型调用的协议工具描述会直接影响模型是否能正确选择工具所以描述要具体比如“创建一篇内容并返回 articleId”。4.2 实现内容创建工具我们使用 MCP Python SDK 的 FastMCP 封装来编写服务端。这里先创建一个轻量级的内容创建工具它负责校验参数并返回新内容的 ID 和状态。# mcp_server/server.py import time from mcp.server.fastmcp import FastMCP mcp FastMCP( mcp-content-system, instructions面向多商户内容平台提供内容创建、检索、审核等工具能力 ) mcp.tool() def create_article( merchant_id: str, title: str, content: str, is_paid: bool False, price: float 0.0 ) - dict: 创建一篇内容商户ID必传付费内容需要设置价格。 # 实际项目中应写入数据库这里先返回构建结果 article_id ART_ str(int(time.time() * 1000)) return { status: success, article_id: article_id, merchant_id: merchant_id, title: title, content_length: len(content), is_paid: is_paid, price: price, status: draft } if __name__ __main__: mcp.run()在这个示例里create_article接收商户 ID、标题、正文以及是否付费、价格两个可选参数。返回结果里包含内容长度、状态等信息这样调用方可以确认内容是否创建成功。从实际落地的角度create_article内部还应该做好几件事校验商户是否存在并处于可用状态校验正文内容是否为空校验付费内容的价格是否大于 0对内容做敏感词初筛将内容以草稿状态写入数据库。这里为了保持示例简洁用返回结构代替了持久化逻辑读者可以按自己的项目情况补充。4.3 实现内容检索与推荐工具内容平台最常用的能力之一就是检索。我们希望支持按关键词检索、按商户过滤、按付费状态过滤。这个工具适合封装成 MCP 工具因为后续无论是用户端的搜索接口、管理后台的内容列表还是 AI 推荐系统都需要类似的检索能力。# mcp_server/tools/search_tools.py from mcp.server.fastmcp import FastMCP import mcp_server.server as server server.mcp.tool() def search_articles( keyword: str , merchant_id: str , is_paid: bool | None None, limit: int 10 ) - dict: 按关键词检索已发布内容支持按商户和付费状态过滤。 # 实际项目中使用 SQL 查询示例直接返回空结果 return { status: success, total: 0, items: [] }注意这个示例为了调试方便直接返回了空列表。真正落地时这里应该执行 SQL 查询并保证在WHERE条件中强制携带status 2已发布避免把草稿内容检索出来。同时如果当前搜索场景要求商户隔离必须把merchant_id作为必填条件防止串号。检索工具适合做基础实现推荐工具则更复杂。推荐工具通常需要读取用户行为数据、内容标签、热度数据然后组合排序。我们可以把推荐工具单独封装方便后续接入大模型或算法服务。server.mcp.tool() def recommend_articles( user_id: str , limit: int 10 ) - dict: 根据用户行为推荐内容供内容流场景使用。 # 实际项目中可以结合 Redis 热度和用户标签计算 return { status: success, user_id: user_id, items: [] }这个工具的边界很清晰它只负责返回推荐结果不负责用户行为采集。行为采集应该由业务后端在阅读、点赞等事件中完成。4.4 实现内容审核工具内容审核是多商户内容系统里最容易被忽视、但最容易出问题的环节。审核工具可以做两层第一层是本地敏感词过滤第二层是调用专业内容安全服务。专业内容安全服务通常需要申请接口权限本文不展开具体品牌只做一个可以替换实现的占位。# mcp_server/tools/audit_tools.py import mcp_server.server as server server.mcp.tool() def audit_article( article_id: str, title: str, content: str ) - dict: 对标题和正文进行合规初筛返回是否通过。 # 第一层本地敏感词过滤 risk_words [示例违规词A, 示例违规词B] title_hit [w for w in risk_words if w in title] content_hit [w for w in risk_words if w in content] # 第二层专业内容安全服务调用可以写在这里 # external_result call_security_api(title, content) passed len(title_hit) 0 and len(content_hit) 0 return { status: success, article_id: article_id, passed: passed, title_hit: title_hit, content_hit: content_hit }审核工具的返回值非常关键。如果passed为 false业务系统需要决定是把内容打回草稿还是进入人工复审队列。这里的一个最佳实践是机器审核通过不代表完全没问题建议设置“机审通过 人工抽查”的双重机制尤其是涉及付费内容时更要谨慎。4.5 启动 MCP Server 并验证工具注册把上面代码整合到server.py后可以启动服务验证工具是否成功注册。python -m mcp_server.server如果你使用的是 MCP SDK 提供的调试工具可以用 MCP Inspector 或默认的 stdio 调试模式查看工具列表。没有 Inspector 时也可以在代码里打印注册结果或通过客户端调用测试。启动后MCP Server 会等待客户端连接。在实际业务系统中MCP Server 通常作为子进程被业务后端拉起而不是一直占用终端。这样可以减少人工干预也更适合部署到服务器。5. 付费内容与多商户系统数据模型设计5.1 核心表结构设计MCP 工具层负责提供能力真正存储业务数据还是要靠数据库。我们设计四张核心表商户表、内容表、订单表、结算表。商户表保存多商户的基本信息CREATE TABLE merchant ( id BIGINT PRIMARY KEY AUTO_INCREMENT, merchant_no VARCHAR(32) NOT NULL UNIQUE COMMENT 商户编号, name VARCHAR(128) NOT NULL COMMENT 商户名称, status TINYINT NOT NULL DEFAULT 1 COMMENT 1可用 0禁用, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) COMMENT 多商户表;内容表保存商户发布的内容并且用is_paid和price区分免费与付费内容。CREATE TABLE content ( id BIGINT PRIMARY KEY AUTO_INCREMENT, merchant_id BIGINT NOT NULL, title VARCHAR(255) NOT NULL, summary VARCHAR(512) DEFAULT , body LONGTEXT NOT NULL, cover_url VARCHAR(512) DEFAULT , is_paid TINYINT NOT NULL DEFAULT 0 COMMENT 0免费 1付费, price DECIMAL(10,2) NOT NULL DEFAULT 0.00 COMMENT 单价, status TINYINT NOT NULL DEFAULT 0 COMMENT 0草稿 1待审 2已发布 3下架, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_merchant_status (merchant_id, status) ) COMMENT 内容表;这里的status字段是内容状态机的核心。从草稿到待审到已发布再到下架每个状态变化都应该在业务层做校验。免费内容可以直接发布付费内容必须经过审核、绑定收款账户后才会允许上架。订单表保存用户购买付费内容的记录CREATE TABLE orders ( id BIGINT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(64) NOT NULL UNIQUE, merchant_id BIGINT NOT NULL, content_id BIGINT NOT NULL, buyer_id BIGINT NOT NULL, pay_amount DECIMAL(10,2) NOT NULL, pay_status TINYINT NOT NULL DEFAULT 0 COMMENT 0待支付 1已支付 2已退款, pay_channel VARCHAR(32) DEFAULT , paid_at DATETIME DEFAULT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_content_buyer (content_id, buyer_id) ) COMMENT 付费内容订单表;订单表最需要关注的是order_no的唯一性。每次下单必须使用全局唯一的订单号避免支付回调时无法定位订单。为了支持多商户分账还需要一张结算表按日和商户口径汇总CREATE TABLE merchant_settlement ( id BIGINT PRIMARY KEY AUTO_INCREMENT, merchant_no VARCHAR(32) NOT NULL, settle_date DATE NOT NULL, total_amount DECIMAL(12,2) NOT NULL DEFAULT 0, platform_income DECIMAL(12,2) NOT NULL DEFAULT 0, merchant_income DECIMAL(12,2) NOT NULL DEFAULT 0, status TINYINT NOT NULL DEFAULT 0 COMMENT 0未结算 1已结算, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_merchant_date (merchant_no, settle_date) ) COMMENT 商户分账表;结算表用merchant_no settle_date做唯一键防止同一天重复结算。5.2 多商户数据隔离原则数据表建好后最容易踩的坑就是多商户数据串号。比如查询内容时漏掉了merchant_id条件商户 A 的运营人员就可能看到并修改商户 B 的内容。这是多商户系统的红线问题。要避免这个问题可以遵守以下原则。第一所有内容查询必须强制携带merchant_id除非是对平台管理员开放的特殊查询接口。第二在 DAO 层或 ORM 层做统一拦截而不是靠每个开发人员自己记得加条件。第三前端菜单和按钮权限也要按商户角色隔离。第四在测试环节专门设计“商户 A 登录访问商户 B 数据”的用例确保隔离有效。在实际项目中我比较推荐在 ORM 查询基类中内置一个tenant_scope()方法所有查询都通过它来追加租户条件。这样即使某个查询漏写了merchant_id框架层也能兜底。5.3 付费解锁状态机设计付费内容的阅读权限和订单状态强相关。一个用户能否阅读某篇付费内容取决于是否有一笔状态为“已支付”的订单。这里有一个常用的状态机。待支付用户点击解锁后创建订单此时订单等待支付。已支付支付渠道回调成功用户可以阅读全量内容。已退款订单发生退款后阅读权限应立即收回或在该用户下一次访问时校验失效。设计时要注意两个关键点第一支付回调必须做签名验证第二用户阅读权限的校验不能只依赖前端按钮显示后端接口也要在返回正文内容前校验订单状态。否则用户绕过前端直接请求接口就能拿到未购买的付费内容。5.4 分账和结算设计分账是付费内容系统里比较敏感的部分。常见做法是“T1 结算”今天产生的订单明天生成结算单平台按抽成比例扣除佣金后剩余金额进入商户可提现余额。平台抽成比例建议做成可配置项按商户维度设置。不同商户的议价能力可能不同有的抽成 10%有的抽成 5%全平台写死一个比例并不合适。在本文示例中我们在配置里预留了PLATFORM_COMMISSION_RATE实际项目可以把它扩展成商户表字段commission_rate。对账也是一个重要环节。每天定时任务统计前一天的付费订单总额和支付渠道的结算账单进行比对金额不一致时触发告警。这里特别提醒不要在正式的支付环境里随便修改订单金额所有涉及金额变更的操作都应该有审计日志并且在测试环境充分验证后再上线。6. 后端业务接口与 MCP 联动6.1 后端调用 MCP 的统一封装业务后端通过 MCP 客户端调用 MCP Server。我们封装一个统一的调用函数避免每个接口都重复写连接逻辑。# backend/app/mcp_client.py from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def call_mcp_tool(tool_name: str, arguments: dict): server_params StdioServerParameters( commandpython, args[mcp_server/server.py] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(tool_name, arguments) return result这里需要注意每次调用都会启动一个子进程频繁调用时性能不太好。实际项目中推荐启动一个常驻的 MCP 客户端连接或者把 MCP Server 部署成远程服务通过 HTTP/SSE 方式调用。6.2 内容发布接口有了 MCP 客户端封装业务接口就可以直接调用内容创建工具。这里以 FastAPI 为例实现一个内容发布接口。# backend/app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .mcp_client import call_mcp_tool app FastAPI(titleMCP 内容系统 API) class ArticleCreate(BaseModel): merchant_id: str title: str content: str is_paid: bool False price: float 0 app.post(/api/articles) async def create_article(req: ArticleCreate): result await call_mcp_tool(create_article, req.model_dump()) return {code: 0, data: result}在上面的代码中ArticleCreate是请求体模型model_dump()是 Pydantic v2 的方法。如果你还在使用 Pydantic v1需要改成dict()。这里还需要补充分层逻辑接口层负责参数校验业务层负责调用 MCP 工具和落库不要把数据库操作全堆在路由函数里。6.3 付费解锁流程付费解锁是交易域的核心接口我建议把它写在业务服务中而不是做成 MCP 工具。整个流程可以拆成五步。# backend/app/order_service.py import hashlib import time from datetime import datetime from decimal import Decimal def generate_order_no(merchant_id: int, content_id: int) - str: ts int(time.time() * 1000) raw f{merchant_id}-{content_id}-{ts} return ORD hashlib.sha1(raw.encode()).hexdigest()[:20].upper() async def unlock_content(buyer_id: int, content_id: int): # 1. 查询内容确认是付费内容且状态为已发布 # 2. 判断用户是否已经购买存在已支付订单则直接返回阅读权限 # 3. 查询内容价格创建待支付订单 # 4. 调用支付渠道创建支付单返回支付参数 # 5. 异步等待支付回调回调成功后将订单状态改为已支付 pass在实现时要注意创建订单和调用支付渠道之间可能发生重复请求。建议在创建订单前先查一次 Redis用“内容ID 用户ID 商品ID”作为幂等键防止用户重复点击解锁时生成多笔订单。更关键的是支付回调。支付渠道回调通知到达后后端必须验证签名、校验金额、更新订单状态。更新订单时最好加上状态条件比如UPDATE orders SET pay_status 1 WHERE order_no ? AND pay_status 0避免回调重复通知导致状态覆盖。6.4 运行和验证完成以上代码后启动 MCP Server再启动后端接口。uvicorn backend.app.main:app --reload --port 8000然后通过 curl 测试内容创建接口。curl -X POST http://127.0.0.1:8000/api/articles \ -H Content-Type: application/json \ -d {merchant_id:M001,title:支付系统实战,content:正文内容,is_paid:true,price:9.9}如果一切正常接口会返回类似下面的 JSON{ code: 0, data: { status: success, article_id: ART_1700000000000, merchant_id: M001, title: 支付系统实战, content_length: 6, is_paid: true, price: 9.9, status: draft } }这说明 MCP 工具已经被成功调用内容创建流程跑通了。接下来你可以继续把内容写入 MySQL、把草稿状态切换为待审核然后接入审核工具。7. 常见问题与排查思路问题现象常见原因解决思路MCP Server 启动失败Python 版本或 SDK 版本不兼容检查 Python 版本升级或降级 MCP SDK工具调用超时stdio 模式每个请求都新建子进程改为常驻连接或部署远程 MCP Server工具注册不上装饰器导入路径不一致确认所有 MCP 工具统一注册到同一个 FastMCP 实例支付回调重复通知回调接口没有做幂等使用订单号做唯一约束加状态条件更新多商户数据串号查询条件漏传 merchant_idORM 层统一做租户隔离测试用例覆盖付费内容可以被未购买用户访问后端未校验订单状态返回正文前校验支付订单不能只靠前端隐藏AI 内容审核误判敏感词库过于简单接入专业内容安全服务增加人工复核下面展开几个高频问题。第一个是“MCP 工具注册不上”。这通常发生在多文件工程中比如server.py里创建了mcp实例但search_tools.py里又新建了一个FastMCP实例导致工具注册到了不同服务上。解决方法是把mcp实例单独拆到一个mcp_instance.py文件所有工具文件都从该文件导入同一个实例。第二个是“调用超时”。MCP 的 stdio 模式适合本地调试但如果是高并发业务建议部署远程 MCP Server通过 HTTP 或 SSE 方式连接客户端与服务端保持长连接避免频繁创建子进程。第三个是“支付回调重复通知”。绝大多数支付渠道为了保证通知可靠都会提供重试机制。处理办法是回调接口必须幂等先按订单号查库如果订单已经是已支付状态直接返回成功不再重复处理。第四个是“付费内容越权访问”。这个问题在多商户场景里尤其严重。用户 A 购买了商户 X 的内容如果接口返回正文时只校验用户是否登录不校验该用户是否购买了当前内容就会造成越权。正确做法是每次返回内容正文之前都通过订单表查询当前用户对当前内容是否拥有有效已支付订单。8. 最佳实践与工程建议8.1 MCP 工具设计的最佳实践工具命名建议采用“动词 名词”的方式比如create_article、audit_article、search_articles。这样模型在选择工具时更容易理解。工具描述要写清楚入参含义和返回结构因为描述本身会影响模型对工具的理解。工具内部要做好异常捕获。MCP 工具返回给大模型的信息会直接影响模型的下一步判断如果直接把数据库异常堆栈抛给模型既不安全也容易误导。建议统一捕获异常返回结构化的错误信息比如{status: error, message: article not found}。在业务落库时记住一条原则MCP 工具适合无状态的计算和查询不适合承载强事务。如果你需要在单个流程里同时更新内容表和订单表建议在业务后端使用本地事务而不是通过 MCP 工具分步调用。8.2 付费系统安全边界付费内容系统直接涉及资金安全需要守住几条安全底线。第一金额计算必须使用 Decimal不能使用浮点数相加否则会出现精度误差。第二用户下单时后端要以数据库价格为准不能信任前端传过来的金额防止篡改。第三支付回调验签是强制要求验签失败直接拒绝更新订单。第四生产环境数据库账号使用最小权限应用账号不应该有 DROP、TRUNCATE 权限。第五所有退款操作要经过人工审核流程并且保留完整操作日志。数据库变更也是一个不可忽视的风险点。涉及生产环境的内容表、订单表结构变更时必须先备份并在测试环境完整验证后再执行。删除数据时一定记得加 WHERE 条件并且先 SELECT 确认影响范围。8.3 多商户隔离与灰度发布多商户内容系统的开发难点往往不在功能实现而在数据边界控制。建议从以下角度设计隔离能力。数据隔离所有内容、订单、结算数据都带merchant_id查询强制过滤。配置隔离每个商户可以配置不同的抽成比例、审核策略、内容封面规范。发布隔离商户的内容默认先到草稿状态平台可配置是否需要人工审核。灰度隔离在平台升级或模板变更时可以先在部分商户灰度观察数据后再全量放开。内容发布还可以设计定时上架能力。商户提前编辑好内容定时到某个时间点自动发布这样可以减少人工操作。定时任务建议在业务后端实现不要在 MCP 工具里做复杂调度。8.4 从 MCP 到 Agent Skill 的演进MCP 解决的是“模型如何调用工具”的问题它属于通信协议层。而 Agent Skill 更多是智能体侧的高层技能封装可以把多个 MCP 工具调用、提示词模板和业务规则组合成一个完整的技能。在内容系统场景里MCP 负责提供能力通道比如create_article、search_articles、audit_articleAgent Skill 则负责编排逻辑比如“写一篇文章并自动审核”这个技能先调用内容生成能力再调用审核工具最后把结果返回给用户。两者完全可以共存建议团队在初期先专注把 MCP 工具设计好后续再考虑 Agent Skill 的编排。另外MCP 生态里已经有很多成熟的第三方服务比如设计稿场景的 Figma MCP、蓝湖 MCP它们把设计工具能力标准化后提供给 AI 模型调用。做内容系统时也可以参考这些服务的工具拆分方式先定义清晰的资源再定义可执行的动作
