☰
AI Agent 技能分享|MCP Server 多租户隔离实战:用 TenantId 与 Row-Level Security 守住数据边界,TaoToken 统一 Key 通道配置
2026/9/28 19:24:08 网站建设 项目流程

1. 为什么 Agent 接入 MCP Server 后,租户隔离反而更难做了

订单系统接上 MCP Server 之后,我最担心的不是模型把订单号说错,而是它把别人的订单说对了。多租户系统里很多表共用一套数据库,只靠 TenantId 区分数据。普通后台页面的查询条件写在固定代码里,相对容易检查;Agent 的输入更自由,同一个 Tool 还可能被不同用户、不同客户端反复调用。只要某一层漏掉租户条件,就可能形成真实的数据泄露。

这篇继续用订单系统做例子,把租户隔离放到四层:访问令牌确认当前用户属于哪个租户;MCP Tool 不接受模型传入的 tenant_id;数据访问层每条查询强制携带 TenantId;数据库用 Row-Level Security 再兜一层。缓存、向量库、日志和异步任务也要使用同一个租户边界,只保护主查询,其他地方迟早会漏。

如果你正在用 AI Agent 调 MCP Server 查业务数据,这篇的配置骨架可以直接抄。下面所有示例都基于一个统一 Key 通道来管理模型侧调用,避免每个租户各自维护一套密钥导致审计困难。

2. TaoToken 前置:统一 Key 通道与 MCP Server 的关系

MCP Server 本身不负责模型调用,但 Agent 在决定调用哪个 Tool、传什么参数时,需要经过模型。多租户场景下,如果每个租户、每个客户端都各自配一套模型 Key,审计和限流会非常混乱。我试过用 TaoToken 做统一 Key 通道,把模型侧调用收敛到一个入口,MCP Server 只关心业务身份和租户边界。

TaoToken 的定位是统一模型 API 通道,官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 入口是 https://taotoken.net/api 。它不替代你的 MCP Server,也不替代数据库,只解决模型调用这一层的 Key 管理和通道统一。

具体来说,你需要先在控制台创建一个 API Key,然后把它配置到 Agent 运行环境里。MCP Server 侧仍然使用自己的访问令牌体系,两者不要混用。模型 Key 用于模型对话和 Tool 选择,MCP 访问令牌用于确认租户身份,职责分开。

创建 Key 的入口在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ,Key 管理页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。如果你只是先验证模型能不能正常调 Tool,可以用模型对话页面快速试一下:https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。

长期跑编码类 Agent 或者多租户自动化任务,建议看 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,Claude Code 相关配置参考 https://taotoken.net/claudecode-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite 。

注意:TaoToken 的 Key 只用于模型调用通道,不要把它当作 MCP Server 的租户身份令牌。两者混用会让越权排查变得非常困难。

3. 可复制配置:TenantId 注入与 RLS 策略骨架

3.1 先定义不可变的请求身份

我会在进入 MCP Tool 之前,把令牌中的身份整理成一个不可变对象。这样后续任何一层都不能悄悄改写租户。

from dataclasses import dataclass @dataclass(frozen=True) class RequestIdentity: subject: str tenant_id: str scopes: frozenset[str] token_id: str | None = None def require_scope(identity: RequestIdentity, required: str) -> None: if required not in identity.scopes: raise PermissionError("当前用户没有调用该工具的权限")

远程 MCP Server 应验证令牌的签名、Issuer、Audience、过期时间和 Scope。多租户服务要固定可信 Issuer,拒绝其他 Realm 的令牌,并验证令牌是否真正签发给当前 MCP Server。

请求身份进入 ContextVar,避免多个并发请求互相覆盖:

from contextvars import ContextVar, Token current_identity: ContextVar[RequestIdentity | None] = ContextVar( "current_identity", default=None, ) def bind_identity(identity: RequestIdentity) -> Token: return current_identity.set(identity) def reset_identity(token: Token) -> None: current_identity.reset(token) def get_verified_identity() -> RequestIdentity: identity = current_identity.get() if identity is None: raise PermissionError("请求尚未完成身份认证") return identity

不要把身份放进普通全局变量。异步服务同时处理租户 A 和租户 B 的请求时,全局变量可能被后一个请求覆盖。

3.2 MCP Tool 只接受业务参数

下面这个 Tool 看起来规规矩矩,SQL 也做了参数化,但 tenant_id 出现在 Tool Schema 中,模型可以填写它:

@mcp.tool() def get_order(tenant_id: str, order_no: str) -> dict: sql = text( """ SELECT OrderNo, OrderStatus, TotalAmount FROM dbo.Orders WHERE TenantId = :tenant_id AND OrderNo = :order_no """ ) ...

问题在于权限边界已经交给了模型。用户可以通过提示词影响模型,让它去查别的租户。正确的 Tool 只接受业务参数:

from typing import Annotated from pydantic import Field @mcp.tool() def get_order_status( order_no: Annotated[ str, Field(min_length=6, max_length=32, pattern=r"^[A-Za-z0-9_-]+$"), ], ) -> dict: identity = get_verified_identity() require_scope(identity, "order.read") order = repository.get_order(identity.tenant_id, order_no) if order is None: return {"found": False, "order": None} return {"found": True, "order": order}

tenant_id 来自经过验证的访问令牌或可信运行环境,不来自 Tool 参数、聊天记录、长期记忆,也不来自模型生成的摘要。查不到时统一返回 found=false,不要区分“订单不存在”和“订单存在但属于其他租户”,后者会暴露订单号是否有效。

3.3 Repository 不允许省略 tenant_id

我会让数据访问层的所有公开方法都要求传入租户:

from sqlalchemy import text class OrderRepository: def __init__(self, engine): self.engine = engine def get_order(self, tenant_id: str, order_no: str) -> dict | None: statement = text( """ SELECT TOP (1) OrderNo, OrderStatus, TotalAmount, CreatedAt FROM dbo.v_AgentOrderSummary WHERE TenantId = :tenant_id AND OrderNo = :order_no """ ) with self.engine.connect() as connection: row = connection.execute( statement, { "tenant_id": tenant_id, "order_no": order_no, }, ).fetchone() return dict(row._mapping) if row else None

不提供 get_order(order_no) 这样的重载,也不允许 Repository 自己使用某个默认租户。默认值在单租户 Demo 中很方便,到了共享数据库里就是隐患。

3.4 SQL Server Row-Level Security 兜底

应用层条件必须保留,但我不想把全部希望放在开发人员每次都记得写 TenantId 上。SQL Server 可以使用 Row-Level Security 做数据库兜底。

先创建安全谓词函数:

CREATE SCHEMA Security; GO CREATE FUNCTION Security.fn_TenantPredicate ( @TenantId NVARCHAR(64) ) RETURNS TABLE WITH SCHEMABINDING AS RETURN ( SELECT 1 AS Allowed WHERE @TenantId = CONVERT( NVARCHAR(64), SESSION_CONTEXT(N'tenant_id') ) ); GO

再把策略绑定到订单表:

CREATE SECURITY POLICY Security.OrderTenantPolicy ADD FILTER PREDICATE Security.fn_TenantPredicate(TenantId) ON dbo.Orders, ADD BLOCK PREDICATE Security.fn_TenantPredicate(TenantId) ON dbo.Orders AFTER INSERT, ADD BLOCK PREDICATE Security.fn_TenantPredicate(TenantId) ON dbo.Orders AFTER UPDATE WITH (STATE = ON); GO

每次取得数据库连接后,先写入当前租户:

from sqlalchemy import text def set_tenant_context(connection, tenant_id: str) -> None: connection.execute( text( """ EXEC sys.sp_set_session_context @key = N'tenant_id', @value = :tenant_id, @read_only = 1 """ ), {"tenant_id": tenant_id}, )

查询时两层条件一起存在:

with engine.connect() as connection: set_tenant_context(connection, identity.tenant_id) row = connection.execute(statement, params).fetchone()

应用层的 WHERE TenantId = :tenant_id 方便阅读、索引选择和代码审查;RLS 用来防止某次新增查询忘记租户条件。两层作用不同,不建议因为有 RLS 就删掉应用层条件。

连接池场景要特别小心。连接会被不同请求复用,租户上下文必须在每次借出连接后设置,不能只在创建连接时设置。最好使用事务或统一连接包装器,避免业务代码绕过初始化步骤。

3.5 写操作、缓存、向量库、日志、后台任务

更新语句不能只按订单号定位:

-- 不推荐 UPDATE dbo.Orders SET OrderStatus = 'CLOSED' WHERE OrderNo = @OrderNo; -- 推荐 UPDATE dbo.Orders SET OrderStatus = 'CLOSED' WHERE TenantId = @TenantId AND OrderNo = @OrderNo AND OrderStatus = 'PENDING'; IF @@ROWCOUNT <> 1 THROW 51001, N'订单不存在、无权访问或状态不允许修改', 1;

缓存 Key 至少包含租户、资源类型和业务主键:

def order_cache_key(tenant_id: str, order_no: str) -> str: return f"tenant:{tenant_id}:order:{order_no}"

向量库过滤应该进入服务端查询,而不是先搜再过滤:

results = vector_store.search( query=query, filters={ "tenant_id": identity.tenant_id, "visibility": "agent", }, limit=5, )

日志只记录 trace_id、tenant_id、user_id、tool_name、参数摘要或哈希、结果条数、耗时、结果码。后台任务消息中保存经过服务端确认的身份快照,Worker 执行时重新检查租户和资源归属。

4. 验证请求:用 TaoToken 统一通道跑越权测试

配置完成后,用 TaoToken 的模型对话页面或 API 通道发起一次 Tool 调用验证。模型侧配置示例:

export TAOTOKEN_API_KEY="你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

准备两个租户,各自创建相同订单号:

租户订单号金额
tenant_aORDER-001100 元
tenant_bORDER-001900 元

然后执行这些测试:

A 的令牌查询 ORDER-001,只能返回 100 元;B 的令牌查询相同订单号,只能返回 900 元;Tool 参数中出现 tenant_b 字样,不会改变当前身份;A 的令牌尝试确认 B 创建的 operation_id,必须失败;缓存预热后交换请求顺序,结果仍然正确;向量库放入高度相似的两份文档,只能召回当前租户版本;使用数据库账号执行遗漏 TenantId 的查询,RLS 仍只返回当前租户;没有设置 SESSION_CONTEXT 的连接应返回空结果,而不是全库数据。

最后一条很关键。安全默认值应该是“没有身份就看不到数据”,不能是“没有身份就跳过过滤”。

成功结果应该类似这样:

{ "found": true, "order": { "OrderNo": "ORDER-001", "OrderStatus": "PENDING", "TotalAmount": 100 } }

如果 A 的令牌返回了 900 元,说明某一层租户条件被绕过了,优先检查 Tool Schema 是否还暴露 tenant_id、连接池是否复用了上一个租户的 SESSION_CONTEXT、缓存 Key 是否缺少租户前缀。

5. 本篇常见错排查

5.1 Tool Schema 里还有 tenant_id

这是最常见的越权入口。模型看到 tenant_id 参数就会尝试填写,用户也可以通过提示词影响它。检查所有 @mcp.tool() 装饰的函数签名,确保没有租户相关参数。

5.2 连接池复用导致 SESSION_CONTEXT 串租户

连接归还池中后,SESSION_CONTEXT 可能保留上一个租户的值。必须在每次借出连接后重新设置,或者使用 sp_reset_connection 清理。测试方法:租户 A 查询后立即用租户 B 查询同一订单号,看是否返回错误数据。

5.3 缓存 Key 缺少租户前缀

如果不同租户可以使用相同订单号,缓存 Key 只写 order:{order_no} 会导致后查询的租户命中先前租户的数据。检查所有缓存、限流、幂等、分布式锁的 Key 命名。

5.4 向量库先搜后过滤

先取 top 50 再在 Python 里过滤,等于把其他租户的数据拉到了应用内存。过滤条件必须进入向量数据库的服务端查询,或者按租户拆分 Collection。

5.5 日志和后台任务泄露租户数据

日志平台权限通常比业务数据库更宽,保留时间也更长。后台任务如果只传 job_id 不传租户快照,Worker 可能使用默认租户执行。检查日志字段和任务消息结构。

5.6 RLS 策略未覆盖 INSERT/UPDATE

只加 FILTER PREDICATE 只能控制查询,写操作还需要 BLOCK PREDICATE。否则 Agent 可能通过写操作影响其他租户的数据。

6. 继续接入与验证

多租户隔离不是在 SQL 后面补一句 TenantId = ? 就结束了。身份从哪里来、连接池怎么复用、缓存如何命名、向量检索在哪里过滤、后台任务怎样恢复上下文,任何一个环节都可能成为泄露入口。

我比较认可的分工是:令牌负责证明当前是谁,应用层负责把身份带到每次操作,数据库负责在查询遗漏时兜底。模型只填写订单号、状态、日期这类业务参数,不参与决定自己属于哪个租户。只要 tenant_id 还能被模型、用户输入或历史摘要改写,这条边界就还没有真正建立起来。

如果你要先把模型调用通道统一起来,可以从 API Keys 页面创建 Key:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ,接入细节看文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。验证模型能否正确选择 Tool,用模型对话页面最快:https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。长期跑多租户 Agent 任务,建议直接上 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询