feat(conversational): implement route permissions and access control

- Added  to  for defining access control on routes.
- Introduced  decorator to specify permissions needed for route access.
- Enhanced  with methods to check route access based on user permissions.
- Updated documentation across multiple languages to reflect changes in route permissions and usage examples.
- Added tests to verify permission handling and redirection to  for unauthorized access.
This commit is contained in:
Lorenze Jay
2026-06-06 15:09:07 -07:00
parent 913a3abead
commit 9b251b1de4
9 changed files with 462 additions and 14 deletions

View File

@@ -316,6 +316,8 @@ RouterConfig(
route_descriptions={
"INTERNET_SEARCH": "تجاوز الـ docstring لهذا المسار فقط.",
},
route_permissions={"INTERNET_SEARCH": "web_search"}, # تجاوز اختياري
permission_denied_route="permission_denied", # الافتراضي
default_intent="converse", # يُستخدم عند فشل LLM أو غيابه
fallback_intent="converse", # يُستخدم عندما يعيد LLM مساراً غير صالح
intent_field="intent",
@@ -350,6 +352,58 @@ Routes:
`RouterConfig.prompt` مخصص لـ **تأطير النطاق** (شخصية المساعد، قواعد العمل، النبرة). فهرس المسارات يُبنى تلقائياً — لا تُدرج المسارات في `prompt`؛ سيختل التزامن لحظة إضافة معالج جديد.
### أذونات المسارات
استخدم `@listen(..., required_permissions=[...])` لحماية المسارات بعد أن يختار الموجّه مساراً صالحاً. يتحقق المفوّض الافتراضي من مجموعة `permissions` في `self.state`؛ وتُعاد الجولات المرفوضة إلى `permission_denied`.
```python
from crewai import Flow
from crewai.experimental.conversational import (
ConversationConfig,
ConversationState,
RouterConfig,
)
from crewai.flow import listen, start
from pydantic import Field
class AppState(ConversationState):
permissions: set[str] = Field(default_factory=set)
@ConversationConfig(
router=RouterConfig(
prompt="أرسل بحث الويب إلى INTERNET_SEARCH وطلبات التدقيق إلى ADMIN_REPORT.",
),
)
class SupportFlow(Flow[AppState]):
conversational = True
@start()
def load_permissions(self) -> None:
if self.state.session_ready:
return
self.state.permissions = {"web_search"}
self.state.session_ready = True
@listen("INTERNET_SEARCH", required_permissions=["web_search"])
def internet_search(self) -> str:
...
@listen("ADMIN_REPORT", required_permissions=["admin"])
def admin_report(self) -> str:
...
@listen("permission_denied")
def deny(self) -> str:
reply = "ليست لديك صلاحية لتنفيذ هذا الإجراء."
self.append_assistant_message(reply)
return reply
```
استخدم `RouterConfig.route_permissions` عندما تحتاج إلى تكوين الأذونات بعيداً عن المعالجات. للتفويض المخصص، تجاوز `can_access_route(required_permissions)` بدلاً من تجاوز `route_turn`.
### المسارات المدمجة
| المسار | المعالج | الغرض |

View File

@@ -333,6 +333,8 @@ router_config = RouterConfig(
route_descriptions={
"INTERNET_SEARCH": "Override the docstring for this one route.",
},
route_permissions={"INTERNET_SEARCH": "web_search"}, # optional override
permission_denied_route="permission_denied", # default
default_intent="converse", # used when LLM call fails or no LLM available
fallback_intent="converse", # used when LLM returns an invalid route
intent_field="intent",
@@ -370,6 +372,63 @@ Routes:
`RouterConfig.prompt` is for **domain framing** (assistant persona, business rules, voice). The route catalog is auto-built — don't list routes in `prompt`; they'll drift the moment you add a handler.
### Route permissions
Use `@listen(..., required_permissions=[...])` to gate protected routes after the router
selects a valid route. The default authorizer checks for a `permissions`
collection on `self.state`; denied turns redirect to `permission_denied`.
```python
from crewai import Flow
from crewai.experimental.conversational import (
ConversationConfig,
ConversationState,
RouterConfig,
)
from crewai.flow import listen, start
from pydantic import Field
class AppState(ConversationState):
permissions: set[str] = Field(default_factory=set)
@ConversationConfig(
router=RouterConfig(
prompt="Send web lookups to INTERNET_SEARCH and audits to ADMIN_REPORT.",
),
)
class SupportFlow(Flow[AppState]):
conversational = True
@start()
def load_permissions(self) -> None:
if self.state.session_ready:
return
self.state.permissions = {"web_search"}
self.state.session_ready = True
@listen("INTERNET_SEARCH", required_permissions=["web_search"])
def internet_search(self) -> str:
...
@listen("ADMIN_REPORT", required_permissions=["admin"])
def admin_report(self) -> str:
...
@listen("permission_denied")
def deny(self) -> str:
reply = "You do not have permission for that action."
self.append_assistant_message(reply)
return reply
```
Use `RouterConfig.route_permissions` when permissions need to be configured
away from the handlers. For custom auth, override
`can_access_route(required_permissions)` instead of overriding
`route_turn`.
### Built-in routes
| Route | Handler | Purpose |

View File

@@ -317,6 +317,8 @@ RouterConfig(
route_descriptions={
"INTERNET_SEARCH": "이 라우트만 docstring 대신 사용할 설명.",
},
route_permissions={"INTERNET_SEARCH": "web_search"}, # 선택적 오버라이드
permission_denied_route="permission_denied", # 기본값
default_intent="converse", # LLM 호출 실패 또는 LLM 없음일 때 사용
fallback_intent="converse", # LLM이 잘못된 라우트를 반환할 때 사용
intent_field="intent",
@@ -351,6 +353,58 @@ Routes:
`RouterConfig.prompt`는 **도메인 프레이밍** (어시스턴트 페르소나, 비즈니스 규칙, 톤)을 위한 자리입니다. 라우트 카탈로그는 자동 생성되니 `prompt` 안에 라우트 목록을 넣지 마세요. 핸들러를 추가하는 순간 동기화가 깨집니다.
### 라우트 권한
router가 유효한 라우트를 선택한 뒤 보호된 라우트를 게이트하려면 `@listen(..., required_permissions=[...])`를 사용하세요. 기본 authorizer는 `self.state`의 `permissions` 컬렉션을 확인합니다. 거부된 턴은 `permission_denied`로 리다이렉트됩니다.
```python
from crewai import Flow
from crewai.experimental.conversational import (
ConversationConfig,
ConversationState,
RouterConfig,
)
from crewai.flow import listen, start
from pydantic import Field
class AppState(ConversationState):
permissions: set[str] = Field(default_factory=set)
@ConversationConfig(
router=RouterConfig(
prompt="웹 조회는 INTERNET_SEARCH로, 감사 요청은 ADMIN_REPORT로 보내세요.",
),
)
class SupportFlow(Flow[AppState]):
conversational = True
@start()
def load_permissions(self) -> None:
if self.state.session_ready:
return
self.state.permissions = {"web_search"}
self.state.session_ready = True
@listen("INTERNET_SEARCH", required_permissions=["web_search"])
def internet_search(self) -> str:
...
@listen("ADMIN_REPORT", required_permissions=["admin"])
def admin_report(self) -> str:
...
@listen("permission_denied")
def deny(self) -> str:
reply = "이 작업을 수행할 권한이 없습니다."
self.append_assistant_message(reply)
return reply
```
권한을 핸들러 밖에서 구성해야 한다면 `RouterConfig.route_permissions`를 사용하세요. 커스텀 인증에는 `route_turn`을 오버라이드하는 대신 `can_access_route(required_permissions)`를 오버라이드하세요.
### 빌트인 라우트
| 라우트 | 핸들러 | 목적 |

View File

@@ -318,6 +318,8 @@ RouterConfig(
route_descriptions={
"INTERNET_SEARCH": "Sobrescreve a docstring só desta rota.",
},
route_permissions={"INTERNET_SEARCH": "web_search"}, # override opcional
permission_denied_route="permission_denied", # padrão
default_intent="converse", # usado quando a chamada ao LLM falha ou não há LLM
fallback_intent="converse", # usado quando o LLM retorna rota inválida
intent_field="intent",
@@ -352,6 +354,58 @@ Routes:
`RouterConfig.prompt` é para **enquadramento de domínio** (persona do assistente, regras de negócio, voz). O catálogo de rotas é auto-gerado — não liste rotas em `prompt`; elas vão sair de sincronia assim que você adicionar um handler.
### Permissões de rota
Use `@listen(..., required_permissions=[...])` para proteger rotas depois que o router selecionar uma rota válida. O autorizador padrão verifica uma coleção `permissions` em `self.state`; turnos negados redirecionam para `permission_denied`.
```python
from crewai import Flow
from crewai.experimental.conversational import (
ConversationConfig,
ConversationState,
RouterConfig,
)
from crewai.flow import listen, start
from pydantic import Field
class AppState(ConversationState):
permissions: set[str] = Field(default_factory=set)
@ConversationConfig(
router=RouterConfig(
prompt="Envie buscas web para INTERNET_SEARCH e auditorias para ADMIN_REPORT.",
),
)
class SupportFlow(Flow[AppState]):
conversational = True
@start()
def load_permissions(self) -> None:
if self.state.session_ready:
return
self.state.permissions = {"web_search"}
self.state.session_ready = True
@listen("INTERNET_SEARCH", required_permissions=["web_search"])
def internet_search(self) -> str:
...
@listen("ADMIN_REPORT", required_permissions=["admin"])
def admin_report(self) -> str:
...
@listen("permission_denied")
def deny(self) -> str:
reply = "Você não tem permissão para essa ação."
self.append_assistant_message(reply)
return reply
```
Use `RouterConfig.route_permissions` quando as permissões precisarem ser configuradas fora dos handlers. Para autorização customizada, sobrescreva `can_access_route(required_permissions)` em vez de sobrescrever `route_turn`.
### Rotas embutidas
| Rota | Handler | Propósito |