Parlant 生产 API 如何配置 JWT 鉴权与按操作速率限制API Hardening【免费下载链接】parlantBuild reliable customer-facing AI agents with Parlant: an interaction control harness optimized for controlled, consistent, and predictable LLM interactions.项目地址: https://gitcode.com/GitHub_Trending/pa/parlant当你把 Parlant 部署到生产环境时API 会暴露 Agent、Customer、Session 等管理接口。Parlant 自带的ProductionAuthorizationPolicy只放行少数操作如READ_AGENT、CREATE_GUEST_SESSION默认不做任何身份校验要让 API 只接受持有合法 JWT 的调用方并为每个操作operation设置独立的速率限制需要子类化ProductionAuthorizationPolicy并通过configure_container注册到p.Server中。本文基于 docs/production/api-hardening.md 和源码 src/parlant/api/authorization.py 说明这条路径。适用前提Parlant 要求 Python 3.10 及以上见 docs/quickstart/installation.md通过pip install parlant安装服务器默认监听http://localhost:8800。机制AuthorizationPolicy 如何拦截每个请求Parlant 的 API 加固由两部分组成Authorization Policies——控制谁可以访问哪些资源、执行哪些操作Rate Limiters——按操作频率限制请求防止滥用。所有策略都继承抽象基类AuthorizationPolicy它定义三个方法check_permission权限检查、check_rate_limit速率检查和组合两者的authorize。默认实现是顺序调用前两个方法任一拒绝就抛出授权异常源码 src/parlant/api/authorization.py#L164-L169。Parlant 用一个Operation枚举覆盖了全部 API 操作Agent、Customer、Session、Guideline、Journey 等的增删改查完整列表见 src/parlant/api/authorization.py#L32-L127权限和速率限制都以操作为粒度。内置策略有两个DevelopmentAuthorizationPolicy放行一切操作、不限速文档明确说明只适用于开发环境ProductionAuthorizationPolicy生产用策略其默认check_permission只放行READ_AGENT、CREATE_GUEST_SESSION、READ_SESSION、LIST_EVENTS、CREATE_CUSTOMER_EVENT五个操作src/parlant/api/authorization.py#L260-L270其余操作一律拒绝。文档建议实际部署中不要从零实现策略而是子类化ProductionAuthorizationPolicy做定制。ProductionAuthorizationPolicy的构造器还预留了两个扩展点self.default_limiter默认限流器可整体替换self.specific_limitersdict[Operation, 异步函数]为特定操作安装自定义限流函数检查时优先命中它未命中才走default_limitersrc/parlant/api/authorization.py#L273-L276。第一步实现带 JWT 鉴权的自定义策略文档给出的参考实现docs/production/api-hardening.md从请求头提取Bearertoken用jwt库校验并为type m2mmachine-to-machine的 token 放行一组管理操作。import parlant.sdk as p import jwt from fastapi import HTTPException from limits import RateLimitItemPerMinute, RateLimitItemPerHour from limits.storage import RedisStorage from limits.strategies import SlidingWindowCounterRateLimiter class CustomAuthorizationPolicy(p.ProductionAuthorizationPolicy): def __init__(self, secret_key: str, algorithm: str HS256): super().__init__() self.secret_key secret_key self.algorithm algorithm async def _extract_token(self, request: fastapi.Request) - dict | None: Extract and validate JWT token from request auth_header request.headers.get(Authorization) if not auth_header or not auth_header.startswith(Bearer ): return None token auth_header.split( )[1] try: payload jwt.decode(token, self.secret_key, algorithms[self.algorithm]) return payload except jwt.JWTError: # Raise 403 for invalid tokens, None for missing tokens is OK raise HTTPException( status_code403, detailInvalid access token ) async def check_permission( self, request: fastapi.Request, operation: p.Operation ) - bool: Enhanced permission checking with M2M token support token_payload await self._extract_token(request) # If we have a valid M2M (machine-to-machine) token, allow additional operations if token_payload and token_payload.get(type) m2m: m2m_operations { # Allow M2M tokens to perform administrative operations p.Operation.CREATE_AGENT, p.Operation.READ_AGENT, p.Operation.UPDATE_AGENT, p.Operation.DELETE_AGENT, p.Operation.CREATE_CUSTOMER, p.Operation.READ_CUSTOMER, p.Operation.UPDATE_CUSTOMER, p.Operation.DELETE_CUSTOMER, p.Operation.CREATE_CUSTOMER_SESSION, p.Operation.LIST_SESSIONS, p.Operation.UPDATE_SESSION, p.Operation.DELETE_SESSION, # Add other operations your M2M integration needs } if operation in m2m_operations: return True # For all other cases, delegate to the parent ProductionAuthorizationPolicy return await super().check_permission(request, operation)几点需要注意的语义都来自文档和源码无 token 与坏 token 的处理不同请求头里没有Authorization: Bearer ...时_extract_token返回None随后交给父类ProductionAuthorizationPolicy.check_permission判断token 解码失败jwt.JWTError则直接抛 403Invalid access token。M2M 白名单是可编辑的m2m_operations集合里注释明确写了 Add other operations your M2M integration needs按你的集成需要增删操作。示例中的secret_key、algorithm默认HS256是构造参数实际使用时替换为你自己的密钥值。第二步按操作配置速率限制替换 default_limiter文档推荐的主路径ProductionAuthorizationPolicy文档推荐的常见做法是覆盖self.default_limiter。注意BasicRateLimiter的限制是按 IP 地址计的——配置RateLimitItemPerMinute(100)表示每个 IP 地址每分钟 100 次请求。from limits import RateLimitItemPerMinute, RateLimitItemPerHour from limits.storage import RedisStorage from limits.strategies import SlidingWindowCounterRateLimiter # Example with Redis storage and custom limits class CustomAuthorizationPolicy(p.ProductionAuthorizationPolicy): def __init__(self, ...): super().__init__() # ... self.default_limiter p.BasicRateLimiter( rate_limit_item_per_operation{ # Use the default rate limit for most operations **self.default_limiter.rate_limit_item_per_operation, # Override specific operations with custom limits p.Operation.READ_SESSION: RateLimitItemPerMinute(200), p.Operation.LIST_EVENTS: RateLimitItemPerMinute(1000), }, # Use a custom storage backend (e.g., Redis) storageRedisStorage(redis://localhost:6379), # Use a custom window strategy limiter_typeSlidingWindowCounterRateLimiter, )其中**self.default_limiter.rate_limit_item_per_operation保留了父类已有的默认限额如READ_AGENT: RateLimitItemPerMinute(30)、LIST_EVENTS: RateLimitItemPerMinute(240)见 src/parlant/api/authorization.py#L225-L235再叠加你对特定操作的覆盖。BasicRateLimiter基于limits库可选项有均来自文档限流项RateLimitItemPerMinute(n)、RateLimitItemPerSecond(n)、RateLimitItemPerHour(n)存储RedisStorage()、MemoryStorage()等limits库提供的存储不传storage时源码默认MemoryStoragesrc/parlant/api/authorization.py#L289窗口策略MovingWindowRateLimiter默认、FixedWindowRateLimiter、SlidingWindowCounterRateLimiter。未配置限额的操作会落到BasicRateLimiter的内部默认值RateLimitItemPerMinute(100)src/parlant/api/authorization.py#L290。关于 IP 的识别BasicRateLimiter按顺序取x-forwarded-for头第一个值、x-real-ip、cf-connecting-ip最后才是request.client.hostsrc/parlant/api/authorization.py#L318-L329。也就是说如果你把 Parlant 放在反向代理后面限流 key 由代理写入的x-forwarded-for决定。如果需要更彻底的掌控也可以直接子类化抽象RateLimiter类从零实现再赋给self.default_limiter文档认为上面覆盖default_limiter的方式适合大多数场景。可选为单个操作挂自定义限流函数第二种方式是用self.specific_limiters给特定操作装自定义函数签名接收request和operation返回布尔值表示是否未超限class CustomAuthorizationPolicy(p.ProductionAuthorizationPolicy): def __init__(self, ...): super().__init__() # ... self.specific_limiters[p.Operation.DELETE_AGENT] self._custom_delete_limiter async def _custom_delete_limiter( self, request: fastapi.Request, operation: p.Operation ) - bool: # Implement your custom logic here ...check_rate_limit先查specific_limiters命中则用自定义函数否则走default_limiter因此这个分支只影响你显式注册的操作。第三步把策略注册到服务器通过configure_container钩子把自定义策略放进依赖容器容器机制见 docs/advanced/engine-extensions.mdasync def configure_container( container: p.Container ) - p.Container: container[p.AuthorizationPolicy] CustomAuthorizationPolicy( secret_keyyour-jwt-secret-key, algorithmHS256, ) return container这里的your-jwt-secret-key是文档示例占位符替换为你自己的 JWT 签名密钥。然后把它传给p.Serverasync def main(): # Create Parlant server with custom authorization async with p.Server( configure_containerconfigure_container, ) as server: # Your agent logic here await server.serve() if __name__ __main__: asyncio.run(main())验证如何判断鉴权与限流生效Parlant 在 API 层注册了统一的异常处理src/parlant/api/app.py#L202-L222这给出了两种失败对应的 HTTP 状态权限被拒AuthorizationException包括缺少客户端 IP 的鉴权失败→403 FORBIDDEN超出速率限制RateLimitExceededException→429 TOO MANY REQUESTS坏 token 在你的策略里直接抛HTTPException(status_code403, detailInvalid access token)。因此验证顺序是鉴权带合法 JWTAuthorization: Bearer token请求一个 M2M 操作如创建 customer session应正常返回不带 token 请求超出父类放行范围的操作应收到 403提供错误签名的 token 应收到Invalid access token的 403。限流用同一 IP 对某操作连续请求超过配置的每分钟配额。例如把p.Operation.LIST_EVENTS配为RateLimitItemPerMinute(2)时仓库测试 tests/api/test_authorization.py#L52-L63 展示的正是这个模式前两次check返回True第三次返回False对应的 HTTP 表现即前两次正常、第三次 429。测试还验证了配额按 IP 隔离不同x-forwarded-for各有一份配额以及x-forwarded-for优先于客户端 IP。限制与注意事项DevelopmentAuthorizationPolicy放行一切只用于开发生产必须使用ProductionAuthorizationPolicy或其子类。ProductionAuthorizationPolicy.configure_app默认设置 CORSallow_origins[*]源码注释建议在子类中覆盖为更严格的来源白名单src/parlant/api/authorization.py#L237-L252。本文的 JWT 校验基于PyJWT示例中的import jwt文档未列出额外pip install命令jwt与limits库需按你的环境自行安装p.BasicRateLimiter、p.ProductionAuthorizationPolicy、p.Operation等均由parlant.sdk导出src/parlant/sdk.py#L77-L83。文档同时提到若需要同时完全接管权限与限流也可以直接子类化抽象AuthorizationPolicy从零实现但文档推荐在ProductionAuthorizationPolicy基础上扩展。完成配置后的自然延伸是继续参考 docs/production/input-moderation.md 为输入侧增加内容审核与本文的 API 层防护配合使用。【免费下载链接】parlantBuild reliable customer-facing AI agents with Parlant: an interaction control harness optimized for controlled, consistent, and predictable LLM interactions.项目地址: https://gitcode.com/GitHub_Trending/pa/parlant创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Parlant 生产 API 如何配置 JWT 鉴权与按操作速率限制（API Hardening）
Parlant 生产 API 如何配置 JWT 鉴权与按操作速率限制API Hardening【免费下载链接】parlantBuild reliable customer-facing AI agents with Parlant: an interaction control harness optimized for controlled, consistent, and predictable LLM interactions.项目地址: https://gitcode.com/GitHub_Trending/pa/parlant当你把 Parlant 部署到生产环境时API 会暴露 Agent、Customer、Session 等管理接口。Parlant 自带的ProductionAuthorizationPolicy只放行少数操作如READ_AGENT、CREATE_GUEST_SESSION默认不做任何身份校验要让 API 只接受持有合法 JWT 的调用方并为每个操作operation设置独立的速率限制需要子类化ProductionAuthorizationPolicy并通过configure_container注册到p.Server中。本文基于 docs/production/api-hardening.md 和源码 src/parlant/api/authorization.py 说明这条路径。适用前提Parlant 要求 Python 3.10 及以上见 docs/quickstart/installation.md通过pip install parlant安装服务器默认监听http://localhost:8800。机制AuthorizationPolicy 如何拦截每个请求Parlant 的 API 加固由两部分组成Authorization Policies——控制谁可以访问哪些资源、执行哪些操作Rate Limiters——按操作频率限制请求防止滥用。所有策略都继承抽象基类AuthorizationPolicy它定义三个方法check_permission权限检查、check_rate_limit速率检查和组合两者的authorize。默认实现是顺序调用前两个方法任一拒绝就抛出授权异常源码 src/parlant/api/authorization.py#L164-L169。Parlant 用一个Operation枚举覆盖了全部 API 操作Agent、Customer、Session、Guideline、Journey 等的增删改查完整列表见 src/parlant/api/authorization.py#L32-L127权限和速率限制都以操作为粒度。内置策略有两个DevelopmentAuthorizationPolicy放行一切操作、不限速文档明确说明只适用于开发环境ProductionAuthorizationPolicy生产用策略其默认check_permission只放行READ_AGENT、CREATE_GUEST_SESSION、READ_SESSION、LIST_EVENTS、CREATE_CUSTOMER_EVENT五个操作src/parlant/api/authorization.py#L260-L270其余操作一律拒绝。文档建议实际部署中不要从零实现策略而是子类化ProductionAuthorizationPolicy做定制。ProductionAuthorizationPolicy的构造器还预留了两个扩展点self.default_limiter默认限流器可整体替换self.specific_limitersdict[Operation, 异步函数]为特定操作安装自定义限流函数检查时优先命中它未命中才走default_limitersrc/parlant/api/authorization.py#L273-L276。第一步实现带 JWT 鉴权的自定义策略文档给出的参考实现docs/production/api-hardening.md从请求头提取Bearertoken用jwt库校验并为type m2mmachine-to-machine的 token 放行一组管理操作。import parlant.sdk as p import jwt from fastapi import HTTPException from limits import RateLimitItemPerMinute, RateLimitItemPerHour from limits.storage import RedisStorage from limits.strategies import SlidingWindowCounterRateLimiter class CustomAuthorizationPolicy(p.ProductionAuthorizationPolicy): def __init__(self, secret_key: str, algorithm: str HS256): super().__init__() self.secret_key secret_key self.algorithm algorithm async def _extract_token(self, request: fastapi.Request) - dict | None: Extract and validate JWT token from request auth_header request.headers.get(Authorization) if not auth_header or not auth_header.startswith(Bearer ): return None token auth_header.split( )[1] try: payload jwt.decode(token, self.secret_key, algorithms[self.algorithm]) return payload except jwt.JWTError: # Raise 403 for invalid tokens, None for missing tokens is OK raise HTTPException( status_code403, detailInvalid access token ) async def check_permission( self, request: fastapi.Request, operation: p.Operation ) - bool: Enhanced permission checking with M2M token support token_payload await self._extract_token(request) # If we have a valid M2M (machine-to-machine) token, allow additional operations if token_payload and token_payload.get(type) m2m: m2m_operations { # Allow M2M tokens to perform administrative operations p.Operation.CREATE_AGENT, p.Operation.READ_AGENT, p.Operation.UPDATE_AGENT, p.Operation.DELETE_AGENT, p.Operation.CREATE_CUSTOMER, p.Operation.READ_CUSTOMER, p.Operation.UPDATE_CUSTOMER, p.Operation.DELETE_CUSTOMER, p.Operation.CREATE_CUSTOMER_SESSION, p.Operation.LIST_SESSIONS, p.Operation.UPDATE_SESSION, p.Operation.DELETE_SESSION, # Add other operations your M2M integration needs } if operation in m2m_operations: return True # For all other cases, delegate to the parent ProductionAuthorizationPolicy return await super().check_permission(request, operation)几点需要注意的语义都来自文档和源码无 token 与坏 token 的处理不同请求头里没有Authorization: Bearer ...时_extract_token返回None随后交给父类ProductionAuthorizationPolicy.check_permission判断token 解码失败jwt.JWTError则直接抛 403Invalid access token。M2M 白名单是可编辑的m2m_operations集合里注释明确写了 Add other operations your M2M integration needs按你的集成需要增删操作。示例中的secret_key、algorithm默认HS256是构造参数实际使用时替换为你自己的密钥值。第二步按操作配置速率限制替换 default_limiter文档推荐的主路径ProductionAuthorizationPolicy文档推荐的常见做法是覆盖self.default_limiter。注意BasicRateLimiter的限制是按 IP 地址计的——配置RateLimitItemPerMinute(100)表示每个 IP 地址每分钟 100 次请求。from limits import RateLimitItemPerMinute, RateLimitItemPerHour from limits.storage import RedisStorage from limits.strategies import SlidingWindowCounterRateLimiter # Example with Redis storage and custom limits class CustomAuthorizationPolicy(p.ProductionAuthorizationPolicy): def __init__(self, ...): super().__init__() # ... self.default_limiter p.BasicRateLimiter( rate_limit_item_per_operation{ # Use the default rate limit for most operations **self.default_limiter.rate_limit_item_per_operation, # Override specific operations with custom limits p.Operation.READ_SESSION: RateLimitItemPerMinute(200), p.Operation.LIST_EVENTS: RateLimitItemPerMinute(1000), }, # Use a custom storage backend (e.g., Redis) storageRedisStorage(redis://localhost:6379), # Use a custom window strategy limiter_typeSlidingWindowCounterRateLimiter, )其中**self.default_limiter.rate_limit_item_per_operation保留了父类已有的默认限额如READ_AGENT: RateLimitItemPerMinute(30)、LIST_EVENTS: RateLimitItemPerMinute(240)见 src/parlant/api/authorization.py#L225-L235再叠加你对特定操作的覆盖。BasicRateLimiter基于limits库可选项有均来自文档限流项RateLimitItemPerMinute(n)、RateLimitItemPerSecond(n)、RateLimitItemPerHour(n)存储RedisStorage()、MemoryStorage()等limits库提供的存储不传storage时源码默认MemoryStoragesrc/parlant/api/authorization.py#L289窗口策略MovingWindowRateLimiter默认、FixedWindowRateLimiter、SlidingWindowCounterRateLimiter。未配置限额的操作会落到BasicRateLimiter的内部默认值RateLimitItemPerMinute(100)src/parlant/api/authorization.py#L290。关于 IP 的识别BasicRateLimiter按顺序取x-forwarded-for头第一个值、x-real-ip、cf-connecting-ip最后才是request.client.hostsrc/parlant/api/authorization.py#L318-L329。也就是说如果你把 Parlant 放在反向代理后面限流 key 由代理写入的x-forwarded-for决定。如果需要更彻底的掌控也可以直接子类化抽象RateLimiter类从零实现再赋给self.default_limiter文档认为上面覆盖default_limiter的方式适合大多数场景。可选为单个操作挂自定义限流函数第二种方式是用self.specific_limiters给特定操作装自定义函数签名接收request和operation返回布尔值表示是否未超限class CustomAuthorizationPolicy(p.ProductionAuthorizationPolicy): def __init__(self, ...): super().__init__() # ... self.specific_limiters[p.Operation.DELETE_AGENT] self._custom_delete_limiter async def _custom_delete_limiter( self, request: fastapi.Request, operation: p.Operation ) - bool: # Implement your custom logic here ...check_rate_limit先查specific_limiters命中则用自定义函数否则走default_limiter因此这个分支只影响你显式注册的操作。第三步把策略注册到服务器通过configure_container钩子把自定义策略放进依赖容器容器机制见 docs/advanced/engine-extensions.mdasync def configure_container( container: p.Container ) - p.Container: container[p.AuthorizationPolicy] CustomAuthorizationPolicy( secret_keyyour-jwt-secret-key, algorithmHS256, ) return container这里的your-jwt-secret-key是文档示例占位符替换为你自己的 JWT 签名密钥。然后把它传给p.Serverasync def main(): # Create Parlant server with custom authorization async with p.Server( configure_containerconfigure_container, ) as server: # Your agent logic here await server.serve() if __name__ __main__: asyncio.run(main())验证如何判断鉴权与限流生效Parlant 在 API 层注册了统一的异常处理src/parlant/api/app.py#L202-L222这给出了两种失败对应的 HTTP 状态权限被拒AuthorizationException包括缺少客户端 IP 的鉴权失败→403 FORBIDDEN超出速率限制RateLimitExceededException→429 TOO MANY REQUESTS坏 token 在你的策略里直接抛HTTPException(status_code403, detailInvalid access token)。因此验证顺序是鉴权带合法 JWTAuthorization: Bearer token请求一个 M2M 操作如创建 customer session应正常返回不带 token 请求超出父类放行范围的操作应收到 403提供错误签名的 token 应收到Invalid access token的 403。限流用同一 IP 对某操作连续请求超过配置的每分钟配额。例如把p.Operation.LIST_EVENTS配为RateLimitItemPerMinute(2)时仓库测试 tests/api/test_authorization.py#L52-L63 展示的正是这个模式前两次check返回True第三次返回False对应的 HTTP 表现即前两次正常、第三次 429。测试还验证了配额按 IP 隔离不同x-forwarded-for各有一份配额以及x-forwarded-for优先于客户端 IP。限制与注意事项DevelopmentAuthorizationPolicy放行一切只用于开发生产必须使用ProductionAuthorizationPolicy或其子类。ProductionAuthorizationPolicy.configure_app默认设置 CORSallow_origins[*]源码注释建议在子类中覆盖为更严格的来源白名单src/parlant/api/authorization.py#L237-L252。本文的 JWT 校验基于PyJWT示例中的import jwt文档未列出额外pip install命令jwt与limits库需按你的环境自行安装p.BasicRateLimiter、p.ProductionAuthorizationPolicy、p.Operation等均由parlant.sdk导出src/parlant/sdk.py#L77-L83。文档同时提到若需要同时完全接管权限与限流也可以直接子类化抽象AuthorizationPolicy从零实现但文档推荐在ProductionAuthorizationPolicy基础上扩展。完成配置后的自然延伸是继续参考 docs/production/input-moderation.md 为输入侧增加内容审核与本文的 API 层防护配合使用。【免费下载链接】parlantBuild reliable customer-facing AI agents with Parlant: an interaction control harness optimized for controlled, consistent, and predictable LLM interactions.项目地址: https://gitcode.com/GitHub_Trending/pa/parlant创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考