CrewAI Run Automation Tool

`InvokeCrewAIAutomationTool`

InvokeCrewAIAutomationTool 提供了 CrewAI Platform API 与外部 crew 服务的集成。这个工具让你可以在 CrewAI agents 内部调用并与 CrewAI Platform automations 交互，从而在不同 crew 工作流之间实现无缝集成。

安装

uv pip install 'crewai[tools]'

要求

可访问 CrewAI Platform API
用于认证的有效 bearer token
可访问 CrewAI Platform automation endpoints 的网络

用法

下面展示如何在 CrewAI agent 中使用该工具：

from crewai import Agent, Task, Crew
from crewai_tools import InvokeCrewAIAutomationTool

# Initialize the tool
automation_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://data-analysis-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Data Analysis Crew",
    crew_description="Analyzes data and generates insights"
)

# Create a CrewAI agent that uses the tool
automation_coordinator = Agent(
    role='Automation Coordinator',
    goal='Coordinate and execute automated crew tasks',
    backstory='I am an expert at leveraging automation tools to execute complex workflows.',
    tools=[automation_tool],
    verbose=True
)

# Create a task for the agent
analysis_task = Task(
    description="Execute data analysis automation and provide insights",
    agent=automation_coordinator,
    expected_output="Comprehensive data analysis report"
)

# Create a crew with the agent
crew = Crew(
    agents=[automation_coordinator],
    tasks=[analysis_task],
    verbose=2
)

# Run the crew
result = crew.kickoff()
print(result)

工具参数

Argument	Type	Required	Default	Description
crew_api_url	`str`	Yes	None	CrewAI Platform automation API 的基础 URL
crew_bearer_token	`str`	Yes	None	API 认证用 bearer token
crew_name	`str`	Yes	None	crew automation 的名称
crew_description	`str`	Yes	None	该 crew automation 的作用描述
max_polling_time	`int`	No	600	等待任务完成的最长时间（秒）
crew_inputs	`dict`	No	None	定义自定义输入 schema 字段的字典

环境变量

CREWAI_API_URL=https://your-crew-automation.crewai.com  # Alternative to passing crew_api_url
CREWAI_BEARER_TOKEN=your_bearer_token_here              # Alternative to passing crew_bearer_token

高级用法

使用动态参数的自定义输入 schema

from crewai import Agent, Task, Crew
from crewai_tools import InvokeCrewAIAutomationTool
from pydantic import Field

# Define custom input schema
custom_inputs = {
    "year": Field(..., description="Year to retrieve the report for (integer)"),
    "region": Field(default="global", description="Geographic region for analysis"),
    "format": Field(default="summary", description="Report format (summary, detailed, raw)")
}

# Create tool with custom inputs
market_research_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://state-of-ai-report-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="State of AI Report",
    crew_description="Retrieves a comprehensive report on state of AI for a given year and region",
    crew_inputs=custom_inputs,
    max_polling_time=15 * 60  # 15 minutes timeout
)

# Create an agent with the tool
research_agent = Agent(
    role="Research Coordinator",
    goal="Coordinate and execute market research tasks",
    backstory="You are an expert at coordinating research tasks and leveraging automation tools.",
    tools=[market_research_tool],
    verbose=True
)

# Create and execute a task with custom parameters
research_task = Task(
    description="Conduct market research on AI tools market for 2024 in North America with detailed format",
    agent=research_agent,
    expected_output="Comprehensive market research report"
)

crew = Crew(
    agents=[research_agent],
    tasks=[research_task]
)

result = crew.kickoff()

多阶段自动化工作流

from crewai import Agent, Task, Crew, Process
from crewai_tools import InvokeCrewAIAutomationTool

# Initialize different automation tools
data_collection_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://data-collection-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Data Collection Automation",
    crew_description="Collects and preprocesses raw data"
)

analysis_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://analysis-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Analysis Automation",
    crew_description="Performs advanced data analysis and modeling"
)

reporting_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://reporting-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Reporting Automation",
    crew_description="Generates comprehensive reports and visualizations"
)

# Create specialized agents
data_collector = Agent(
    role='Data Collection Specialist',
    goal='Gather and preprocess data from various sources',
    backstory='I specialize in collecting and cleaning data from multiple sources.',
    tools=[data_collection_tool]
)

data_analyst = Agent(
    role='Data Analysis Expert',
    goal='Perform advanced analysis on collected data',
    backstory='I am an expert in statistical analysis and machine learning.',
    tools=[analysis_tool]
)

report_generator = Agent(
    role='Report Generation Specialist',
    goal='Create comprehensive reports and visualizations',
    backstory='I excel at creating clear, actionable reports from complex data.',
    tools=[reporting_tool]
)

# Create sequential tasks
collection_task = Task(
    description="Collect market data for Q4 2024 analysis",
    agent=data_collector
)

analysis_task = Task(
    description="Analyze collected data to identify trends and patterns",
    agent=data_analyst
)

reporting_task = Task(
    description="Generate executive summary report with key insights and recommendations",
    agent=report_generator
)

# Create a crew with sequential processing
crew = Crew(
    agents=[data_collector, data_analyst, report_generator],
    tasks=[collection_task, analysis_task, reporting_task],
    process=Process.sequential,
    verbose=2
)

result = crew.kickoff()

用例

分布式 crew 编排

协调多个专门的 crew automation 以处理复杂的多阶段工作流
让不同的 automation service 之间无缝交接，以更全面地执行任务
通过在多个 CrewAI Platform automations 之间分配工作负载来扩展处理能力

跨平台集成

将 CrewAI agents 与 CrewAI Platform automations 打通，支持混合本地-云工作流
在保持本地控制和编排的同时利用专门的 automations
让本地 agents 与基于云的 automation services 安全协作

企业自动化管道

创建企业级自动化管道，将本地智能与云端算力结合起来
实现跨多个 automation service 的复杂业务工作流
为数据分析、报表和决策提供可扩展、可重复的流程

动态工作流组合

根据任务需求动态串联不同的 automation service 来组合工作流
让自动化选择可以基于数据特征或业务规则自适应变化
创建灵活、可复用的 automation 组件，可按多种方式组合

专业领域处理

让通用 agents 可以访问特定领域的 automations（财务分析、法律研究、技术文档等）
利用预构建的专业 crew automations，而无需重建复杂领域逻辑
让 agents 通过有针对性的 automation services 获取专家级能力

自定义输入 schema

在定义 crew_inputs 时，使用 Pydantic Field 对象来指定输入参数：

from pydantic import Field

crew_inputs = {
    "required_param": Field(..., description="This parameter is required"),
    "optional_param": Field(default="default_value", description="This parameter is optional"),
    "typed_param": Field(..., description="Integer parameter", ge=1, le=100)  # With validation
}

错误处理

该工具对常见场景提供了完整的错误处理：

API Connection Errors：与 CrewAI Platform 的网络连接问题
Authentication Errors：无效或过期的 bearer token
Timeout Errors：超过最大轮询时间的任务
Task Failures：执行期间失败的 crew automations
Input Validation Errors：传递给 automation endpoints 的参数无效

API 端点

该工具与两个主要 API endpoint 交互：

POST {crew_api_url}/kickoff：启动一个新的 crew automation 任务
GET {crew_api_url}/status/{crew_id}：检查正在运行任务的状态

备注

工具会每秒自动轮询状态 endpoint，直到完成或超时
成功的任务会直接返回结果，而失败的任务会返回错误信息
bearer token 应妥善保管，不要在生产环境中硬编码
对 bearer token 等敏感配置，建议使用环境变量
自定义输入 schema 必须与目标 crew automation 期望的参数兼容