按用户隔离存储数据

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "auth": {
    "path": "./src/security/auth.py:auth"
  }
}

from langgraph_sdk import Auth

auth = Auth()

# ... 你的 @auth.authenticate 处理器 ...

@auth.on.store
async def authorize_store(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.store.value,
):
    """要求用户身份作为命名空间的第一段。"""
    namespace = tuple(value["namespace"])
    if not namespace or namespace[0] != ctx.user.identity:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Not authorized to access this namespace.",
        )

from langgraph_sdk import Auth

auth = Auth()

# ... 你的 @auth.authenticate 处理器 ...

@auth.on.store
async def scope_store(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.store.value,
):
    """通过重写命名空间来按用户隔离存储数据。"""
    namespace = tuple(value["namespace"]) if value.get("namespace") else ()
    if not namespace or namespace[0] != ctx.user.identity:
        namespace = (ctx.user.identity, *namespace)
    value["namespace"] = namespace

from langgraph.store.base import BaseStore

async def save_preference(state: State, *, store: BaseStore):
    await store.aput(
        ("memories", "preferences"),  # 逻辑命名空间
        "settings",
        {"theme": "dark"},
    )

async def load_preference(state: State, *, store: BaseStore):
    item = await store.aget(
        ("memories", "preferences"),
        "settings",
    )
    return item.value if item else {}

from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore

async def save_preference(state: State, *, store: BaseStore, config: RunnableConfig):
    user_id = config["configurable"]["langgraph_auth_user_id"]
    await store.aput(
        (user_id, "memories", "preferences"),
        "settings",
        {"theme": "dark"},
    )

from deepagents.backends import StoreBackend

StoreBackend(
    namespace=lambda rt: ("memories",),  # 认证添加用户身份
)

import asyncio

from langgraph_sdk import get_client


async def main():
    alice = get_client(
        url="http://localhost:60058",
        headers={"Authorization": "Bearer user1-token"},
    )
    bob = get_client(
        url="http://localhost:60058",
        headers={"Authorization": "Bearer user2-token"},
    )

    # Alice 写入一个存储项
    await alice.store.put_item(
        ["memories"],
        key="note",
        value={"text": "Alice private note"},
    )

    # Bob 无法读取 Alice 的项
    bob_item = await bob.store.get_item(["memories"], key="note")
    if bob_item is None or bob_item["value"]["text"] != "Alice private note":
        print("✅ Bob correctly denied access to Alice's item")
    else:
        print("❌ Bob should not see Alice's store item")

    # Bob 在相同的逻辑命名空间中写入自己的项
    await bob.store.put_item(
        ["memories"],
        key="note",
        value={"text": "Bob private note"},
    )

    alice_item = await alice.store.get_item(["memories"], key="note")
    bob_item = await bob.store.get_item(["memories"], key="note")
    assert alice_item["value"]["text"] == "Alice private note"
    assert bob_item["value"]["text"] == "Bob private note"
    print("✅ Each user sees only their own store data")


if __name__ == "__main__":
    asyncio.run(main())