FastMCP 无状态会话状态Stateless Session State完整指南基于认证主体的服务端状态隔离方案【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp导读本文系统讲解 FastMCP 在 2026-07-28 协议时代提出的无状态会话状态stateless session state设计。现代 MCP 协议按构造即无状态——每个请求都会构建全新的Connection请求一返回连接内内存状态即被丢弃。本文深入剖析 FastMCP 如何以**认证主体authenticated principal**而非会话 ID 作为隔离墙通过session: UserSession依赖注入与session_id: SessionId显式参数两种模式在无 TTL 约束的服务端存储上提供跨调用状态购物车、对话、累积上下文并给出源码级实现原理、完整可运行示例与安全边界分析。读完你将掌握如何用注入式按用户状态、如何用 create-then-pass 会话生命周期、如何配置持久化存储以及无认证场景下必须遵守的安全约束。背景协议天然无状态带来的静默丢失问题在 2026-07-28 时代的协议构造中每个请求都会构建一个全新的Connectionconnection.session_id恒为Noneconnection.state是一个全新的 dict请求返回即被丢弃。这意味着ctx.session_id每次请求都会临时生成一个uuid4而ctx.set_state/ctx.get_state写进去的数据永远不会往返round-trip——不报错只是静默丢数据。想实现跨调用状态购物车、对话上下文、累积上下文的用户没有任何安全机制可用且失败完全不可见。问题的关键在于现代请求中唯一稳定且不可伪造的标识符是认证主体——get_access_token().claims[sub]或者(client_id, issuer, subject)三元组。线上传输的其他一切标识符都由客户端声明、都可被伪造。这正是新设计的起点。核心模型服务端存储 主体隔离 不设 TTL新模型将状态放在服务端位于服务器已持有的唯一AsyncKeyValuepy-key-value存储中——即session_state_store。框架只调用get/put/delete从不施加 TTL——保留策略完全由存储决定在你传入的存储上配置Redis TTL、py-key-value TTL 包装器均可。不存在第二个存储也不存在框架自有的 TTL 旋钮。从源码看server.py 中FastMCP.__init__接收session_state_store: AsyncKeyValue | None None存放在self._state_storage并在惰性初始化后通过server._state_store暴露给会话层使用。隔离来自认证主体而非会话 ID。状态以(principal, session_id)为键。主体 B 发起的请求只进入 B 自己的命名空间——无论它传什么session_id都永远无法寻址 A 的键ID 只用于组织主体内部的会话。句柄是裸的uuid4字符串不进行密封sealed——主体前缀才是隔离墙。会话遵循先创建后校验create-then-validate一个从未由当前主体下的create_session铸造过的 ID 会被直接拒绝而不会解析成一个空会话。存储键格式见 sessions.py 的_principal_segment与session_storage_key# 认证请求principal 被 sha256 哈希成定长、分隔符安全的键段 session:{sha256(principal)}:{session_id} # 未认证请求折叠到共享的 anon 命名空间 session:anon:{session_id}哈希处理保证了任意主体字符串都无法注入:键分隔符也限制了键长主体身份issuer、client_id、subject永远不会以原文嵌入存储键或日志。两种显式模式按用户注入或按会话传参一个工具有目的地、精确地选择其中一种模式。设计上刻意不存在给 ID 就用、不给就用默认的可选参数——那会让 Agent 忘记传 ID 的调用静默路由进共享的每用户桶而这正是该特性要消除的不可见劣化故障。模式一每用户状态 —— 依赖注入Per-user state, injectedfrom fastmcp.server.sessions import UserSession mcp.tool async def remember(fact: str, session: UserSession) - str: await session.set(fact, fact) return notedsession: UserSession是依赖注入的与ctx: Context相同以请求的认证主体为键不出现在输入 schema 中Agent 无需传递任何东西。它要求认证——没有主体时抛出清晰的错误。适用于每个用户一个桶正是你想要的场景。从源码看sessions.py 中UserSession仅是注入注解它本身继承自Session是为了让框架基于类型的注入检测器能够识别它自身不添加行为处理器实际收到的是一个普通Session因此get/set/delete/clear访问器照常工作。注入过程由_CurrentSession依赖sessions.py实现无认证主体时抛SessionAuthError若声明session: UserSession | None则可注入None以支持未认证请求上的可选分支。底层通过current_principal()sessions.py读取当前请求的 token 并序列化(client_id, issuer, subject)三元组——注意两个使用同一 OAuth client 的用户只要 token 验证器提供了 subject就是不同的主体。测试佐证见 test_session_provider.pysession不出现在 input_schema 的属性中未认证调用抛出ToolError同一主体连续两次调用后购物车累积为[apple, banana]不同主体user-a / user-b各自得到隔离的桶。模式二独立会话 —— 显式参数Distinct sessions, an argumentfrom fastmcp.server.sessions import SessionId from fastmcp.server.dependencies import get_session mcp.tool async def add_to_cart(item: str, session_id: SessionId) - str: session await get_session(session_id) cart await session.get(cart, default[]) cart.append(item) await session.set(cart, cart) return f{len(cart)} itemssession_id: SessionId是一个必填的字符串参数——它确实在 schema 中由 Agent 提供。SessionId是标记类型marker type框架会为该参数的描述自动填充协议说明Session identifier. Use a tool to create a session, then pass the resulting id here to persist state across calls in the same session.于是工具变成自我教学的——Agent 读取 schema 就能学会先创建、再传递的契约无需任何手工提示。描述中不指名任何具体工具组合composition可以重命名生命周期工具挂载到命名空间下会暴露为child_create_session因此描述指向能力而非某个在挂载下可能不存在的名字。实现上SessionId定义为Annotated[str, _SessionIdMarker()]sessions.py。schema 构建时function_parsing.py 通过session_id_parameter_names(fn)扫描Annotated[str, _SessionIdMarker()]元数据将SESSION_ID_DESCRIPTIONsessions.py追加到作者已有描述之后而非覆盖。检测器还会展开functools.partial已按位置绑定的参数从工具实参表面剔除、按关键字绑定的仍保留确保与工具真实的参数表面一致详见 sessions.py 及测试 test_sessions.py。独立的await get_session(session_id)dependencies.py把 ID 解析为以(principal, session_id)为键的Session并校验它是在当前主体下创建的——未知或外部 ID 抛InvalidSession而不是打开一个全新桶。错误信息刻意保持笼统Invalid or unknown session.具体原因哪个 ID、哪个主体只记 debug 日志避免攻击者区分未知 ID与属于别人的 ID见 sessions.py。它是普通函数而非Context方法因此不需要前台上下文可以从taskTrue工具的 worker 中调用它和get_server()一样通过任务感知的 server 解析器工作。当用户需要多个会话时使用此模式。Session 对象单键单字典的异步访问器Sessionsessions.py是对服务端存储的异步访问器作用域限定为一个(principal, session_id)桶session.id—— 会话 ID对session_id解析出的会话有效注入的UserSession返回None因为它没有独立 ID内部基于主体的键刻意不对外暴露await session.get(key, defaultNone)await session.set(key, value)await session.delete(key)await session.clear()—— 清空用户状态但保持会话有效await session.end()—— 删除会话end_session调用的就是它会话状态以单字典存于单个键下。该字典在state子字典中保存用户状态并附带一个_created标记因此已创建但为空的会话与缺失的会话可以区分开即使存储会折叠空字典。要点get/set/delete对子字典做读-改-写绝不触碰标记clear重置子字典但保留标记会话仍可解析end删除整个键。用户状态被命名空间隔离在state之下这是用户即使把键命名为_created也不会与标记冲突的原因test_sessions.py 专门验证了这一点。每会话一个键意味着每会话一个 TTL即存储的 TTL写入时刷新——无需维护键索引end就是一次删除。权衡同一会话的并发写入会在读-改-写上竞争会话状态通常很小且由单个 Agent 串行驱动因此可接受——这是被明确记录而非隐藏的取舍。SessionProvider会话 ID 的铸造与终结会话 ID 由SessionProvider铸造它贡献两个工具create_session()—— 铸造一个不可猜测的uuid4在当前主体下记录该会话返回字符串 IDend_session(session_id: SessionId)—— 校验 ID 后删除会话使其不再可解析。当工具接收session_id时注册它——provider 正是添加此类功能的惯用方式from fastmcp.server.sessions import SessionProvider mcp.add_provider(SessionProvider())从源码看sessions.py 中create_session()实际铸造str(uuid4())并调用session._create()写入_created标记与空的state子字典end_session()走get_session校验后调用session.end()删除键。SessionProvider(Provider)不持有存储用服务器的也不设 TTL用存储的只负责铸造与终结归属于自己的 ID。关于不强制注册的演进删除过时的扫描检查设计文档明确指出不强制注册 provider——而且曾经有过。早期版本在 list/resolve 时扫描工具集若发现session_id工具没有 provider 就抛错。但这个检查必须推理整个组合管线——对 provider 做isinstance、解包命名空间化的 provider、工具变换、会话可见性、启用状态——产生了误报破坏了合法服务器命名空间化的 provider、会话被禁用的工具。它已被删除。该保证从来不需要它get_session会校验 ID 是否被记录create-then-validate因此没有 provider 的服务器根本无法铸造 ID每次get_session都会拒绝——错误配置在工具首次运行时即被发现而不是一个安全漏洞。已自行铸造 ID 的应用应如何选择当应用已经自行铸造标识符对话 ID、工作流 ID时把它们当作普通字符串参数而非SessionId且不要注册 providerSessionId特指由create_session支撑的 create-then-pass 契约。create_session在无认证场景下意义最大——此时不可猜测的 ID 是防御调用者猜测进他人会话的唯一手段。安全模型两层隔离主体即边界以(principal, session_id)为键的安全含义分两层已认证 → 强隔离。principal是经过校验的 token 主体不可伪造。B 进入 B 的命名空间无论 B 传什么 IDA 的数据都不可达。猜测毫无意义出现在 Agent 上下文或日志中的会话 ID 也无害没有 principal 它就不是能力凭证。调用者自选的 ID 在这里是安全的。未认证 → 仅单租户安全。没有 principal键就只是共享命名空间中的 IDID 变成 bearer 能力暴露在日志/对话中就会泄露会话。create_session的uuid4提供的是猜测抵抗不是隔离。文档加粗声明不是租户边界没有认证时强制使用铸造 ID绝不要把会话当作客户端之间的墙。一句话总结设计哲学隔离来自认证Isolation is authID 只是组织the id is organization。没有任何 ID 方案能替代 principal——这正是为什么密封句柄不带来任何承重价值、从而被放弃的原因。同时不是 FastMCP 职责范围的是传输安全用 TLS、静态加密存储的职责、以及一个恶意的已授权客户端在其权限范围内行事。测试对这两层均有覆盖test_sessions.py 验证同一 ID 在不同主体下键不同、B 传 A 的 ID 只到达自己的空桶test_session_provider.py 以模拟认证主体驱动完整的注入 存储路径。配置存储会话持久化落点会话状态驻留在服务器配置的session_state_store中。配置方式from fastmcp import FastMCP # 传入一个 AsyncKeyValue 存储例如 py-key-value 适配器 server FastMCP(my-server, session_state_storemy_store)框架对该存储只调用get/put/delete不施加任何 TTL保留策略Redis TTL、py-key-value TTL 包装器等完全由你传入的存储决定。这意味着会话多久过期是部署层面的策略问题而非框架层面的旋钮——每会话一个键因此每会话一个 TTL写入时刷新。注意该存储由所有请求共享见 server.py 中Session-scoped state store (shared across all requests)的注释。迁移说明从旧版 ctx.session_id / set_state 到新模式旧 APIctx.session_id、ctx.set_state/ctx.get_state见 context.py 与set_state/get_state实现在无状态协议下是静默丢数据的session_id每次请求新铸uuid4set_state写入的连接内 dict 请求结束即弃。迁移路径按用户持久状态把ctx.set_state(key, v)/ctx.get_state(key)迁移为注入session: UserSession参数的await session.set(key, v)/await session.get(key)——同样不出现在输入 schema 中但数据真正落到服务端存储并跨请求存活多会话/跨调用会话迁移为session_id: SessionId参数 SessionProvider的 create-then-pass 生命周期原有依赖ctx.session_id生成请求内唯一键如日志关联的场景其语义与新会话体系不同需要单独评估。一个关键差异值得注意旧ctx.session_id对非 HTTP 传输会即时生成一个 IDcontext.py 中from uuid import uuid4的路径而新模型里未铸造的 ID 一律被get_session拒绝——从自动给你一个 ID到必须先铸造、后使用这正是消除静默丢失、让失败可见的核心转变。与重写计划的对应从原型到最终 API设计文档附带的 rework plan 明确列出了从当前原型到上述最终形态的七个动作最终实现均已落地移除Scope枚举与scope参数ctx.get_state/set_state回归请求作用域行为——当前context.py中的set_state/get_state保持请求级语义移除SessionCodec/密封—— ID 是裸uuid4sessions.pySession对象的异步访问器与单字典键方案——已实现sessions.pysession: UserSession注入主体为键、无认证报错接入与Context相同的参数检测路径——已实现_CurrentSession依赖session_id: SessionId标记类型schema 中为字符串、自动填充描述、独立await get_session(id)解析器并校验 IDtask worker 可用——已实现SessionProvider(Provider)提供create_session/end_session通过add_provider显式注册不强制存在——已实现测试重写覆盖两种模式、主体隔离、无认证行为与end_session——已落地于 test_sessions.py 与 test_session_provider.py。实践要点速查一个工具只选一种模式刻意避免ID 可选则默认的参数——那会静默把忘传 ID 的调用路由进共享桶每用户一个桶用session: UserSession注入、不在 schema、要求认证多会话用session_id: SessionId必填参数、Agent 提供、create-then-pass注册SessionProvider()以提供create_session/end_session生命周期工具漏注册不会报错但工具第一次真正调用时就会失败因为 ID 无法铸造、get_session必然拒绝无认证时ID 是 bearer 能力强制使用铸造 ID不要把会话当作客户端之间的墙——真正的租户边界只有认证主体TTL 归存储管需要过期策略就在传入的session_state_store上配置框架不设 TTL先创建、后校验get_session拒绝未铸造或他人主体的 ID抛InvalidSession错误信息刻意不泄露具体原因。# 一个完整的两种模式合体示例可运行骨架 from fastmcp import FastMCP from fastmcp.server.sessions import SessionProvider, SessionId, UserSession from fastmcp.server.dependencies import get_session mcp FastMCP(shop, session_state_storemy_store) # my_store 为 AsyncKeyValue mcp.add_provider(SessionProvider()) # 为 session_id 模式提供生命周期工具 mcp.tool async def remember(fact: str, session: UserSession) - str: 每用户桶注入式Agent 无需传参要求认证 await session.set(fact, fact) return noted mcp.tool async def add_to_cart(item: str, session_id: SessionId) - str: 多会话桶Agent 先 create_session 拿 ID再每次传入 session await get_session(session_id) cart await session.get(cart, default[]) cart.append(item) await session.set(cart, cart) return f{len(cart)} items上述骨架对应仓库中的端到端测试结构test_session_provider.py 的build_injected_server与build_id_server即两种模式的完整服务器形态可在你的项目中直接对照落地。【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
