身份验证 & 访问控制

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

仅限 Python

我们目前仅在 Python 部署中支持自定义身份验证和授权，版本为 langgraph-api>=0.0.11。对 LangGraph.JS 的支持即将添加。

核心概念

身份验证 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. 客户端应用程序 (前端)

   • Web 应用、移动应用或 API 客户端
   • 收集时间敏感的用户凭据并发送给身份验证提供商
   • 从身份验证提供商接收令牌
   • 在对 LangGraph 后端的请求中包含这些令牌

以下是这些组件通常如何交互

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

    Client->>Auth: 1. Login (username/password)
    Auth-->>Client: 2. Return token
    Client->>LG: 3. Request with token
    Note over LG: 4. Validate token (@auth.authenticate)
    LG-->>Auth:  5. Fetch user info
    Auth-->>LG: 6. Confirm validity
    Note over LG: 7. Apply access control (@auth.on.*)
    LG-->>Client: 8. Return resources

你的 LangGraph 中的 @auth.authenticate 处理程序处理步骤 4-6，而你的 @auth.on 处理程序实现步骤 7。

身份验证

LangGraph 中的身份验证作为每个请求的中间件运行。你的 @auth.authenticate 处理程序接收请求信息，并且应该

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

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    # Validate credentials (e.g., API key, JWT token)
    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 user info - only identity and is_authenticated are required
    # Add any additional fields you need for authorization
    return {
        "identity": "user-123",        # Required: unique user identifier
        "is_authenticated": True,      # Optional: assumed True by default
        "permissions": ["read", "write"] # Optional: for permission-based auth
        # You can add more custom fields if you want to implement other auth patterns
        "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): Authorization 标头值 (例如，"Bearer")

在我们的许多教程中，我们将仅显示 “authorization” 参数以保持简洁，但你可以选择接受更多信息，以根据需要实现你的自定义身份验证方案。

授权

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

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

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

@auth.on
async def add_owner(
    ctx: Auth.types.AuthContext,
    value: dict  # The payload being sent to this access method
) -> dict:  # Returns a filter dict that restricts access to resources
    """Authorize all access to threads, runs, crons, and assistants.

    This handler does two things:
        - Adds a value to resource metadata (to persist with the resource so it can be filtered later)
        - Returns a filter (to restrict access to existing resources)

    Args:
        ctx: Authentication context containing user info, permissions, the path, and
        value: The request payload sent to the endpoint. For creation
              operations, this contains the resource parameters. For read
              operations, this contains the resource being accessed.

    Returns:
        A filter dictionary that LangGraph uses to restrict access to resources.
        See [Filter Operations](#filter-operations) for supported operators.
    """
    # Create filter to restrict access to just this user's resources
    filters = {"owner": ctx.user.identity}

    # Get or create the metadata dictionary in the payload
    # This is where we store persistent info about the resource
    metadata = value.setdefault("metadata", {})

    # Add owner to metadata - if this is a create or update operation,
    # this information will be saved with the resource
    # So we can filter by it later in read operations
    metadata.update(filters)

    # Return filters to restrict access
    # These filters are applied to ALL operations (create, read, update, search, etc.)
    # to ensure users can only access their own resources
    return filters

资源特定处理程序

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

1. 经过身份验证的用户能够创建线程、读取线程、在线程上创建运行
2. 只有具有 “assistants:create” 权限的用户才允许创建新的助手
3. 所有其他端点（例如，例如，删除助手、cron 作业、存储）都对所有用户禁用。

支持的处理程序

有关支持的资源和操作的完整列表，请参阅下面的支持的资源部分。

# Generic / global handler catches calls that aren't handled by more specific handlers
@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"
    )

# Matches the "thread" resource and all actions - create, read, update, delete, search
# Since this is **more specific** than the generic @auth.on handler, it will take precedence
# over the generic handler for all actions on the "threads" resource
@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."
        )
    # Setting metadata on the thread being created
    # will ensure that the resource contains an "owner" field
    # Then any time a user tries to access this thread or runs within the thread,
    # we can filter by owner
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

# Thread creation. This will match only on thread create actions
# Since this is **more specific** than both the generic @auth.on handler and the @auth.on.threads handler,
# it will take precedence for any "create" actions on the "threads" resources
@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    # Setting metadata on the thread being created
    # will ensure that the resource contains an "owner" field
    # Then any time a user tries to access this thread or runs within the thread,
    # we can filter by owner
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

# Reading a thread. Since this is also more specific than the generic @auth.on handler, and the @auth.on.threads handler,
# it will take precedence for any "read" actions on the "threads" resource
@auth.on.threads.read
async def on_thread_read(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.read.value
):
    # Since we are reading (and not creating) a thread,
    # we don't need to set metadata. We just need to
    # return a filter to ensure users can only see their own threads
    return {"owner": ctx.user.identity}

# Run creation, streaming, updates, etc.
# This takes precedenceover the generic @auth.on handler and the @auth.on.threads handler
@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
    # Inherit thread's access control
    return {"owner": ctx.user.identity}

# Assistant creation
@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 处理程序。但是，update 线程的请求将由全局处理程序处理，因为我们没有针对该资源和操作的更特定处理程序。创建、更新的请求，

过滤器操作

授权处理程序可以返回 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 的资源。有关更多信息，请参阅参考此处。

常用访问模式

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

单所有者资源

这种常用模式允许你将所有线程、助手、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}

基于权限的访问

此模式允许你根据 权限 控制访问。如果你希望某些角色对资源具有更广泛或更受限制的访问权限，这将非常有用。

# In your auth handler:
@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    ...
    return {
        "identity": "user-123",
        "is_authenticated": True,
        "permissions": ["threads:write", "threads:read"]  # Define permissions in auth
    }

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.<resource>.<action>.value。例如

@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.threads.create.value  # Specific type for thread creation
):
    ...

@auth.on.threads
async def on_threads(
    ctx: Auth.types.AuthContext,
    value: Auth.types.on.threads.value  # Union type of all thread actions
):
    ...

@auth.on
async def on_all(
    ctx: Auth.types.AuthContext,
    value: dict  # Union type of all possible actions
):
    ...

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

支持的操作和类型

以下是所有支持的操作处理程序

资源	处理程序	描述	值类型
线程	@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

关于运行

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

下一步

有关实施细节

• 查看关于设置身份验证的入门教程
• 请参阅关于实施自定义身份验证处理程序的操作指南

评论