CrewAI Run 自动化工具

使 CrewAI 智能体能够调用 CrewAI Platform 自动化，并在你的工作流中利用外部团队服务。

`InvokeCrewAIAutomationTool`

InvokeCrewAIAutomationTool 提供与外部团队服务之间的 CrewAI Platform API 集成。该工具让你可以在 CrewAI 智能体内调用 CrewAI Platform 自动化并与之交互，从而实现不同团队工作流之间的无缝集成。

安装

bash

uv pip install 'crewai[tools]'

要求

CrewAI Platform API 访问权限
用于身份认证的有效 bearer token
对 CrewAI Platform 自动化端点的网络访问权限

用量

以下是如何在 CrewAI 智能体中使用该工具：

python

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)

工具参数

参数	类型	必填	默认值	描述
crew_api_url	`str`	是	None	CrewAI Platform 自动化 API 的基础 URL
crew_bearer_token	`str`	是	None	用于 API 身份认证的 bearer token
crew_name	`str`	是	None	团队自动化的名称
crew_description	`str`	是	None	团队自动化功能的描述
max_polling_time	`int`	否	600	等待任务完成的最长时间（秒）
crew_inputs	`dict`	否	None	定义自定义输入模式字段的字典

环境变量

bash

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

高级用法

带动态参数的自定义输入模式

python

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()

多阶段自动化工作流

python

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()

使用场景

分布式团队编排

协调多个专用团队自动化，处理复杂的多阶段工作流
在不同自动化服务之间实现无缝交接，以完成全面的任务执行
通过将工作负载分配到多个 CrewAI Platform 自动化来扩展处理能力

跨平台集成

将 CrewAI 智能体与 CrewAI Platform 自动化连接起来，构建本地与云端混合的工作流
在保持本地控制与编排的同时，利用专用自动化
支持本地智能体与云端自动化服务之间的安全协作

企业级自动化流水线

创建结合本地智能与云端处理能力的企业级自动化流水线
实现跨越多个自动化服务的复杂业务工作流
为数据分析、报告和决策提供可扩展、可重复的流程

动态工作流组合

根据任务需求，通过串联不同自动化服务来动态组合工作流
实现自适应处理，让自动化的选择取决于数据特征或业务规则
创建可灵活组合、可复用的自动化组件

专用领域处理

通过通用智能体访问领域专用自动化（财务分析、法律研究、技术文档）
无需重建复杂的领域逻辑，即可利用预构建的专用团队自动化
使智能体能够通过有针对性的自动化服务获得专家级能力

自定义输入模式

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

python

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 连接错误：与 CrewAI Platform 的网络连接问题
身份认证错误：无效或已过期的 bearer token
超时错误：超出最大轮询时间的任务
任务失败：在执行过程中失败的团队自动化
输入校验错误：传递给自动化端点的无效参数

API 端点

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

POST {crew_api_url}/kickoff：启动新的团队自动化任务
GET {crew_api_url}/status/{crew_id}：检查正在运行任务的状态

说明

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

CrewAI Run 自动化工具

InvokeCrewAIAutomationTool

安装

要求

用量

工具参数

环境变量

高级用法

带动态参数的自定义输入模式

多阶段自动化工作流

使用场景

分布式团队编排

跨平台集成

企业级自动化流水线

动态工作流组合

专用领域处理

自定义输入模式

错误处理

API 端点

说明

友情链接

`InvokeCrewAIAutomationTool`