AI Agent 技能分享｜从“只读”到“可写”：让 MCP Server 安全修改数据库
AI Agent 技能分享｜从“只读”到“可写”：让 MCP Server 安全修改数据库
之前做了一个只读 MCP Server，Agent 可以查订单，但碰不到任意 SQL。只读服务跑通以后，很快就会有人问：既然订单能查，能不能顺便帮我取消订单、加备注、改状态？
技术上当然可以。给数据库账号补上 INSERT、UPDATE 权限，再写一个 Tool 就能跑起来。真正麻烦的是另外几个问题：
- Agent 因为超时重试，会不会连续提交两次？
- 用户只是询问取消规则，模型会不会直接执行取消？
- A 租户传入 B 租户的订单号怎么办？
- 用户确认的是订单 A，执行时参数被换成订单 B 怎么办？
- 数据库写到一半报错，业务表和审计表会不会只成功一张？
- 线上出问题以后，能不能还原是谁让 Agent 做了什么？
所以这次不做 execute_sql，也不允许模型直接传一段 UPDATE。我们做一个两阶段的订单取消流程：
用户提出取消订单
↓
Agent 调用 prepare_order_cancel
↓
服务端检查订单、权限和业务状态
↓
返回操作预览和 operation_id
↓
客户端向用户展示确认框
↓
用户确认后调用 confirm_order_cancel
↓
事务内创建取消申请、更新操作状态、写审计日志
注意，示例提交的是“取消申请”，不是直接删除订单。越靠近核心业务，MCP Tool 越应该复用原系统的业务规则，而不是绕过原系统修改几张表。
最危险的不是 SQL 注入，而是合法地做错事
下面这种代码显然不能上线：
@mcp.tool()
def execute_sql(sql: str) -> dict:
with engine.begin() as connection:
result = connection.execute(text(sql))
return {"rowcount": result.rowcount}
它有 SQL 注入、越权和误删数据等问题，这些都比较容易想到。
还有一种写法看上去安全一些：
@mcp.tool()
def update_order_status(order_no: str, status: str) -> dict:
...
SQL 可以做参数化，状态也可以限制成枚举，但业务风险仍然很大。模型只要选错状态，或者在用户没有确认时调用这个 Tool，数据库就会执行一条语法正确、权限正常、业务结果错误的更新。
写操作和查询不一样。查询错了通常是答案不准确，写错了会形成真实副作用。数据库安全只能挡住“不允许访问什么”，它无法判断“这次操作是不是用户真正想做的”。
MCP 官方 Tools 安全建议也明确要求服务端校验输入、实施访问控制、限制调用频率并清理输出；对敏感操作，客户端应向用户展示调用参数并请求确认，同时保留工具调用日志。MCP Tools 安全说明
不开放万能写工具
这次只提供两个窄工具：
prepare_order_cancel
检查订单是否允许取消，创建一条短期操作意图，返回确认预览。
confirm_order_cancel
根据 operation_id 提交取消申请，只接受已经准备并经过确认的操作。
它们不能修改价格、收货地址和支付状态，也不能执行任意 SQL。Tool 的能力边界就是业务边界。
整个实现分成三层：
MCP Tool
├─ 参数格式校验
├─ 从可信上下文取得用户身份
└─ 返回适合模型理解的结果
↓
数据库存储过程
├─ 再次校验租户和订单状态
├─ 锁定操作记录
├─ 保证幂等
├─ 事务提交
└─ 写审计日志
↓
数据库权限
├─ 禁止直接修改 Orders
└─ 只允许执行指定存储过程
即使 MCP Server 以后被误改，数据库账号也没有直接 UPDATE Orders 的权限。
项目准备
目录继续保持简单：
mcp-order-writer/
├─ server.py
├─ requirements.txt
└─ .env.example
requirements.txt：
mcp[cli]>=2,<3
SQLAlchemy>=2,<3
pyodbc>=5,<6
pydantic>=2,<3
创建环境并安装依赖：
python -m venv .venv
# Windows
.venv\Scripts\activate
pip install -r requirements.txt
.env.example：
SQLSERVER_URL=mssql+pyodbc://mcp_writer:请替换密码@127.0.0.1:1433/OrderDb?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes
MCP_TENANT_ID=tenant_demo
MCP_USER_ID=user_demo
MCP_SCOPES=order.read,order.cancel.request
这里的环境变量身份只方便本机演示。远程服务必须从经过验证的访问令牌中取得 tenant_id、user_id 和 Scope，不能让模型传入，也不能在多用户服务里使用全局环境变量冒充当前用户。
先建操作意图，而不是直接改订单
准备阶段需要一张操作意图表。它记录用户准备执行什么、操作何时过期，以及最后是否已经提交。
CREATE TABLE dbo.AgentOperationIntent
(
OperationId UNIQUEIDENTIFIER NOT NULL PRIMARY KEY,
TenantId NVARCHAR(64) NOT NULL,
UserId NVARCHAR(64) NOT NULL,
OperationType VARCHAR(64) NOT NULL,
TargetType VARCHAR(32) NOT NULL,
TargetId BIGINT NOT NULL,
PayloadJson NVARCHAR(MAX) NOT NULL,
PayloadHash CHAR(64) NOT NULL,
OperationStatus VARCHAR(20) NOT NULL,
ExpiresAt DATETIME2(0) NOT NULL,
ResultId BIGINT NULL,
CreatedAt DATETIME2(0) NOT NULL
CONSTRAINT DF_AgentOperationIntent_CreatedAt DEFAULT SYSUTCDATETIME(),
ConfirmedAt DATETIME2(0) NULL,
CONSTRAINT CK_AgentOperationIntent_Status
CHECK (OperationStatus IN ('PREPARED', 'SUBMITTED', 'EXPIRED', 'REJECTED')),
CONSTRAINT CK_AgentOperationIntent_PayloadJson
CHECK (ISJSON(PayloadJson) = 1)
);
GO
CREATE INDEX IX_AgentOperationIntent_Scope
ON dbo.AgentOperationIntent(TenantId, UserId, OperationStatus, ExpiresAt);
GO
取消申请单单独保存：
CREATE TABLE dbo.OrderCancelRequest
(
RequestId BIGINT IDENTITY(1, 1) NOT NULL PRIMARY KEY,
TenantId NVARCHAR(64) NOT NULL,
OrderId BIGINT NOT NULL,
RequestedBy NVARCHAR(64) NOT NULL,
Reason NVARCHAR(200) NOT NULL,
RequestStatus VARCHAR(20) NOT NULL,
SourceOperationId UNIQUEIDENTIFIER NOT NULL,
CreatedAt DATETIME2(0) NOT NULL
CONSTRAINT DF_OrderCancelRequest_CreatedAt DEFAULT SYSUTCDATETIME(),
CONSTRAINT UQ_OrderCancelRequest_SourceOperation
UNIQUE(SourceOperationId)
);
GO
SourceOperationId 上的唯一约束是最后一道幂等防线。即使应用层判断失效，同一个操作意图也不能生成两张取消申请单。
最后是审计表：
CREATE TABLE dbo.AgentToolAudit
(
AuditId BIGINT IDENTITY(1, 1) NOT NULL PRIMARY KEY,
TraceId UNIQUEIDENTIFIER NOT NULL,
TenantId NVARCHAR(64) NOT NULL,
UserId NVARCHAR(64) NOT NULL,
ToolName VARCHAR(100) NOT NULL,
OperationId UNIQUEIDENTIFIER NULL,
TargetType VARCHAR(32) NULL,
TargetId BIGINT NULL,
ParameterSummary NVARCHAR(1000) NULL,
ResultCode VARCHAR(50) NOT NULL,
CreatedAt DATETIME2(0) NOT NULL
CONSTRAINT DF_AgentToolAudit_CreatedAt DEFAULT SYSUTCDATETIME()
);
GO
CREATE INDEX IX_AgentToolAudit_TraceId
ON dbo.AgentToolAudit(TraceId);
GO
审计表不保存访问令牌，也不保存完整请求头。取消原因可以保留，但如果业务中可能包含手机号、证件号等敏感信息，还需要在进入日志前脱敏。
准备阶段的存储过程
准备操作时，数据库要重新查订单，不能相信 Agent 在上一轮返回的订单状态。
CREATE OR ALTER PROCEDURE dbo.usp_AgentPrepareOrderCancel
@OperationId UNIQUEIDENTIFIER,
@TraceId UNIQUEIDENTIFIER,
@TenantId NVARCHAR(64),
@UserId NVARCHAR(64),
@OrderNo NVARCHAR(32),
@Reason NVARCHAR(200),
@PayloadJson NVARCHAR(MAX),
@PayloadHash CHAR(64),
@ExpiresAt DATETIME2(0)
AS
BEGIN
SET NOCOUNT ON;
SET XACT_ABORT ON;
DECLARE @OrderId BIGINT;
DECLARE @OrderStatus VARCHAR(20);
DECLARE @TotalAmount DECIMAL(18, 2);
SELECT TOP (1)
@OrderId = o.OrderId,
@OrderStatus = o.OrderStatus,
@TotalAmount = o.TotalAmount
FROM dbo.Orders AS o
WHERE o.TenantId = @TenantId
AND o.OrderNo = @OrderNo
AND o.IsDeleted = 0;
IF @OrderId IS NULL
THROW 51001, N'订单不存在或无权访问', 1;
IF @OrderStatus NOT IN ('PENDING', 'PAID')
THROW 51002, N'当前订单状态不允许申请取消', 1;
BEGIN TRANSACTION;
INSERT INTO dbo.AgentOperationIntent
(
OperationId,
TenantId,
UserId,
OperationType,
TargetType,
TargetId,
PayloadJson,
PayloadHash,
OperationStatus,
ExpiresAt
)
VALUES
(
@OperationId,
@TenantId,
@UserId,
'ORDER_CANCEL',
'ORDER',
@OrderId,
@PayloadJson,
@PayloadHash,
'PREPARED',
@ExpiresAt
);
INSERT INTO dbo.AgentToolAudit
(
TraceId,
TenantId,
UserId,
ToolName,
OperationId,
TargetType,
TargetId,
ParameterSummary,
ResultCode
)
VALUES
(
@TraceId,
@TenantId,
@UserId,
'prepare_order_cancel',
@OperationId,
'ORDER',
@OrderId,
CONCAT(N'order_no=', @OrderNo),
'PREPARED'
);
COMMIT TRANSACTION;
SELECT
@OperationId AS OperationId,
@OrderId AS OrderId,
@OrderNo AS OrderNo,
@OrderStatus AS OrderStatus,
@TotalAmount AS TotalAmount,
@Reason AS Reason,
@ExpiresAt AS ExpiresAt;
END;
GO
准备阶段只写操作意图，不修改订单，也不创建取消申请。它的作用是把“用户准备取消哪张订单、使用什么理由”冻结下来，供客户端展示确认。
确认阶段要锁住操作记录
确认请求可能因为网络超时被调用多次，也可能有两个并发请求同时进入。这里只在 Python 里先查一次状态不够，状态检查和写入必须放进同一个数据库事务。
CREATE OR ALTER PROCEDURE dbo.usp_AgentConfirmOrderCancel
@OperationId UNIQUEIDENTIFIER,
@TraceId UNIQUEIDENTIFIER,
@TenantId NVARCHAR(64),
@UserId NVARCHAR(64)
AS
BEGIN
SET NOCOUNT ON;
SET XACT_ABORT ON;
DECLARE @Status VARCHAR(20);
DECLARE @ExpiresAt DATETIME2(0);
DECLARE @TargetId BIGINT;
DECLARE @PayloadJson NVARCHAR(MAX);
DECLARE @RequestId BIGINT;
DECLARE @Reason NVARCHAR(200);
BEGIN TRANSACTION;
SELECT
@Status = OperationStatus,
@ExpiresAt = ExpiresAt,
@TargetId = TargetId,
@PayloadJson = PayloadJson,
@RequestId = ResultId
FROM dbo.AgentOperationIntent WITH (UPDLOCK, HOLDLOCK)
WHERE OperationId = @OperationId
AND TenantId = @TenantId
AND UserId = @UserId
AND OperationType = 'ORDER_CANCEL';
IF @Status IS NULL
BEGIN
ROLLBACK TRANSACTION;
THROW 51003, N'操作不存在或无权确认', 1;
END;
IF @Status = 'SUBMITTED'
BEGIN
COMMIT TRANSACTION;
SELECT
@OperationId AS OperationId,
@RequestId AS RequestId,
CAST(1 AS BIT) AS AlreadyProcessed;
RETURN;
END;
IF @Status <> 'PREPARED'
BEGIN
ROLLBACK TRANSACTION;
THROW 51004, N'操作状态不允许提交', 1;
END;
IF @ExpiresAt <= SYSUTCDATETIME()
BEGIN
UPDATE dbo.AgentOperationIntent
SET OperationStatus = 'EXPIRED'
WHERE OperationId = @OperationId;
COMMIT TRANSACTION;
THROW 51005, N'确认已过期，请重新发起', 1;
END;
SELECT @Reason = JSON_VALUE(@PayloadJson, '$.reason');
IF NOT EXISTS
(
SELECT 1
FROM dbo.Orders AS o WITH (UPDLOCK, HOLDLOCK)
WHERE o.OrderId = @TargetId
AND o.TenantId = @TenantId
AND o.OrderStatus IN ('PENDING', 'PAID')
AND o.IsDeleted = 0
)
BEGIN
ROLLBACK TRANSACTION;
THROW 51006, N'订单状态已变化，请重新确认', 1;
END;
INSERT INTO dbo.OrderCancelRequest
(
TenantId,
OrderId,
RequestedBy,
Reason,
RequestStatus,
SourceOperationId
)
VALUES
(
@TenantId,
@TargetId,
@UserId,
@Reason,
'PENDING',
@OperationId
);
SET @RequestId = SCOPE_IDENTITY();
UPDATE dbo.AgentOperationIntent
SET
OperationStatus = 'SUBMITTED',
ResultId = @RequestId,
ConfirmedAt = SYSUTCDATETIME()
WHERE OperationId = @OperationId;
INSERT INTO dbo.AgentToolAudit
(
TraceId,
TenantId,
UserId,
ToolName,
OperationId,
TargetType,
TargetId,
ParameterSummary,
ResultCode
)
VALUES
(
@TraceId,
@TenantId,
@UserId,
'confirm_order_cancel',
@OperationId,
'ORDER',
@TargetId,
NULL,
'SUBMITTED'
);
COMMIT TRANSACTION;
SELECT
@OperationId AS OperationId,
@RequestId AS RequestId,
CAST(0 AS BIT) AS AlreadyProcessed;
END;
GO
UPDLOCK, HOLDLOCK 让同一个 OperationId 的确认过程串行化。第一个请求提交成功后，后续请求读到 SUBMITTED，直接返回原来的 RequestId，不会重复插入。
事务里还重新检查了一次订单状态。用户看到确认框到真正点击确认之间，订单可能已经发货。准备阶段允许取消，不代表十分钟后的执行阶段仍然允许取消。
数据库账号只允许执行这两个过程
MCP Server 使用单独的数据库账号，不授予业务表写权限：
CREATE USER [mcp_writer] FOR LOGIN [mcp_writer];
GO
GRANT EXECUTE ON OBJECT::dbo.usp_AgentPrepareOrderCancel
TO [mcp_writer];
GO
GRANT EXECUTE ON OBJECT::dbo.usp_AgentConfirmOrderCancel
TO [mcp_writer];
GO
DENY INSERT, UPDATE, DELETE ON OBJECT::dbo.Orders
TO [mcp_writer];
GO
DENY INSERT, UPDATE, DELETE ON OBJECT::dbo.OrderCancelRequest
TO [mcp_writer];
GO
存储过程通过所有权链访问表，调用账号不需要直接获得表权限。实际部署前要按当前数据库 Owner、Schema Owner 和签名方式验证权限，不能只看到 GRANT EXECUTE 就默认一定生效。
我会使用 mcp_writer 单独登录数据库测试：执行两个存储过程应当成功，直接运行下面的语句必须失败。
UPDATE dbo.Orders
SET OrderStatus = 'CLOSED'
WHERE OrderId = 1001;
Python 侧取得可信身份
先定义一个请求身份对象：
from dataclasses import dataclass
@dataclass(frozen=True)
class Identity:
tenant_id: str
user_id: str
scopes: frozenset[str]
def require_scope(identity: Identity, required: str) -> None:
if required not in identity.scopes:
raise PermissionError("当前用户没有执行此操作的权限")
本机演示可以从环境变量读取：
import os
def get_identity() -> Identity:
scopes = frozenset(
value.strip()
for value in os.environ.get("MCP_SCOPES", "").split(",")
if value.strip()
)
return Identity(
tenant_id=os.environ["MCP_TENANT_ID"],
user_id=os.environ["MCP_USER_ID"],
scopes=scopes,
)
生产环境中的 get_identity() 必须由鉴权中间件提供。中间件验证令牌签名、Issuer、Audience、有效期和 Scope 后，再把身份绑定到当前请求上下文。
不要把它改成下面这样：
def confirm_order_cancel(
operation_id: str,
tenant_id: str,
user_id: str,
) -> dict:
...
一旦身份字段进入 Tool Schema，模型就能填写。身份应来自服务器已经验证的令牌，而不是工具参数。
MCP 官方授权教程建议在服务访问用户数据、数据库或需要审计时使用授权机制，并采用短期令牌、最小权限 Scope、Audience 校验和 HTTPS。MCP 官方授权教程
写 MCP Server
下面把数据库过程包成两个 Tool。代码为了便于阅读使用环境变量身份，远程部署时按上一节替换为请求级身份。
import hashlib
import json
import logging
import os
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Annotated
from mcp.server import MCPServer
from pydantic import Field
from sqlalchemy import create_engine, text
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("order-writer-mcp")
@dataclass(frozen=True)
class Identity:
tenant_id: str
user_id: str
scopes: frozenset[str]
def get_identity() -> Identity:
scopes = frozenset(
value.strip()
for value in os.environ.get("MCP_SCOPES", "").split(",")
if value.strip()
)
return Identity(
tenant_id=os.environ["MCP_TENANT_ID"],
user_id=os.environ["MCP_USER_ID"],
scopes=scopes,
)
def require_scope(identity: Identity, required: str) -> None:
if required not in identity.scopes:
raise PermissionError("当前用户没有执行此操作的权限")
def json_value(value):
if isinstance(value, datetime):
return value.isoformat()
if isinstance(value, Decimal):
return float(value)
if isinstance(value, uuid.UUID):
return str(value)
return value
def row_to_dict(row) -> dict:
return {
key: json_value(value)
for key, value in row._mapping.items()
}
engine = create_engine(
os.environ["SQLSERVER_URL"],
pool_pre_ping=True,
pool_size=5,
max_overflow=5,
)
mcp = MCPServer("Safe Order Writer")
@mcp.tool()
def prepare_order_cancel(
order_no: Annotated[
str,
Field(min_length=6, max_length=32, pattern=r"^[A-Za-z0-9_-]+$"),
],
reason: Annotated[
str,
Field(min_length=2, max_length=200),
],
) -> dict:
"""检查订单并生成取消申请预览。本工具不会取消订单，也不会创建取消申请。"""
identity = get_identity()
require_scope(identity, "order.cancel.request")
operation_id = uuid.uuid4()
trace_id = uuid.uuid4()
expires_at = datetime.now(UTC) + timedelta(minutes=10)
payload = {
"order_no": order_no,
"reason": reason.strip(),
}
payload_json = json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
payload_hash = hashlib.sha256(payload_json.encode("utf-8")).hexdigest()
statement = text(
"""
EXEC dbo.usp_AgentPrepareOrderCancel
@OperationId = :operation_id,
@TraceId = :trace_id,
@TenantId = :tenant_id,
@UserId = :user_id,
@OrderNo = :order_no,
@Reason = :reason,
@PayloadJson = :payload_json,
@PayloadHash = :payload_hash,
@ExpiresAt = :expires_at
"""
)
try:
with engine.connect() as connection:
row = connection.execute(
statement,
{
"operation_id": operation_id,
"trace_id": trace_id,
"tenant_id": identity.tenant_id,
"user_id": identity.user_id,
"order_no": order_no,
"reason": reason.strip(),
"payload_json": payload_json,
"payload_hash": payload_hash,
"expires_at": expires_at,
},
).fetchone()
connection.commit()
preview = row_to_dict(row)
logger.info(
"tool=prepare_order_cancel trace_id=%s operation_id=%s tenant=%s user=%s",
trace_id,
operation_id,
identity.tenant_id,
identity.user_id,
)
return {
"prepared": True,
"requires_user_confirmation": True,
"operation_id": str(operation_id),
"preview": preview,
"message": "请向用户展示订单号、金额、取消原因和有效期。只有用户明确确认后才能提交。",
}
except Exception:
logger.exception(
"tool=prepare_order_cancel failed trace_id=%s tenant=%s user=%s",
trace_id,
identity.tenant_id,
identity.user_id,
)
raise RuntimeError("暂时无法生成取消预览，请稍后重试")
@mcp.tool()
def confirm_order_cancel(
operation_id: Annotated[
str,
Field(pattern=r"^[0-9a-fA-F-]{36}$"),
],
) -> dict:
"""提交已经准备好的订单取消申请。客户端调用本工具前必须取得用户确认。"""
identity = get_identity()
require_scope(identity, "order.cancel.request")
try:
parsed_operation_id = uuid.UUID(operation_id)
except ValueError as exc:
raise ValueError("operation_id 格式不正确") from exc
trace_id = uuid.uuid4()
statement = text(
"""
EXEC dbo.usp_AgentConfirmOrderCancel
@OperationId = :operation_id,
@TraceId = :trace_id,
@TenantId = :tenant_id,
@UserId = :user_id
"""
)
try:
with engine.connect() as connection:
row = connection.execute(
statement,
{
"operation_id": parsed_operation_id,
"trace_id": trace_id,
"tenant_id": identity.tenant_id,
"user_id": identity.user_id,
},
).fetchone()
connection.commit()
result = row_to_dict(row)
logger.info(
"tool=confirm_order_cancel trace_id=%s operation_id=%s tenant=%s user=%s already_processed=%s",
trace_id,
parsed_operation_id,
identity.tenant_id,
identity.user_id,
result["AlreadyProcessed"],
)
return {
"submitted": True,
"operation_id": str(parsed_operation_id),
"request_id": result["RequestId"],
"already_processed": bool(result["AlreadyProcessed"]),
"message": "取消申请已提交，订单尚未被直接删除或关闭。",
}
except Exception:
logger.exception(
"tool=confirm_order_cancel failed trace_id=%s operation_id=%s tenant=%s user=%s",
trace_id,
parsed_operation_id,
identity.tenant_id,
identity.user_id,
)
raise RuntimeError("取消申请提交失败，请根据 operation_id 查询结果后再决定是否重试")
if __name__ == "__main__":
mcp.run("streamable-http")
这里没有让模型传 confirmed=True。模型自己填写的布尔值不能证明用户确认过。确认动作应由 MCP Host 或业务前端在展示参数后触发，并把 confirm_order_cancel 配置为始终需要人工批准。
不同客户端的审批配置方式不同，下面只是策略含义，不是通用配置文件：
tools:
prepare_order_cancel:
approval: never
confirm_order_cancel:
approval: always
即使客户端已经弹过确认框，MCP Server 仍然要检查 Scope、租户、操作归属、有效期和订单实时状态。客户端确认解决的是“用户是否同意”，服务端校验解决的是“这个用户现在是否有权执行”。两者不能互相替代。
幂等键为什么不能每次临时生成
很多文章会给写工具增加一个 idempotency_key 参数，然后让模型自己生成 UUID：
def cancel_order(order_no: str, idempotency_key: str):
...
这只解决了一半问题。如果超时重试时模型重新生成一个 UUID，数据库看到的是两次不同请求，仍然会重复执行。
这次使用 operation_id 的原因就在这里：它由准备阶段生成，后续确认和重试都必须复用同一个值。
准备操作：operation_id = A
首次确认：operation_id = A
超时重试：operation_id = A
并发重试：operation_id = A
数据库通过状态锁和唯一约束保证只有一张取消申请。即使客户端没有收到第一次成功响应，也可以安全地用同一个 operation_id 重试。
如果业务接口本身已经支持幂等键，MCP Server 应把 operation_id 原样传给内部 API，不要在中间层重新生成。
不要把数据库异常原样返回给模型
存储过程会抛出明确的业务错误，但线上不应该把连接字符串、SQL、表名和堆栈全部返回给 Agent。
对外可以使用稳定错误码：
{
"success": false,
"error_code": "ORDER_STATE_CHANGED",
"message": "订单状态已变化，请重新确认"
}
服务端日志再记录详细信息：
trace_id
operation_id
tenant_id
user_id
tool_name
database_error_number
elapsed_ms
这样 Agent 知道下一步应该重新准备操作，运维也能通过 trace_id 找到真实异常。
还有一个细节：确认接口超时，不要立即告诉用户“提交失败”。超时只代表客户端没有及时收到结果，不代表事务一定回滚。更准确的提示是“提交结果未知，请根据 operation_id 查询后再重试”。
生产环境最好再提供一个只读 Tool：
get_operation_result(operation_id)
它只能查询当前租户、当前用户创建的操作，用于处理网络超时后的结果确认。
怎么测试这套写入流程
正常流程只能证明代码会跑，下面这些测试才决定它能不能上线。
重复确认
同一个 operation_id 连续调用两次 confirm_order_cancel：
第一次：创建 RequestId=1008，AlreadyProcessed=false
第二次：仍返回 RequestId=1008，AlreadyProcessed=true
数据库里只能有一条 OrderCancelRequest。
并发确认
用十个并发请求确认同一个 operation_id。最终仍然只能生成一条取消申请，其余请求返回相同结果。
跨租户确认
租户 A 准备操作，换成租户 B 的令牌确认。数据库查询不到该操作，应返回无权确认，不能因为知道 operation_id 就执行成功。
换用户确认
同一租户内，用户 A 准备操作，用户 B 尝试确认。除非业务明确允许代办，否则也应拒绝。
操作过期
把 ExpiresAt 调整到当前时间之前，再执行确认，应把状态改为 EXPIRED，并要求重新准备。
订单状态变化
准备时订单是 PAID，确认前订单已经发货。确认阶段必须重新检查状态，不能按十分钟前的预览继续提交。
数据库权限绕过
使用 mcp_writer 登录后直接执行 UPDATE Orders，必须被数据库拒绝。
事务回滚
在插入取消申请后人为制造异常，确认操作意图、取消申请和审计日志不会出现一半成功、一半失败。
上线前再补几道限制
写操作比查询更适合做严格限流。例如同一个用户一分钟只能准备五次取消操作，同一个订单不能同时存在多条待处理申请。
还可以增加这些约束：
- 取消原因长度和字符范围；
- 单用户、单租户的调用频率；
- 同一订单的待处理申请唯一性；
- 操作意图十分钟后失效；
- 高金额订单需要更高等级 Scope；
- 非工作时间进入人工队列；
- 所有写 Tool 默认要求客户端确认；
- 审计日志进入独立、不可随业务数据一起修改的存储。
远程 MCP Server 不能只靠一个写死的共享 Token。当前 MCP 授权机制把受保护的 MCP Server 作为资源服务器，要求验证访问令牌确实签发给当前资源；安全建议也明确反对 Token Passthrough，并要求正确验证 Audience。MCP 授权规范
如果已有内部 API，就不要绕过它直写数据库
本文使用存储过程，是为了把幂等、事务和数据库权限讲清楚。现实项目里，如果订单系统已经有成熟的取消申请 API，MCP Server 应优先调用原 API：
AI Agent
↓
MCP Server
├─ 验证身份和 Scope
├─ 生成操作预览
├─ 检查用户确认
└─ 传递 operation_id
↓
订单系统 API
├─ 校验订单状态
├─ 保证幂等
├─ 执行业务事务
└─ 写业务审计
原系统通常已经处理库存、优惠券、支付、消息通知等联动。MCP Server 直接改一张订单表，很可能让订单状态变了，库存和支付状态却没有跟上。
MCP 是面向 Agent 的能力入口，不应该变成绕开原有业务层的后门。
最后
让 Agent 查数据库，重点是控制它能看什么；让 Agent 写数据库，重点变成控制它何时能做、能做几次，以及失败后能不能知道究竟做没做。
这套两阶段方案多了一次准备操作和一次用户确认，看起来没有直接 UPDATE 那么痛快，但它把几个关键问题放回了代码和数据库：
- 用户确认前不产生核心业务副作用；
- 操作目标和参数在准备阶段被冻结；
- 确认操作绑定当前租户和用户；
- 重试和并发不会重复创建申请；
- 订单状态在真正写入前重新检查；
- 业务写入、状态更新和审计记录处于同一事务。
模型仍然负责理解用户意图和选择工具，但权限、确认、事务、幂等和审计不依赖模型自觉。这才是把 MCP Server 接到真实业务系统时，比较踏实的分工方式。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
3
0- 0
已为社区贡献8条内容
所有评论(0)