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.py（HTTP 客户端）、client_stdio.py（进程内 STDIO 客户端）。

HTTP transport（推荐用于理解会话隔离）

分别打开两个终端：

# 终端 1：启动服务端 uv run python examples/persistent_state/server.py # 终端 2：运行客户端 uv run python examples/persistent_state/client.py

STDIO transport（进程内运行）

uv run python examples/persistent_state/client_stdio.py

STDIO 场景下客户端直接在进程内以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 f"Stored '{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 f"Key '{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)读取，键不存在时返回None；
ctx.session_id暴露当前会话标识符，便于观察"同一会话/不同会话"；
最后server.run(transport="streamable-http")以 HTTP 方式启动。

四、客户端脚本：三种场景逐行验证隔离性

client.py 通过StreamableHttpTransport(url="http://127.0.0.1:8000/mcp")与Client上下文管理器建立三个相互独立的 HTTP 连接，完整对应 README 中Example output的三种场景。

场景一：Alice 第一次连接，写入并读回

transport1 = StreamableHttpTransport(url=URL) async with Client(transport=transport1) 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 在自己的会话中写入user=Alice、secret=alice-password，随后两次读取均能命中。

场景二：Bob 连接，状态完全隔离

transport2 = StreamableHttpTransport(url=URL) async with Client(transport=transport2) 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"}) # Bob

Bob 的会话中读取 Alice 写入的键全部返回 "not found"，写入自己的user=Bob后可以读到——同一个键user在不同会话中互不影响。

场景三：Alice 重连，得到全新会话

transport3 = StreamableHttpTransport(url=URL) async with Client(transport=transport3) as alice_again: await alice_again.call_tool("get_value", {"key": "user"}) # not found

Alice 重新建立连接后，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 found

client_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, *, serializable=True) -> 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( key=prefixed_key, value=StateValue(value=value), ttl=self._STATE_TTL_SECONDS, )

默认serializable=True：值会被包装成StateValue（定义于 server.py 的模型，仅含value: Any字段）写入服务端配置的状态存储，TTL 默认86400秒（24 小时，见 context.py）；
值必须是 JSON 可序列化的（字典、列表、字符串、数字等）；若传入 HTTP client、数据库连接等不可序列化对象，会抛出TypeError，提示改用set_state(key, value, serializable=False)；
serializable=False的值存放在请求级字典_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(key=prefixed_key) return result.value if result is not None else None

读取顺序为：先检查请求级状态（serializable=False写入的遮蔽值），未命中再查询会话级状态存储；两者都不存在时返回None。配套的delete_state(key)会同时清理请求级与会话级两处数据。

4. 底层存储可替换

ctx.set_state/get_state最终操作的是self.fastmcp._state_store。默认情况下 FastMCP 使用进程内内存存储；若需要跨进程、多副本共享会话状态，可以在 FastMCP 构造函数传入session_state_store=AsyncKeyValue自定义存储后端。这一点与 docs/servers/sessions.mdx、docs/servers/storage-backends.mdx 中介绍的会话与存储抽象保持一致：会话状态的生命周期与 TTL 由底层存储决定。

六、会话生命周期 API：create_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 客户端）用serializable=False，避免误入状态存储造成序列化错误；
默认内存存储不跨进程：单进程演示（本示例）完全够用；需要多副本或持久化时配置自定义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），仅供参考