以编程方式使用文档

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

核心概念

身份验证与授权

虽然这些术语经常被交替使用,但它们代表了不同的安全概念:

  • * **身份验证** ("AuthN")验证 _你是谁_ 。这作为中间件为每个请求运行。
  • * **授权** ("AuthZ")决定 _你可以做什么_。这会按资源验证用户的权限和角色。

在 LangSmith 中,身份验证由您的 @auth.authenticate 处理程序处理,授权由您的 @auth.on 处理程序处理。

默认安全模型

LangSmith 提供不同的安全默认值:

LangSmith

  • * 默认使用 LangSmith API 密钥
  • * 需要在 x-api-key 标头中提供有效的 API 密钥
  • * 可以使用您的身份验证处理程序进行自定义

Self-hosted

  • * 无默认身份验证
  • * 完全灵活地实施您的安全模式
  • * 您控制身份验证和授权的各个方面

系统架构

典型的身份验证设置涉及三个主要组件:

1. **身份验证提供者** (Identity Provider/IdP) * 管理用户身份和凭证的专用服务 * 处理用户注册、登录、密码重置等 * 身份验证成功后颁发令牌(JWT、会话令牌等) * 示例:Auth0、Supabase Auth、Okta 或您自己的身份验证服务器 2. **代理服务器** (资源服务器) * 您的代理或 LangGraph 应用程序,包含业务逻辑和受保护的资源 * 使用身份验证提供者验证令牌 * 根据用户身份和权限实施访问控制 * 不直接存储用户凭证 3. **客户端应用程序** (前端) * Web 应用、移动应用或 API 客户端 * 收集时间敏感的用户凭证并将其发送至身份验证提供者 * 从身份验证提供者接收令牌 * 将这些令牌包含在向代理服务器发出的请求中

这些组件通常如何交互:

sequenceDiagram
    participant Client as Client App
    participant Auth as Auth Provider
    participant LG as Agent Server

    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

您的 @auth.authenticate 处理程序在 LangGraph 中处理步骤 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(b"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"]

Supported Parameters

@auth.authenticate 处理程序可以按名称接受以下任何参数:

  • * request (Request):原始 ASGI 请求对象
  • * path (str):请求路径,例如, "/threads/abcd-1234-abcd-1234/runs/abcd-1234-abcd-1234/stream"
  • * method (str):HTTP 方法,例如, "GET"
  • * 路径_params (dict[str, str]):URL 路径参数,例如, {"thread_id": "abcd-1234-abcd-1234", "run_id": "abcd-1234-abcd-1234"}
  • * 查询_params (dict[str, str]):URL 查询参数,例如, {"stream": "true"}
  • * headers (dict[bytes, bytes]):请求头
  • * authorization (str | None):Authorization 标头值(例如, "Bearer <token>")

在我们的许多教程中,为简洁起见,我们只显示 "authorization" 参数,但您可以根据需要选择接受更多信息 来实现您的自定义身份验证方案。

代理身份验证

自定义身份验证允许委托访问。您在 @auth.authenticate 中返回的值被添加到运行上下文,给予代理用户范围的凭据,使它们能够代表用户访问资源。

sequenceDiagram
  %% Actors
  participant ClientApp as Client
  participant AuthProv  as Auth Provider
  participant LangGraph as Agent Server
  participant SecretStore as Secret Store
  participant ExternalService as External Service

  %% Platform login / AuthN
  ClientApp  ->> AuthProv: 1. Login (username / password)
  AuthProv   -->> ClientApp: 2. Return token
  ClientApp  ->> LangGraph: 3. Request with token

  Note over LangGraph: 4. Validate token (@auth.authenticate)
  LangGraph  -->> AuthProv: 5. Fetch user info
  AuthProv   -->> LangGraph: 6. Confirm validity

  %% Fetch user tokens from secret store
  LangGraph  ->> SecretStore: 6a. Fetch user tokens
  SecretStore -->> LangGraph: 6b. Return tokens

  Note over LangGraph: 7. Apply access control (@auth.on.*)

  %% External Service round-trip
  LangGraph  ->> ExternalService: 8. Call external service (with header)
  Note over ExternalService: 9. External service validates header and executes action
  ExternalService  -->> LangGraph: 10. Service response

  %% Return to caller
  LangGraph  -->> ClientApp: 11. Return resources

身份验证后,平台会创建一个特殊的配置对象,通过可配置上下文传递给您的图和所有节点。 此对象包含当前用户的信息,包括您从 @auth.authenticate 处理程序返回的任何自定义字段。

要使代理能够代表用户操作,请使用 自定义身份验证中间件。这将允许代理代表用户与外部系统(如 MCP 服务器、外部数据库甚至其他代理)交互。

有关更多信息,请参阅 使用自定义身份验证 guide.

MCP 代理身份验证

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

授权

身份验证后,LangGraph 调用您的 @auth.on 处理程序来控制对特定资源(例如,线程、助手、定时任务)的访问。这些处理程序可以:

  1. 通过修改 value["metadata"] 字典直接添加资源创建期间要保存的元数据。请参阅 支持的操列表 获取每个操作的值可以采用的类型列表。
  2. Filter resources by metadata during search/list or read operations by returning a 过滤器字典.
  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

<a id="resource-specific-handlers"></a> ### 特定于资源的处理程序

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

  1. 已认证的用户能够创建线程、读取线程以及在线程上创建运行
  2. 只有拥有 "assistants:create" 权限的用户才被允许创建新助手
  3. 所有其他端点(例如删除助手、定时任务、存储)均对所有用户禁用。
# 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(
    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}


# 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
):
    # Reject if the user does not have write access
    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}

# 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 线程的请求将由全局处理程序处理,因为我们没有该资源和操作的更具体的处理程序。

<a id="filter-operations"></a> ### 过滤器操作

授权处理程序可以返回 None、布尔值或过滤器字典。

  • * NoneTrue 表示"授权访问所有底层资源"
  • * 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]}}). The value here must be an element of the list or a subset of the elements of the list, respectively. The metadata in the stored resource must be a list/container type.

具有多个键的字典使用逻辑 AND 过滤器处理。例如, {"owner": org_id, "allowed_users": {"$contains": user_id}} 将仅匹配元数据中"owner"为 org_id 且"allowed_users"列表包含 user_id. 请参阅参考 Auth(Auth) 了解更多信息。

常见访问模式

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

单一所有者资源

这种常见模式允许您将所有线程、助手、计划任务和运行绑定到单一用户。它适用于常见的单用户用例,如常规聊天机器人风格的应用程序。

@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. **资源处理程序** (e.g., @auth.on.threads, @auth.on.assistants, @auth.on.crons): 匹配特定资源的所有操作
  3. **操作处理程序** (e.g., @auth.on.threads.create, @auth.on.threads.read): 匹配特定资源上的特定操作

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

<a id="supported-actions"></a> #### 支持的操作和类型

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

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

存储授权与线程和助手不同。处理器必须重写可变的 namespace 字段 value 以按用户划分数据范围,而不是返回元数据过滤器。相关演练,请参阅 按用户隔离存储.

后续步骤

有关实现详情: