身份验证与访问控制

LangSmith 提供灵活的身份验证和授权系统，可以与大多数身份验证方案集成。

核心概念

身份验证与授权

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

身份验证（“AuthN”）用于验证_你是谁_。它会作为中间件在每个请求中运行。
授权（“AuthZ”）用于确定_你能做什么_。它会根据每个资源验证用户的权限和角色。

在 LangSmith 中，身份验证由你的 @auth.authenticate 处理程序负责，授权则由你的 @auth.on 处理程序负责。

默认安全模型

LangSmith 提供不同的默认安全设置：

LangSmith

默认使用 LangSmith API 密钥
要求在 x-api-key 请求头中提供有效的 API 密钥
可以通过你的身份验证处理程序进行自定义

自定义身份验证 LangSmith 的所有套餐均支持自定义身份验证。

自托管

默认不启用身份验证
可以完全自由地实现自己的安全模型
身份验证和授权的所有方面都由你控制

系统架构

典型的身份验证设置包含三个主要组件：

身份验证提供商（身份提供商/IdP）
- 用于管理用户身份和凭据的专用服务
- 负责用户注册、登录、密码重置等操作
- 身份验证成功后签发令牌（JWT、会话令牌等）
- 示例：Auth0、Supabase Auth、Okta 或你自己的身份验证服务器
代理服务器（资源服务器）
- 你的代理或 LangGraph 应用，其中包含业务逻辑和受保护资源
- 使用身份验证提供商验证令牌
- 根据用户身份和权限强制执行访问控制
- 不直接存储用户凭据
客户端应用（前端）
- Web 应用、移动应用或 API 客户端
- 收集时效性较强的用户凭据，并将其发送给身份验证提供商
- 从身份验证提供商接收令牌
- 在向代理服务器发送的请求中包含这些令牌

以下是这些组件通常的交互方式：你在 LangGraph 中的 @auth.authenticate 处理程序负责处理第 4-6 步，而你的 @auth.on 处理程序负责实现第 7 步。

身份验证

LangGraph 中的身份验证会作为中间件在每个请求中运行。你的 @auth.authenticate 处理程序会接收请求信息，并应执行以下操作：

验证凭据
如果凭据有效，则返回包含用户身份和用户信息的用户信息
如果凭据无效，则引发 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(b"x-api-key")
    if not api_key or not is_valid_key(api_key):
        raise Auth.exceptions.HTTPException(
            status_code=401,
            detail="无效的 API 密钥"
        )

    # 返回用户信息——只需要提供 identity 和 is_authenticated
    # 添加授权所需的其他字段
    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 请求对象
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)：Authorization 请求头的值（例如 "Bearer <token>"）

在许多教程中，为了简洁起见，我们只展示 "authorization" 参数，但你也可以根据需要接收更多信息，以实现自定义身份验证方案。

代理身份验证

自定义身份验证支持委托访问。你在 @auth.authenticate 中返回的值会被添加到运行上下文中，使代理能够获得用户范围内的凭据，并代表用户访问资源。完成身份验证后，平台会创建一个特殊的配置对象，并通过可配置上下文将其传递给你的图及其所有节点。该对象包含当前用户的信息，包括你从 @auth.authenticate 处理程序返回的所有自定义字段。要让代理代表用户执行操作，请使用自定义身份验证中间件。这样，代理就可以代表用户与 MCP 服务器、外部数据库甚至其他代理等外部系统交互。更多信息请参阅使用自定义身份验证指南。

使用 MCP 进行代理身份验证

有关如何对代理访问 MCP 服务器进行身份验证的信息，请参阅 MCP 概念指南。

授权

完成身份验证后，LangGraph 会调用你的 @auth.on 处理程序，以控制对特定资源（例如线程、助手和 cron）的访问。这些处理程序可以：

直接修改 value["metadata"] 字典，为资源创建期间要保存的内容添加元数据。每种操作中 value 可接受的类型，请参阅支持的操作表。
在搜索/列表或读取操作期间，根据元数据筛选资源，方法是返回一个筛选字典。
在访问被拒绝时引发 HTTP 异常。

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

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

    此处理程序会执行两项操作：
        - 向资源元数据添加值（与资源一起持久化，以便稍后进行筛选）
        - 返回筛选条件（限制对现有资源的访问）

    参数：
        ctx：包含用户信息、权限、路径的身份验证上下文，以及
        value：发送到端点的请求负载。对于创建操作，
              其中包含资源参数；对于读取操作，其中包含
              正在访问的资源。

    返回：
        LangGraph 用于限制资源访问权限的筛选字典。
        支持的运算符请参阅[筛选操作](#filter-operations)。
    """
    # 创建筛选条件，将访问范围限制为当前用户的资源
    filters = {"owner": ctx.user.identity}

    # 获取或创建负载中的元数据字典
    # 这里用于存储资源的持久化信息
    metadata = value.setdefault("metadata", {})

    # 将所有者添加到元数据中——如果这是创建或更新操作，
    # 此信息将随资源一起保存
    # 这样稍后在读取操作中就可以据此进行筛选
    metadata.update(filters)

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

资源专属处理程序

你可以通过将资源名称和操作名称与 @auth.on 装饰器链式组合，为特定资源和操作注册处理程序。收到请求时，系统会调用与该资源和操作匹配的最具体处理程序。下面是为特定资源和操作注册处理程序的示例。对于以下设置：

经过身份验证的用户可以创建线程、读取线程以及在线程上创建运行
只有拥有 "assistants:create" 权限的用户才能创建新助手
所有其他端点（例如删除助手、cron、存储）均对所有用户禁用。

支持的处理程序 支持的资源和操作的完整列表，请参阅下面的支持的资源部分。

# 通用/全局处理程序会捕获未被更具体处理程序处理的调用
@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="禁止访问"
    )

# 匹配 "thread" 资源及其所有操作——创建、读取、更新、删除、搜索
# 由于它比通用的 @auth.on 处理程序**更具体**，因此会优先于后者
# 对 "threads" 资源的所有操作，该处理程序都会优先于通用处理程序
@auth.on.threads
async def on_thread(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    # 为正在创建的线程设置元数据
    # 确保资源包含 "owner" 字段
    # 之后每当用户尝试访问此线程或在线程中运行时，
    # 我们就可以根据 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
):
    # 如果用户没有写入权限，则拒绝请求
    if "write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="用户缺少所需权限。"
        )
    # 为正在创建的线程设置元数据
    # 确保资源包含 "owner" 字段
    # 之后每当用户尝试访问此线程或在线程中运行时，
    # 我们就可以根据 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="用户缺少所需权限。"
        )

请注意，上面的示例混用了全局处理程序和资源专属处理程序。由于每个请求都由最具体的处理程序处理，因此创建 thread 的请求会匹配 on_thread_create 处理程序，但不会匹配 reject_unhandled_requests 处理程序。不过，更新线程的请求会由全局处理程序处理，因为我们没有为该资源和操作注册更具体的处理程序。

筛选操作

授权处理程序可以返回 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}}）或列表包含匹配（例如 {"allowed_users": {"$contains": [user_id_1, user_id_2]}}）。这里的值必须分别是列表中的元素，或列表元素的子集。已存储资源中的元数据必须是列表/容器类型。

包含多个键的字典会使用逻辑 AND 筛选。例如，{"owner": org_id, "allowed_users": {"$contains": user_id}} 只会匹配元数据中 "owner" 为 org_id 且 "allowed_users" 列表包含 user_id 的资源。更多信息请参阅参考文档 Auth(Auth)。

常见访问模式

以下是一些典型的授权模式：

单一所有者资源

这种常见模式可以将所有线程、助手、cron 和运行限定到单个用户。它适用于普通的单用户场景，例如常规的聊天机器人应用。

@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="未授权"
        )
    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="未授权"
        )
    return _default(ctx, value)

支持的资源

LangGraph 提供三个级别的授权处理程序，从最通用到最具体依次为：

全局处理程序（@auth.on）：匹配所有资源和操作
资源处理程序（例如 @auth.on.threads、@auth.on.assistants、@auth.on.crons）：匹配特定资源的所有操作
操作处理程序（例如 @auth.on.threads.create、@auth.on.threads.read）：匹配特定资源上的特定操作

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

“类型安全” 每个处理程序的 value 参数都可以使用类型提示，路径为 Auth.types.on.<resource>.<action>.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`
Cron	`@auth.on.crons.create`	创建 Cron 作业	`CronsCreate`
	`@auth.on.crons.read`	获取 Cron 作业	`CronsRead`
	`@auth.on.crons.update`	更新 Cron 作业	`CronsUpdate`
	`@auth.on.crons.delete`	删除 Cron 作业	`CronsDelete`
	`@auth.on.crons.search`	列出 Cron 作业	`CronsSearch`
存储	`@auth.on.store`	所有存储操作	`Auth.types.on.store.value`
	`@auth.on.store.put`	存储项目	`Auth.types.on.store.put.value`
	`@auth.on.store.get`	获取项目	`Auth.types.on.store.get.value`
	`@auth.on.store.search`	搜索项目	`Auth.types.on.store.search.value`
	`@auth.on.store.delete`	删除项目	`Auth.types.on.store.delete.value`
	`@auth.on.store.list_namespaces`	列出命名空间	`Auth.types.on.store.list_namespaces.value`

存储授权不同于线程和助手授权。处理程序必须重写 value 中可变的 namespace 字段，以便按用户限定数据范围，而不是返回元数据筛选条件。有关完整示例，请参阅按用户隔离存储。

“关于运行”运行会根据其父线程限定访问范围。这意味着权限通常从线程继承，体现了数据模型的会话特性。除了创建操作以外，所有运行操作（读取、列出）都由线程处理程序控制。之所以提供专门的 create_run 处理程序来创建新运行，是因为该操作包含更多参数，你可以在处理程序中查看这些参数。

后续步骤

有关实现细节：

查看介绍性教程设置身份验证
参阅有关实现自定义身份验证处理程序的操作指南

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Develop your application

Capabilities

How to build

Auth & access control

Server customization

核心概念

身份验证与授权

默认安全模型

LangSmith

自托管

系统架构

身份验证

代理身份验证

使用 MCP 进行代理身份验证

授权

资源专属处理程序

筛选操作

常见访问模式

单一所有者资源

基于权限的访问

支持的资源

支持的操作和类型

后续步骤

​核心概念

​身份验证与授权

​默认安全模型

​LangSmith

​自托管

​系统架构

​身份验证

​代理身份验证

​使用 MCP 进行代理身份验证

​授权

​资源专属处理程序

​筛选操作

​常见访问模式

​单一所有者资源

​基于权限的访问

​支持的资源

​支持的操作和类型

​后续步骤

核心概念

身份验证与授权

默认安全模型

LangSmith

自托管

系统架构

身份验证

代理身份验证

使用 MCP 进行代理身份验证

授权

资源专属处理程序

筛选操作

常见访问模式

单一所有者资源

基于权限的访问

支持的资源

支持的操作和类型

后续步骤