触发器概览
CrewAI AMP 触发器可将你的自动化与团队已在使用的工具中的实时事件连接起来。它们不会轮询系统,也不依赖手动 kickoff,而是监听变化,例如新邮件、日历更新、CRM 状态变化,并立即启动你指定的 crew 或 flow。

集成操作指南
Section titled “集成操作指南”以下深入指南会逐个介绍各个集成的设置和示例工作流:
Gmail 触发器
Google Calendar 触发器
Google Drive 触发器
Outlook 触发器
OneDrive 触发器
Microsoft Teams 触发器
HubSpot 触发器
Salesforce 触发器
Slack 触发器
Zapier 触发器
使用触发器,你可以:
- 响应实时事件 - 当特定条件满足时自动执行工作流
- 与外部系统集成 - 连接 Gmail、Outlook、OneDrive、JIRA、Slack、Stripe 等平台
- 扩展自动化规模 - 无需人工介入即可处理高吞吐量事件
- 保留上下文 - 在 crews 和 flows 中访问触发器数据
查看可用触发器
Section titled “查看可用触发器”要访问并管理你的自动化触发器:
- 在 CrewAI 仪表板中导航到你的部署
- 点击 Triggers 选项卡查看所有可用的触发器集成
此视图会显示该部署可用的所有触发器集成及其当前连接状态。
启用和禁用触发器
Section titled “启用和禁用触发器”每个触发器都可以通过切换开关轻松启用或禁用:
- 已启用(蓝色开关):触发器处于活动状态,当指定事件发生时会自动执行你的部署
- 已禁用(灰色开关):触发器处于非活动状态,不会响应事件
只需点击开关即可更改触发器状态。更改会立即生效。
监控触发执行
Section titled “监控触发执行”跟踪触发执行的性能和历史记录:
构建触发驱动的自动化
Section titled “构建触发驱动的自动化”在构建自动化之前,先了解 crews 和 flows 将接收到的触发 payload 结构会很有帮助。
触发设置清单
Section titled “触发设置清单”在将触发器接入生产环境之前,请确保你:
- 在 Tools & Integrations 下连接集成,并完成任何 OAuth 或 API key 步骤
- 在应响应事件的部署上启用触发器开关
- 提供所需的环境变量(API token、tenant ID、共享密钥)
- 创建或更新任务,使其能够在第一个 crew 任务或 flow 步骤中解析传入 payload
- 决定是否通过
allow_crewai_trigger_context自动传递触发上下文 - 设置监控 - webhook 日志、CrewAI 执行历史以及可选的外部告警
使用 CLI 在本地测试触发器
Section titled “使用 CLI 在本地测试触发器”CrewAI CLI 提供了强大的命令,帮助你在不部署到生产环境的情况下开发和测试触发驱动的自动化。
列出可用触发器
Section titled “列出可用触发器”查看已连接集成的所有可用触发器:
crewai triggers list该命令会根据你已连接的集成显示所有可用触发器,包括:
- 集成名称和连接状态
- 可用的触发器类型
- 触发器名称和描述
模拟触发执行
Section titled “模拟触发执行”在部署前使用真实感 payload 测试你的 crew:
crewai triggers run <trigger_name>例如:
crewai triggers run microsoft_onedrive/file_changed该命令会:
- 在本地执行你的 crew
- 传递完整且真实的触发 payload
- 精确模拟 crew 在生产环境中的调用方式
在 Crew 中使用触发器
Section titled “在 Crew 中使用触发器”你现有的 crew 定义可以无缝与触发器配合,你只需要有一个任务来解析收到的 payload:
@CrewBaseclass MyAutomatedCrew: @agent def researcher(self) -> Agent: return Agent( config=self.agents_config['researcher'], )
@task def parse_trigger_payload(self) -> Task: return Task( config=self.tasks_config['parse_trigger_payload'], agent=self.researcher(), )
@task def analyze_trigger_content(self) -> Task: return Task( config=self.tasks_config['analyze_trigger_data'], agent=self.researcher(), )crew 会自动接收并可通过标准 CrewAI 上下文机制访问触发 payload。
与 Flows 集成
Section titled “与 Flows 集成”对于 flows,你可以更灵活地处理触发数据:
访问触发 payload
Section titled “访问触发 payload”你 flow 中的所有 @start() 方法都会接受一个额外参数 crewai_trigger_payload:
from crewai.flow import Flow, start, listen
class MyAutomatedFlow(Flow): @start() def handle_trigger(self, crewai_trigger_payload: dict = None): """ This start method can receive trigger data """ if crewai_trigger_payload: # Process the trigger data trigger_id = crewai_trigger_payload.get('id') event_data = crewai_trigger_payload.get('payload', {})
# Store in flow state for use by other methods self.state.trigger_id = trigger_id self.state.trigger_type = event_data
return event_data
# Handle manual execution return None
@listen(handle_trigger) def process_data(self, trigger_data): """ Process the data from the trigger """ # ... process the trigger从 Flows 触发 Crews
Section titled “从 Flows 触发 Crews”当在一个被触发的 flow 中启动 crew 时,将触发 payload 一并传递:
@start()def delegate_to_crew(self, crewai_trigger_payload: dict = None): """ Delegate processing to a specialized crew """ crew = MySpecializedCrew()
# Pass the trigger payload to the crew result = crew.crew().kickoff( inputs={ 'a_custom_parameter': "custom_value", 'crewai_trigger_payload': crewai_trigger_payload }, )
return result触发器未触发:
- 确认触发器在部署的 Triggers 选项卡中已启用
- 检查 Tools & Integrations 下的集成连接状态
- 确保所有必需的环境变量都已正确配置
执行失败:
- 查看执行日志以获取错误详情
- 使用
crewai triggers run <trigger_name>在本地测试并查看准确的 payload 结构 - 验证你的 crew 能处理
crewai_trigger_payload参数 - 确保你的 crew 不会期望触发 payload 中未包含的参数
开发问题:
- 在部署前始终先用
crewai triggers run <trigger>进行测试,以查看完整 payload - 记住
crewai run不会模拟触发调用 - 请改用crewai triggers run - 使用
crewai triggers list查看已连接集成可用的触发器 - 部署后,crew 会接收真实触发 payload,因此请先在本地充分测试
自动化触发器会将你的 CrewAI 部署转变为响应式、事件驱动的系统,并与现有业务流程和工具无缝集成。