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），僅供參考