身份验证与访问控制

LangGraph 平台提供了一个灵活的身份验证和授权系统，可以与大多数身份验证方案集成。

核心概念

身份验证 vs 授权

尽管这两个术语经常被互换使用，但它们代表了不同的安全概念：

• 身份验证（“AuthN”）验证你是谁。这作为中间件处理每个请求。
• 授权（“AuthZ”）确定你能做什么。这根据资源的不同验证用户的权限和角色。

在LangGraph平台中，身份验证由你的@auth.authenticate处理器处理，而授权则由你的@auth.on处理器处理。

默认安全模型

LangGraph 平台提供了不同的安全默认设置：

LangGraph 云

• 默认使用 LangSmith API 密钥
• 需要在 x-api-key 头中提供有效的 API 密钥
• 可以通过自定义认证处理器进行自定义

自定义认证

自定义认证在 LangGraph 云的所有计划中都 受支持。

自托管

• 没有默认的身份验证
• 完全灵活，可以实现您的安全模型
• 您控制身份验证和授权的所有方面

自定义认证

自定义认证在 企业 自托管计划中受支持。 自托管轻量级计划不支持本机自定义认证。

系统架构

一个典型的认证设置包括三个主要组件：

1. 认证提供者（身份提供者/IdP）

   • 一个专门的服务，管理用户身份和凭证
   • 处理用户注册、登录、密码重置等操作
   • 在成功认证后颁发令牌（JWT、会话令牌等）
   • 示例：Auth0、Supabase Auth、Okta或自己的认证服务器

2. LangGraph 后端（资源服务器）

   • 包含业务逻辑和受保护资源的LangGraph应用程序
   • 验证来自认证提供者的令牌
   • 根据用户身份和权限实施访问控制
   • 不直接存储用户凭证

3. 客户端应用（前端）

   • 网页应用、移动应用或API客户端
   • 收集时间敏感的用户凭证并发送给认证提供者
   • 从认证提供者接收令牌
   • 将这些令牌包含在请求LangGraph后端时使用

这些组件之间的典型交互如下所示：

sequenceDiagram
    participant Client as Client App
    participant Auth as Auth Provider
    participant LG as LangGraph Backend

    Client->>Auth: 1. 登录（用户名/密码）
    Auth-->>Client: 2. 返回令牌
    Client->>LG: 3. 使用令牌请求
    Note over LG: 4. 验证令牌（@auth.authenticate）
    LG-->>Auth:  5. 获取用户信息
    Auth-->>LG: 6. 确认有效性
    Note over LG: 7. 应用访问控制（@auth.on.*）
    LG-->>Client: 8. 返回资源

您的@auth.authenticate处理器在LangGraph中处理步骤4-6，而您的@auth.on处理器实现步骤7。

认证

在LangGraph中，认证作为中间件处理每个请求。您的@auth.authenticate处理器会接收请求信息，并应执行以下操作：

1. 验证凭据
2. 如果有效，则返回包含用户身份和用户信息的user info
3. 如果无效，则抛出HTTP异常或AssertionError

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    # 验证凭据（例如，API密钥、JWT令牌）
    api_key = headers.get("x-api-key")
    if not api_key or not is_valid_key(api_key):
        raise Auth.exceptions.HTTPException(
            status_code=401,
            detail="Invalid API key"
        )

    # 返回用户信息——仅需唯一标识符和已验证身份
    # 添加授权所需的任何其他字段
    return {
        "identity": "user-123",        # 必须：唯一的用户标识符
        "is_authenticated": True,      # 可选：默认为True
        "permissions": ["read", "write"] # 可选：基于权限的授权
        # 您可以添加更多自定义字段以实现其他认证模式
        "role": "admin",
        "org_id": "org-456"

    }

返回的用户信息可用于：

• 通过ctx.user传递给您的授权处理器
• 在应用程序中通过config["configuration"]["langgraph_auth_user"]

支持的参数

@auth.authenticate处理器可以接受以下任意参数名称：

• request (Request): 原始ASGI请求对象
• body (dict): 解析后的请求正文
• path (str): 请求路径，例如，“/threads/abcd-1234-abcd-1234/runs/abcd-1234-abcd-1234/stream”
• method (str): HTTP方法，例如，“GET”
• path_params (dict[str, str]): URL路径参数，例如，{"thread_id": "abcd-1234-abcd-1234", "run_id": "abcd-1234-abcd-1234"}
• query_params (dict[str, str]): URL查询参数，例如，{"stream": "true"}
• headers (dict[bytes, bytes]): 请求头
• authorization (str | None): 授权头值（例如，“Bearer ”）

在许多教程中，我们将只显示“authorization”参数以简洁起见，但您可以选择接受更多信息，以便实现自定义认证方案。

授权

在认证之后，LangGraph 调用您的 @auth.on 处理程序来控制对特定资源（如线程、助手、计划任务）的访问。这些处理程序可以：

1. 在创建资源时直接修改 value["metadata"] 字典以添加元数据。有关每个操作支持的数据类型列表，请参阅支持的操作表。
2. 在搜索/列表或读取操作期间通过返回一个过滤字典按元数据筛选资源。
3. 如果拒绝访问，则引发一个 HTTP 异常。

如果您只想实现简单的用户范围访问控制，您可以为所有资源和操作使用单个 @auth.on 处理程序。如果您希望根据资源和操作有不同的控制，则可以使用资源特定处理程序。请参阅支持的资源部分以获取支持访问控制的完整资源列表。

@auth.on
async def add_owner(
    ctx: Auth.types.AuthContext,
    value: dict  # 发送到此访问方法的有效负载
) -> dict:  # 返回一个限制访问资源的过滤字典
    """授权对线程、运行、计划任务和助手的所有访问。

    此处理程序执行以下两项操作：
        - 将值添加到资源元数据中（以便与资源一起持久化，以便稍后可以对其进行筛选）
        - 返回一个过滤器（以限制对现有资源的访问）

    参数:
        ctx: 包含用户信息、权限、路径的认证上下文，
        value: 发送到端点的请求有效负载。对于创建操作，这包含资源参数。对于读取操作，这包含要访问的资源。

    返回:
        一个过滤字典，LangGraph 使用它来限制对资源的访问。有关支持的操作符，请参阅[过滤操作](#过滤操作)。
    """
    # 创建过滤器以仅限制对该用户的资源的访问
    filters = {"owner": ctx.user.identity}

    # 获取或创建有效负载中的元数据字典
    # 这是我们存储有关资源的持久信息的地方
    metadata = value.setdefault("metadata", {})

    # 将拥有者添加到元数据中 - 如果这是创建或更新操作，
    # 则此信息将与资源一起保存
    # 因此我们可以在稍后的读取操作中按拥有者筛选
    metadata.update(filters)

    # 返回过滤器以限制访问
    # 这些过滤器应用于所有操作（创建、读取、更新、搜索等）
    # 以确保用户只能访问自己的资源
    return filters

资源特定处理程序

您可以通过将资源和操作名称与 @auth.on 装饰器链接起来，为特定资源和操作注册处理程序。当发出请求时，最具体的匹配该资源和操作的处理程序将被调用。以下是注册特定资源和操作处理程序的一个示例。对于以下设置：

1. 认证用户能够创建线程、读取线程、在线程上创建运行
2. 只有具有“assistants:create”权限的用户才能创建新的助手
3. 所有其他端点（例如删除助手、计划任务、存储）对所有用户均禁用。

支持的处理程序

对于支持的资源和操作的完整列表，请参阅下方的支持的资源部分。

# 通用/全局处理程序捕获未由更具体处理程序处理的调用
@auth.on
async def reject_unhandled_requests(ctx: Auth.types.AuthContext, value: Any) -> False:
    print(f"Request to {ctx.path} by {ctx.user.identity}")
    raise Auth.exceptions.HTTPException(
        status_code=403,
        detail="Forbidden"
    )

# 匹配“thread”资源及其所有操作 - 创建、读取、更新、删除、搜索
# 由于这是比通用 @auth.on 处理程序更具体的处理程序，因此它将在“threads”资源的所有操作上优先于通用处理程序
@auth.on.threads
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    if "write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="User lacks the required permissions."
        )
    # 设置正在创建的线程的元数据
    # 确保资源包含一个“owner”字段
    # 然后每当用户尝试访问此线程或线程内的运行时，
    # 我们可以根据拥有者进行筛选
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

# 线程创建。这将仅匹配线程创建操作
# 由于这是比通用 @auth.on 处理程序和 @auth.on.threads 处理程序更具体的处理程序，
# 它将优先处理“threads”资源上的任何“create”操作
@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    # 设置正在创建的线程的元数据
    # 确保资源包含一个“owner”字段
    # 然后每当用户尝试访问此线程或线程内的运行时，
    # 我们可以根据拥有者进行筛选
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

# 读取线程。由于这也是比通用 @auth.on 处理程序和 @auth.on.threads 处理程序更具体的处理程序，
# 它将优先处理“threads”资源上的任何“read”操作
@auth.on.threads.read
async def on_thread_read(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.read.value
):
    # 由于我们在读取（而不是创建）线程，
    # 我们不需要设置元数据。我们只需要
    # 返回一个过滤器以确保用户只能看到自己的线程
    return {"owner": ctx.user.identity}

# 运行创建、流式传输、更新等
# 这优先于通用 @auth.on 处理程序和 @auth.on.threads 处理程序
@auth.on.threads.create_run
async def on_run_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create_run.value
):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    # 继承线程的访问控制
    return {"owner": ctx.user.identity}

# 助手创建
@auth.on.assistants.create
async def on_assistant_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.assistants.create.value
):
    if "assistants:create" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="User lacks the required permissions."
        )

请注意，我们在上述示例中混合了全局和资源特定处理程序。由于每个请求都由最具体的处理程序处理，因此创建 thread 的请求将匹配 on_thread_create 处理程序但不会匹配 reject_unhandled_requests 处理程序。然而，更新 thread 的请求将由全局处理程序处理，因为我们没有为此资源和操作提供更具体的处理程序。

过滤操作

授权处理程序可以返回 None、一个布尔值或一个过滤字典。 - None 和 True 表示“允许访问所有子资源” - False 表示“拒绝访问所有子资源（引发 403 异常）” - 元数据过滤字典将限制对资源的访问

过滤字典是一个键与资源元数据相匹配的字典。它支持三种操作符：

• 默认值是精确匹配的简写，或"$eq"，如下所示。例如，{"owner": user_id} 将只包括元数据中包含 {"owner": user_id} 的资源。
• $eq：精确匹配（例如，{"owner": {"$eq": user_id}}）——这相当于上面的简写形式，{"owner": user_id}
• $contains：列表成员资格（例如，{"allowed_users": {"$contains": user_id}}）这里的值必须是列表中的元素。存储资源中的元数据必须是列表/容器类型。

具有多个键的字典将使用逻辑 AND 进行处理。例如，{"owner": org_id, "allowed_users": {"$contains": user_id}} 将只匹配元数据中“owner”为 org_id 且“allowed_users”列表包含 user_id 的资源。 更多详细信息请参阅此处的参考文档这里。

常见访问模式

这里是一些典型的授权模式：

单一所有者资源

这种常见的模式允许您将所有线程、助手、定时任务和运行都限定为单个用户。它适用于普通的单用户应用场景，如常规聊天机器人应用。

@auth.on
async def owner_only(ctx: Auth.types.AuthContext, value: dict):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

基于权限的访问控制

这种模式允许您根据**权限**来控制访问。如果希望某些角色具有更广泛的或更严格的资源访问权限，则此模式非常有用。

# 在您的认证处理器中：
@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    ...
    return {
        "identity": "user-123",
        "is_authenticated": True,
        "permissions": ["threads:write", "threads:read"]  # 在认证时定义权限
    }

def _default(ctx: Auth.types.AuthContext, value: dict):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

@auth.on.threads.create
async def create_thread(ctx: Auth.types.AuthContext, value: dict):
    if "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Unauthorized"
        )
    return _default(ctx, value)


@auth.on.threads.read
async def rbac_create(ctx: Auth.types.AuthContext, value: dict):
    if "threads:read" not in ctx.permissions and "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="Unauthorized"
        )
    return _default(ctx, value)

支持的资源

LangGraph 提供三种级别的授权处理器，从最通用到最具体：

1. 全局处理器 (@auth.on)：匹配所有资源和操作
2. 资源处理器（例如，@auth.on.threads，@auth.on.assistants，@auth.on.crons）：匹配特定资源的所有操作
3. 操作处理器（例如，@auth.on.threads.create，@auth.on.threads.read）：匹配特定资源上的特定操作

最具体的匹配处理器将被使用。例如，在线程创建时，@auth.on.threads.create 的优先级高于 @auth.on.threads。 如果注册了更具体的处理器，则该资源和操作不会调用更通用的处理器。

类型安全性

每个处理器都为其 value 参数提供了类型提示，位于 Auth.types.on.<资源>.<操作>.value。例如：

@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.threads.create.value  # 线程创建的具体类型
):
    ...

@auth.on.threads
async def on_threads(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.threads.value  # 所有线程操作的联合类型
):
    ...

@auth.on
async def on_all(
    ctx: Auth.types.AuthContext,
    value: dict  # 所有可能操作的联合类型
):
    ...

更具体的处理器提供更好的类型提示，因为它们处理较少的操作类型。

支持的操作和类型

以下是所有支持的操作处理器：

资源	处理器	描述	值类型
线程	@auth.on.threads.create	创建线程	ThreadsCreate
	@auth.on.threads.read	获取线程	ThreadsRead
	@auth.on.threads.update	更新线程	ThreadsUpdate
	@auth.on.threads.delete	删除线程	ThreadsDelete
	@auth.on.threads.search	列出线程	ThreadsSearch
	@auth.on.threads.create_run	创建或更新运行	RunsCreate
助手	@auth.on.assistants.create	创建助手	AssistantsCreate
	@auth.on.assistants.read	获取助手	AssistantsRead
	@auth.on.assistants.update	更新助手	AssistantsUpdate
	@auth.on.assistants.delete	删除助手	AssistantsDelete
	@auth.on.assistants.search	列出助手	AssistantsSearch
计划任务	@auth.on.crons.create	创建计划任务	CronsCreate
	@auth.on.crons.read	获取计划任务	CronsRead
	@auth.on.crons.update	更新计划任务	CronsUpdate
	@auth.on.crons.delete	删除计划任务	CronsDelete
	@auth.on.crons.search	列出计划任务	CronsSearch

关于运行

运行的访问控制范围限定在其父线程内。这意味着权限通常从线程继承，反映了数据模型的对话性质。除了创建之外，所有运行操作（读取、列出）都由线程的处理器控制。 特别地，有一个专门用于创建新运行的处理器，因为它有更多的参数可以在处理器中查看。

下一步

有关实现细节，请参阅以下内容：

• 查看有关设置身份验证的入门教程
• 参考实现自定义身份验证处理程序的指南

Comments