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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。它不替代你的 MCP Server也不替代数据库只解决模型调用这一层的 Key 管理和通道统一。具体来说你需要先在控制台创建一个 API Key然后把它配置到 Agent 运行环境里。MCP Server 侧仍然使用自己的访问令牌体系两者不要混用。模型 Key 用于模型对话和 Tool 选择MCP 访问令牌用于确认租户身份职责分开。创建 Key 的入口在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 管理页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。如果你只是先验证模型能不能正常调 Tool可以用模型对话页面快速试一下https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。长期跑编码类 Agent 或者多租户自动化任务建议看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Claude Code 相关配置参考 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 。注意TaoToken 的 Key 只用于模型调用通道不要把它当作 MCP Server 的租户身份令牌。两者混用会让越权排查变得非常困难。3. 可复制配置TenantId 注入与 RLS 策略骨架3.1 先定义不可变的请求身份我会在进入 MCP Tool 之前把令牌中的身份整理成一个不可变对象。这样后续任何一层都不能悄悄改写租户。from dataclasses import dataclass dataclass(frozenTrue) 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, defaultNone, ) 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_length6, max_length32, patternr^[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 参数、聊天记录、长期记忆也不来自模型生成的摘要。查不到时统一返回 foundfalse不要区分“订单不存在”和“订单存在但属于其他租户”后者会暴露订单号是否有效。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(Ntenant_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 Ntenant_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 ftenant:{tenant_id}:order:{order_no}向量库过滤应该进入服务端查询而不是先搜再过滤results vector_store.search( queryquery, filters{ tenant_id: identity.tenant_id, visibility: agent, }, limit5, )日志只记录 trace_id、tenant_id、user_id、tool_name、参数摘要或哈希、结果条数、耗时、结果码。后台任务消息中保存经过服务端确认的身份快照Worker 执行时重新检查租户和资源归属。4. 验证请求用 TaoToken 统一通道跑越权测试配置完成后用 TaoToken 的模型对话页面或 API 通道发起一次 Tool 调用验证。模型侧配置示例export TAOTOKEN_API_KEY你的Key export TAOTOKEN_BASE_URLhttps://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 页面创建 Keyhttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入细节看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。验证模型能否正确选择 Tool用模型对话页面最快https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。长期跑多租户 Agent 任务建议直接上 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。