Files
wehub-resource-sync 60e0ffc959
Upgrade checks / Notify on failure (push) Has been cancelled
Upgrade checks / Close issue on success (push) Has been cancelled
Schema Crash Test / Real-world schema crash test (232K schemas) (push) Has been cancelled
Run static analysis / static_analysis (push) Has been cancelled
Tests / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Tests / Tests with lowest-direct dependencies (push) Has been cancelled
Tests / MCP conformance tests (push) Has been cancelled
Tests / Integration tests (push) Has been cancelled
Tests / Package install smoke (push) Has been cancelled
Upgrade checks / Static analysis (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Upgrade checks / Integration tests (push) Has been cancelled
Update MCPServerConfig Schema / update-config-schema (push) Has been cancelled
Update SDK Documentation / update-sdk-docs (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:39:59 +08:00

43 lines
1.1 KiB
Python

"""Example: Persistent session-scoped state.
This demonstrates using Context.get_state() and set_state() to store
data that persists across tool calls within the same MCP session.
Run with:
uv run python examples/persistent_state/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,
}
if __name__ == "__main__":
server.run(transport="streamable-http")