FastMCP 持久化会话状态实战用 Context.get_state / set_state 构建会话级跨工具调用存储【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp输出文章FastMCP 持久化会话状态实战用 Context.get_state / set_state 构建会话级跨工具调用存储FastMCP 提供的会话作用域状态session-scoped state允许服务端在一个 MCP 会话内跨多次工具调用持久化键值数据一次工具调用写入的值后续调用可以读取不同客户端会话之间的状态完全隔离客户端断开重连后得到的是一个全新会话旧状态不复存在。本文以 examples/persistent_state 示例为骨架从运行方式、源码实现到多客户端隔离验证完整讲解如何在 FastMCP 服务端使用Context.get_state()/set_state()实现这一能力。一、示例概览会话状态要解决什么问题无状态 HTTP 请求天然无法记住上一次调用发生了什么。在真实业务中我们常常需要在工具 A 中登录或设置上下文在工具 B 中读取例如设置用户身份后查询其专属数据多个客户端同时连接同一个服务端各自维护互不干扰的状态例如 Alice 和 Bob 同时在线键user分别对应不同值会话结束后自动释放状态断线重连即新会话旧数据不再可见。examples/persistent_state演示的正是这三点README 将其概括为一次工具调用中设置的状态可在后续调用中读取不同客户端状态相互隔离相同键、不同值重新连接创建全新会话状态为空白。二、快速运行HTTP 与 STDIO 两种传输方式示例目录包含三个文件server.py服务端、client.pyHTTP 客户端、client_stdio.py进程内 STDIO 客户端。HTTP transport推荐用于理解会话隔离分别打开两个终端# 终端 1启动服务端 uv run python examples/persistent_state/server.py # 终端 2运行客户端 uv run python examples/persistent_state/client.pySTDIO transport进程内运行uv run python examples/persistent_state/client_stdio.pySTDIO 场景下客户端直接在进程内以Client(server)方式连接服务端对象无需启动独立服务适合快速验证和测试。三、服务端实现三行核心 API 讲透会话状态server.py 只做三件事对应三个工具from fastmcp import FastMCP from fastmcp.server.context import Context server FastMCP(StateExample) server.tool async def set_value(key: str, value: str, ctx: Context) - str: Store a value in session state. await ctx.set_state(key, value) return fStored {key} {value} server.tool async def get_value(key: str, ctx: Context) - str: Retrieve a value from session state. value await ctx.get_state(key) if value is None: return fKey {key} not found return f{key} {value} server.tool async def list_session_info(ctx: Context) - dict[str, str | None]: Get information about the current session. return { session_id: ctx.session_id, transport: ctx.transport, }关键点ctx: Context参数由 FastMCP 依赖注入不出现在工具的 JSON Schema 输入参数中await ctx.set_state(key, value)写入会话状态await ctx.get_state(key)读取键不存在时返回Nonectx.session_id暴露当前会话标识符便于观察同一会话/不同会话最后server.run(transportstreamable-http)以 HTTP 方式启动。四、客户端脚本三种场景逐行验证隔离性client.py 通过StreamableHttpTransport(urlhttp://127.0.0.1:8000/mcp)与Client上下文管理器建立三个相互独立的 HTTP 连接完整对应 README 中Example output的三种场景。场景一Alice 第一次连接写入并读回transport1 StreamableHttpTransport(urlURL) async with Client(transporttransport1) as alice: result await alice.call_tool(list_session_info, {}) console.print(f session [cyan]{result.data[session_id][:8]}[/cyan]) await alice.call_tool(set_value, {key: user, value: Alice}) await alice.call_tool(set_value, {key: secret, value: alice-password}) await alice.call_tool(get_value, {key: user}) await alice.call_tool(get_value, {key: secret})Alice 在自己的会话中写入userAlice、secretalice-password随后两次读取均能命中。场景二Bob 连接状态完全隔离transport2 StreamableHttpTransport(urlURL) async with Client(transporttransport2) as bob: await bob.call_tool(get_value, {key: user}) # not found await bob.call_tool(get_value, {key: secret}) # not found await bob.call_tool(set_value, {key: user, value: Bob}) await bob.call_tool(get_value, {key: user}) # BobBob 的会话中读取 Alice 写入的键全部返回 not found写入自己的userBob后可以读到——同一个键user在不同会话中互不影响。场景三Alice 重连得到全新会话transport3 StreamableHttpTransport(urlURL) async with Client(transporttransport3) as alice_again: await alice_again.call_tool(get_value, {key: user}) # not foundAlice 重新建立连接后session_id已更换之前的user值不可见。这是默认内存存储的预期行为会话结束即状态失效。预期输出运行后结合 rich 输出大致如下Each line below is a separate tool call Alice connects session a9f6eaa3 set user Alice set secret alice-password get user → Alice get secret → alice-password Bob connects (different session) session 0c3bffc5 get user → not found get secret → not found set user Bob get user → Bob Alice reconnects (new session) session e39640e3 get user → not foundclient_stdio.py 以async with Client(server) as alice:直接传入服务端对象执行相同的三段验证逻辑结果一致。五、源码级原理set_state / get_state 背后发生了什么在 fastmcp_slim/fastmcp/server/context.py 中会话状态实现的核心是_make_state_key与两个方法1. 键自动加会话前缀def _make_state_key(self, key: str) - str: Create session-prefixed key for state storage. return f{self.session_id}:{key}写入时set_state(key, value)会把用户传入的键改写成{session_id}:{key}再落库见 context.py。这就是不同客户端相同键互不干扰的根本机制——隔离不是靠额外过滤而是靠键空间的天然前缀划分。2. set_state可序列化值持久化到会话级状态存储async def set_state(self, key, value, *, serializableTrue) - None: prefixed_key self._make_state_key(key) if not serializable: self._request_state[prefixed_key] value return self._request_state.pop(prefixed_key, None) await self.fastmcp._state_store.put( keyprefixed_key, valueStateValue(valuevalue), ttlself._STATE_TTL_SECONDS, )默认serializableTrue值会被包装成StateValue定义于 server.py 的模型仅含value: Any字段写入服务端配置的状态存储TTL 默认86400秒24 小时见 context.py值必须是 JSON 可序列化的字典、列表、字符串、数字等若传入 HTTP client、数据库连接等不可序列化对象会抛出TypeError提示改用set_state(key, value, serializableFalse)serializableFalse的值存放在请求级字典_request_state中只在当前 MCP 请求一次工具调用/资源读取/提示词渲染内有效不会跨请求存活。3. get_state先查请求级再查会话级async def get_state(self, key) - Any: prefixed_key self._make_state_key(key) if prefixed_key in self._request_state: return self._request_state[prefixed_key] result await self.fastmcp._state_store.get(keyprefixed_key) return result.value if result is not None else None读取顺序为先检查请求级状态serializableFalse写入的遮蔽值未命中再查询会话级状态存储两者都不存在时返回None。配套的delete_state(key)会同时清理请求级与会话级两处数据。4. 底层存储可替换ctx.set_state/get_state最终操作的是self.fastmcp._state_store。默认情况下 FastMCP 使用进程内内存存储若需要跨进程、多副本共享会话状态可以在 FastMCP 构造函数 传入session_state_storeAsyncKeyValue自定义存储后端。这一点与 docs/servers/sessions.mdx、docs/servers/storage-backends.mdx 中介绍的会话与存储抽象保持一致会话状态的生命周期与 TTL 由底层存储决定。六、会话生命周期 APIcreate_session / end_session 与状态清理除了Context上的状态方法fastmcp_slim/fastmcp/server/sessions.py 还提供显式的会话生命周期管理create_session()铸造一个不可猜测的uuid4会话 ID 并记录该会话end_session(session_id)使会话失效并删除其全部状态。这两个工具由SessionProvider提供可通过mcp.add_provider(SessionProvider())注册见 sessions.py。从源码注释可以确认end_session会校验会话 ID未知或外部 ID 一律拒绝再删除会话对应键使该 ID 不再可解析。需要主动销毁会话的场景如登出、超时清理应优先使用这套 API而不是依赖 TTL 自然过期。七、会话状态的适用边界与最佳实践会话级 vs 请求级需要跨多次工具调用共享的数据用户偏好、认证令牌、对话上下文用默认的set_state(key, value)仅本次请求内有效的一次性对象数据库连接、HTTP 客户端用serializableFalse避免误入状态存储造成序列化错误默认内存存储不跨进程单进程演示本示例完全够用需要多副本或持久化时配置自定义session_state_store重连即新会话默认行为下客户端断开后会话状态随之释放因此不要把必须长期留存的数据完全寄托于会话状态必要时改用外部持久化键空间隔离session_id前缀保证了多租户互不干扰但也意味着会话级全局键无法被其他会话读取设计跨会话共享数据时需另寻方案如 docs/servers/tasks.mdx 中介绍的独立任务/存储机制。八、小结examples/persistent_state是理解 FastMCP 会话作用域状态的最小闭环三行服务端工具定义 三种客户端场景验证即可讲透set_state/get_state的写入、读取与隔离语义。配合 context.py 的源码可以看到隔离靠会话前缀键、持久化靠可替换的状态存储、生命周期可被SessionProvider显式控制。对于任何需要会话内记忆的 MCP 服务多轮对话工具、分步流程、用户级上下文注入这套模式都值得作为首选方案。【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastMCP 持久化会话状态实战：用 Context.get_state / set_state 构建会话级跨工具调用存储
FastMCP 持久化会话状态实战用 Context.get_state / set_state 构建会话级跨工具调用存储【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp输出文章FastMCP 持久化会话状态实战用 Context.get_state / set_state 构建会话级跨工具调用存储FastMCP 提供的会话作用域状态session-scoped state允许服务端在一个 MCP 会话内跨多次工具调用持久化键值数据一次工具调用写入的值后续调用可以读取不同客户端会话之间的状态完全隔离客户端断开重连后得到的是一个全新会话旧状态不复存在。本文以 examples/persistent_state 示例为骨架从运行方式、源码实现到多客户端隔离验证完整讲解如何在 FastMCP 服务端使用Context.get_state()/set_state()实现这一能力。一、示例概览会话状态要解决什么问题无状态 HTTP 请求天然无法记住上一次调用发生了什么。在真实业务中我们常常需要在工具 A 中登录或设置上下文在工具 B 中读取例如设置用户身份后查询其专属数据多个客户端同时连接同一个服务端各自维护互不干扰的状态例如 Alice 和 Bob 同时在线键user分别对应不同值会话结束后自动释放状态断线重连即新会话旧数据不再可见。examples/persistent_state演示的正是这三点README 将其概括为一次工具调用中设置的状态可在后续调用中读取不同客户端状态相互隔离相同键、不同值重新连接创建全新会话状态为空白。二、快速运行HTTP 与 STDIO 两种传输方式示例目录包含三个文件server.py服务端、client.pyHTTP 客户端、client_stdio.py进程内 STDIO 客户端。HTTP transport推荐用于理解会话隔离分别打开两个终端# 终端 1启动服务端 uv run python examples/persistent_state/server.py # 终端 2运行客户端 uv run python examples/persistent_state/client.pySTDIO transport进程内运行uv run python examples/persistent_state/client_stdio.pySTDIO 场景下客户端直接在进程内以Client(server)方式连接服务端对象无需启动独立服务适合快速验证和测试。三、服务端实现三行核心 API 讲透会话状态server.py 只做三件事对应三个工具from fastmcp import FastMCP from fastmcp.server.context import Context server FastMCP(StateExample) server.tool async def set_value(key: str, value: str, ctx: Context) - str: Store a value in session state. await ctx.set_state(key, value) return fStored {key} {value} server.tool async def get_value(key: str, ctx: Context) - str: Retrieve a value from session state. value await ctx.get_state(key) if value is None: return fKey {key} not found return f{key} {value} server.tool async def list_session_info(ctx: Context) - dict[str, str | None]: Get information about the current session. return { session_id: ctx.session_id, transport: ctx.transport, }关键点ctx: Context参数由 FastMCP 依赖注入不出现在工具的 JSON Schema 输入参数中await ctx.set_state(key, value)写入会话状态await ctx.get_state(key)读取键不存在时返回Nonectx.session_id暴露当前会话标识符便于观察同一会话/不同会话最后server.run(transportstreamable-http)以 HTTP 方式启动。四、客户端脚本三种场景逐行验证隔离性client.py 通过StreamableHttpTransport(urlhttp://127.0.0.1:8000/mcp)与Client上下文管理器建立三个相互独立的 HTTP 连接完整对应 README 中Example output的三种场景。场景一Alice 第一次连接写入并读回transport1 StreamableHttpTransport(urlURL) async with Client(transporttransport1) as alice: result await alice.call_tool(list_session_info, {}) console.print(f session [cyan]{result.data[session_id][:8]}[/cyan]) await alice.call_tool(set_value, {key: user, value: Alice}) await alice.call_tool(set_value, {key: secret, value: alice-password}) await alice.call_tool(get_value, {key: user}) await alice.call_tool(get_value, {key: secret})Alice 在自己的会话中写入userAlice、secretalice-password随后两次读取均能命中。场景二Bob 连接状态完全隔离transport2 StreamableHttpTransport(urlURL) async with Client(transporttransport2) as bob: await bob.call_tool(get_value, {key: user}) # not found await bob.call_tool(get_value, {key: secret}) # not found await bob.call_tool(set_value, {key: user, value: Bob}) await bob.call_tool(get_value, {key: user}) # BobBob 的会话中读取 Alice 写入的键全部返回 not found写入自己的userBob后可以读到——同一个键user在不同会话中互不影响。场景三Alice 重连得到全新会话transport3 StreamableHttpTransport(urlURL) async with Client(transporttransport3) as alice_again: await alice_again.call_tool(get_value, {key: user}) # not foundAlice 重新建立连接后session_id已更换之前的user值不可见。这是默认内存存储的预期行为会话结束即状态失效。预期输出运行后结合 rich 输出大致如下Each line below is a separate tool call Alice connects session a9f6eaa3 set user Alice set secret alice-password get user → Alice get secret → alice-password Bob connects (different session) session 0c3bffc5 get user → not found get secret → not found set user Bob get user → Bob Alice reconnects (new session) session e39640e3 get user → not foundclient_stdio.py 以async with Client(server) as alice:直接传入服务端对象执行相同的三段验证逻辑结果一致。五、源码级原理set_state / get_state 背后发生了什么在 fastmcp_slim/fastmcp/server/context.py 中会话状态实现的核心是_make_state_key与两个方法1. 键自动加会话前缀def _make_state_key(self, key: str) - str: Create session-prefixed key for state storage. return f{self.session_id}:{key}写入时set_state(key, value)会把用户传入的键改写成{session_id}:{key}再落库见 context.py。这就是不同客户端相同键互不干扰的根本机制——隔离不是靠额外过滤而是靠键空间的天然前缀划分。2. set_state可序列化值持久化到会话级状态存储async def set_state(self, key, value, *, serializableTrue) - None: prefixed_key self._make_state_key(key) if not serializable: self._request_state[prefixed_key] value return self._request_state.pop(prefixed_key, None) await self.fastmcp._state_store.put( keyprefixed_key, valueStateValue(valuevalue), ttlself._STATE_TTL_SECONDS, )默认serializableTrue值会被包装成StateValue定义于 server.py 的模型仅含value: Any字段写入服务端配置的状态存储TTL 默认86400秒24 小时见 context.py值必须是 JSON 可序列化的字典、列表、字符串、数字等若传入 HTTP client、数据库连接等不可序列化对象会抛出TypeError提示改用set_state(key, value, serializableFalse)serializableFalse的值存放在请求级字典_request_state中只在当前 MCP 请求一次工具调用/资源读取/提示词渲染内有效不会跨请求存活。3. get_state先查请求级再查会话级async def get_state(self, key) - Any: prefixed_key self._make_state_key(key) if prefixed_key in self._request_state: return self._request_state[prefixed_key] result await self.fastmcp._state_store.get(keyprefixed_key) return result.value if result is not None else None读取顺序为先检查请求级状态serializableFalse写入的遮蔽值未命中再查询会话级状态存储两者都不存在时返回None。配套的delete_state(key)会同时清理请求级与会话级两处数据。4. 底层存储可替换ctx.set_state/get_state最终操作的是self.fastmcp._state_store。默认情况下 FastMCP 使用进程内内存存储若需要跨进程、多副本共享会话状态可以在 FastMCP 构造函数 传入session_state_storeAsyncKeyValue自定义存储后端。这一点与 docs/servers/sessions.mdx、docs/servers/storage-backends.mdx 中介绍的会话与存储抽象保持一致会话状态的生命周期与 TTL 由底层存储决定。六、会话生命周期 APIcreate_session / end_session 与状态清理除了Context上的状态方法fastmcp_slim/fastmcp/server/sessions.py 还提供显式的会话生命周期管理create_session()铸造一个不可猜测的uuid4会话 ID 并记录该会话end_session(session_id)使会话失效并删除其全部状态。这两个工具由SessionProvider提供可通过mcp.add_provider(SessionProvider())注册见 sessions.py。从源码注释可以确认end_session会校验会话 ID未知或外部 ID 一律拒绝再删除会话对应键使该 ID 不再可解析。需要主动销毁会话的场景如登出、超时清理应优先使用这套 API而不是依赖 TTL 自然过期。七、会话状态的适用边界与最佳实践会话级 vs 请求级需要跨多次工具调用共享的数据用户偏好、认证令牌、对话上下文用默认的set_state(key, value)仅本次请求内有效的一次性对象数据库连接、HTTP 客户端用serializableFalse避免误入状态存储造成序列化错误默认内存存储不跨进程单进程演示本示例完全够用需要多副本或持久化时配置自定义session_state_store重连即新会话默认行为下客户端断开后会话状态随之释放因此不要把必须长期留存的数据完全寄托于会话状态必要时改用外部持久化键空间隔离session_id前缀保证了多租户互不干扰但也意味着会话级全局键无法被其他会话读取设计跨会话共享数据时需另寻方案如 docs/servers/tasks.mdx 中介绍的独立任务/存储机制。八、小结examples/persistent_state是理解 FastMCP 会话作用域状态的最小闭环三行服务端工具定义 三种客户端场景验证即可讲透set_state/get_state的写入、读取与隔离语义。配合 context.py 的源码可以看到隔离靠会话前缀键、持久化靠可替换的状态存储、生命周期可被SessionProvider显式控制。对于任何需要会话内记忆的 MCP 服务多轮对话工具、分步流程、用户级上下文注入这套模式都值得作为首选方案。【免费下载链接】fastmcp The fast, Pythonic way to build MCP servers and clients.项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考