身份认证与访问控制

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

核心概念

身份认证与授权的区别

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

• 身份认证（"AuthN"）验证你是谁。它对每个请求以中间件方式运行。
• 授权（"AuthZ"）确定你可以做什么。它按资源逐个验证用户的权限和角色。

在 LangSmith 中，身份认证由你的 @auth.authenticate 处理器负责，授权由你的 @auth.on 处理器负责。

默认安全模型

LangSmith 提供了不同的安全默认值：

LangSmith

• 默认使用 LangSmith API key
• 要求 x-api-key 头中包含有效的 API key
• 可以通过你的身份认证处理器进行自定义
• 当 langgraph.json 中的 auth.allow_langsmith_api_keys 为 true 时，可以在自定义身份认证之外同时接受 LangSmith API key（Agent Server v0.14.0rc2+）

NOTE

自定义身份认证 LangSmith 的所有套餐都支持自定义身份认证。

自托管

• 没有默认身份认证
• 可以完全灵活地实现你的安全模型
• 你可以控制身份认证与授权的所有方面

系统架构

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

1. 身份认证服务商（Identity Provider/IdP）
   • 一个管理用户身份和凭证的专用服务
   • 处理用户注册、登录、密码重置等
   • 在身份认证成功后签发令牌（JWT、会话令牌等）
   • 例如：Auth0、Supabase Auth、Okta 或你自己的认证服务器
2. Agent Server（资源服务器）
   • 你的智能体或 LangGraph 应用，其中包含业务逻辑和受保护的资源
   • 使用身份认证服务商验证令牌
   • 根据用户身份和权限实施访问控制
   • 不直接存储用户凭证
3. 客户端应用（前端）
   • Web 应用、移动应用或 API 客户端
   • 收集时效性强的用户凭证并发送给身份认证服务商
   • 从身份认证服务商接收令牌
   • 在向 Agent Server 发出的请求中包含这些令牌

以下是这些组件通常的交互方式：

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

你在 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(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"] 使用

支持的参数

@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 中返回的值会被添加到运行上下文中，为智能体提供用户范围的凭证，使其能够代表用户访问资源。

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 服务器、外部数据库，甚至其他智能体。

更多信息请参阅使用自定义身份认证指南。

使用 MCP 的智能体身份认证

有关如何让智能体通过 MCP 服务器进行身份认证的信息，请参阅 MCP 概念指南。

授权

身份认证后，LangGraph 会调用你的 @auth.on 处理器来控制对特定资源（例如线程、助手、定时任务）的访问。这些处理器可以：

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. 所有其他端点（例如，例如，删除助手、crons、store）对所有用户均被禁用。

TIP

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

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

过滤器操作

授权处理器可以返回 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)。

常见访问模式

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

单一所有者资源

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

@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。 如果注册了更具体的处理器，则对于该资源和操作不会调用更通用的处理器。

TIP

"类型安全" 每个处理函数在其 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
Store	@auth.on.store	所有 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

Store 授权与线程和助手不同。处理函数必须重写 value 中可变的 namespace 字段，以按用户限定数据范围，而不是返回元数据过滤器。有关演练，请参阅按用户隔离 store。

NOTE

"关于运行"

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

后续步骤

有关实现细节：

• 查看关于设置身份认证的入门教程
• 参阅关于实现自定义认证处理函数的操作指南

---