跳转到内容

触发器概览

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

Automation Triggers Overview

以下深入指南会逐个介绍各个集成的设置和示例工作流:

使用触发器,你可以:

  • 响应实时事件 - 当特定条件满足时自动执行工作流
  • 与外部系统集成 - 连接 Gmail、Outlook、OneDrive、JIRA、Slack、Stripe 等平台
  • 扩展自动化规模 - 无需人工介入即可处理高吞吐量事件
  • 保留上下文 - 在 crews 和 flows 中访问触发器数据

要访问并管理你的自动化触发器:

  1. 在 CrewAI 仪表板中导航到你的部署
  2. 点击 Triggers 选项卡查看所有可用的触发器集成
可用自动化触发器列表
Gmail 部署可用自动化触发器示例

此视图会显示该部署可用的所有触发器集成及其当前连接状态。

每个触发器都可以通过切换开关轻松启用或禁用:

通过切换开关启用或禁用触发器
通过切换开关启用或禁用触发器
  • 已启用(蓝色开关):触发器处于活动状态,当指定事件发生时会自动执行你的部署
  • 已禁用(灰色开关):触发器处于非活动状态,不会响应事件

只需点击开关即可更改触发器状态。更改会立即生效。

跟踪触发执行的性能和历史记录:

由自动化触发的执行列表
由自动化触发的执行列表

在构建自动化之前,先了解 crews 和 flows 将接收到的触发 payload 结构会很有帮助。

在将触发器接入生产环境之前,请确保你:

  • Tools & Integrations 下连接集成,并完成任何 OAuth 或 API key 步骤
  • 在应响应事件的部署上启用触发器开关
  • 提供所需的环境变量(API token、tenant ID、共享密钥)
  • 创建或更新任务,使其能够在第一个 crew 任务或 flow 步骤中解析传入 payload
  • 决定是否通过 allow_crewai_trigger_context 自动传递触发上下文
  • 设置监控 - webhook 日志、CrewAI 执行历史以及可选的外部告警

CrewAI CLI 提供了强大的命令,帮助你在不部署到生产环境的情况下开发和测试触发驱动的自动化。

查看已连接集成的所有可用触发器:

Terminal window
crewai triggers list

该命令会根据你已连接的集成显示所有可用触发器,包括:

  • 集成名称和连接状态
  • 可用的触发器类型
  • 触发器名称和描述

在部署前使用真实感 payload 测试你的 crew:

Terminal window
crewai triggers run <trigger_name>

例如:

Terminal window
crewai triggers run microsoft_onedrive/file_changed

该命令会:

  • 在本地执行你的 crew
  • 传递完整且真实的触发 payload
  • 精确模拟 crew 在生产环境中的调用方式

你现有的 crew 定义可以无缝与触发器配合,你只需要有一个任务来解析收到的 payload:

@CrewBase
class 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,你可以更灵活地处理触发数据:

你 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

当在一个被触发的 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 部署转变为响应式、事件驱动的系统,并与现有业务流程和工具无缝集成。