跳转到内容

发布自定义工具

CrewAI 的工具系统被设计为可扩展的。如果你构建了一个能让他人受益的工具,你可以把它打包成独立的 Python 库,发布到 PyPI,并让任何 CrewAI 用户都能使用 - 无需向 CrewAI 仓库提交 PR。

本指南会完整介绍整个流程:实现 tools 协议、组织包结构,以及发布到 PyPI。

每个 CrewAI 工具都必须满足以下两种接口之一:

继承 crewai.tools.BaseTool 并实现 _run 方法。定义 namedescription,并可选地定义用于输入验证的 args_schema

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
class GeolocateInput(BaseModel):
"""GeolocateTool 的输入模式。"""
address: str = Field(..., description="要进行地理定位的街道地址。")
class GeolocateTool(BaseTool):
name: str = "Geolocate"
description: str = "将街道地址转换为经纬度坐标。"
args_schema: type[BaseModel] = GeolocateInput
def _run(self, address: str) -> str:
# 你的实现
return f"40.7128, -74.0060"

对于更简单的工具,@tool 装饰器可以把函数转换成 CrewAI 工具。函数必须包含 docstring(作为工具描述使用)以及类型标注。

from crewai.tools import tool
@tool("Geolocate")
def geolocate(address: str) -> str:
"""将街道地址转换为经纬度坐标。"""
return "40.7128, -74.0060"

无论采用哪种方式,你的工具都必须:

  • 具有 name - 一个简短、具有描述性的标识符。
  • 具有 description - 告诉 agent 何时以及如何使用该工具。这会直接影响 agent 使用工具的效果,因此要清晰具体。
  • 实现 _run(BaseTool)或提供 函数体(@tool) - 同步执行逻辑。
  • 为所有参数和返回值使用 类型标注
  • 返回一个 字符串 结果,或定义一个可选的 Pydantic 输出模式来表示结构化结果。

如果你的工具执行的是 I/O 密集型工作,请实现 _arun 以支持异步执行:

class GeolocateTool(BaseTool):
name: str = "Geolocate"
description: str = "将街道地址转换为经纬度坐标。"
def _run(self, address: str) -> str:
# 同步实现
...
async def _arun(self, address: str) -> str:
# 异步实现
...

可选:使用 args_schema 进行输入验证

Section titled “可选:使用 args_schema 进行输入验证”

将 Pydantic 模型定义为你的 args_schema,即可获得自动输入验证和更清晰的错误信息。如果你不提供,CrewAI 会根据 _run 方法签名推断它。

from pydantic import BaseModel, Field
class TranslateInput(BaseModel):
"""TranslateTool 的输入模式。"""
text: str = Field(..., description="要翻译的文本。")
target_language: str = Field(
default="en",
description="目标语言的 ISO 639-1 语言代码。",
)

对于已发布的工具,推荐显式提供 schema - 它们会带来更好的 agent 行为和更清晰的用户文档。

可选:使用 result_schema 定义类型化输出

Section titled “可选:使用 result_schema 定义类型化输出”

如果你的工具返回结构化数据,请定义一个 Pydantic 输出模型。对已发布的工具来说,这是个不错的默认值,因为用户和 agent 都可以依赖命名字段。

直接的 Python 调用仍然会收到工具返回的值。当 agent 使用该工具时,CrewAI 会根据输出模型把 JSON 发送给 agent。

CrewAI 可以从 Pydantic 返回注解中推断输出模式:

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
class GeolocateResult(BaseModel):
latitude: float = Field(..., description="十进制度纬度。")
longitude: float = Field(..., description="十进制度经度。")
class GeolocateTool(BaseTool):
name: str = "Geolocate"
description: str = "将街道地址转换为经纬度坐标。"
def _run(self, address: str) -> GeolocateResult:
if "1600 Pennsylvania" in address:
return GeolocateResult(latitude=38.8977, longitude=-77.0365)
return GeolocateResult(latitude=40.7128, longitude=-74.0060)

当工具返回字典时,显式设置 result_schema

class GeolocateTool(BaseTool):
name: str = "Geolocate"
description: str = "将街道地址转换为经纬度坐标。"
result_schema: type[BaseModel] = GeolocateResult
def _run(self, address: str) -> dict[str, float]:
if "1600 Pennsylvania" in address:
return {"latitude": 38.8977, "longitude": -77.0365}
return {"latitude": 40.7128, "longitude": -74.0060}

如果你希望 agent 接收到的是简短的文本摘要而不是 JSON,请在 BaseTool 子类上覆盖 format_output_for_agent

class GeolocateTool(BaseTool):
name: str = "Geolocate"
description: str = "将街道地址转换为经纬度坐标。"
def _run(self, address: str) -> GeolocateResult:
if "1600 Pennsylvania" in address:
return GeolocateResult(latitude=38.8977, longitude=-77.0365)
return GeolocateResult(latitude=40.7128, longitude=-74.0060)
def format_output_for_agent(self, raw_result: object) -> str:
result = GeolocateResult.model_validate(raw_result)
return f"Latitude {result.latitude}, longitude {result.longitude}"

这个覆盖只会改变 agent 看到的内容。你包的直接使用者仍然会从 tool.run(...) 获得正常值。

如果你的工具需要 API key 或其他配置,请用 env_vars 声明出来,让用户知道需要设置什么:

from crewai.tools import BaseTool, EnvVar
class GeolocateTool(BaseTool):
name: str = "Geolocate"
description: str = "将街道地址转换为经纬度坐标。"
env_vars: list[EnvVar] = [
EnvVar(
name="GEOCODING_API_KEY",
description="地理编码服务的 API key。",
required=True,
),
]
def _run(self, address: str) -> str:
...

把项目组织成标准 Python 包。推荐布局如下:

crewai-geolocate/
├── pyproject.toml
├── LICENSE
├── README.md
└── src/
└── crewai_geolocate/
├── __init__.py
└── tools.py
[project]
name = "crewai-geolocate"
version = "0.1.0"
description = "A CrewAI tool for geolocating street addresses."
requires-python = ">=3.10"
dependencies = [
"crewai",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

crewai 声明为依赖,这样用户就会自动获得兼容版本。

重新导出你的工具类,让用户可以直接导入它们:

from crewai_geolocate.tools import GeolocateTool
__all__ = ["GeolocateTool"]
  • 包名:使用 crewai- 前缀(例如 crewai-geolocate)。这样用户在 PyPI 上搜索时更容易找到你的工具。
  • 模块名:使用下划线(例如 crewai_geolocate)。
  • 工具类名:使用以 Tool 结尾的 PascalCase(例如 GeolocateTool)。

在发布前,先验证你的工具能在 crew 中正常工作:

from crewai import Agent, Crew, Task
from crewai_geolocate import GeolocateTool
agent = Agent(
role="位置分析师",
goal="为给定地址查找坐标。",
backstory="地理空间数据专家。",
tools=[GeolocateTool()],
)
task = Task(
description="查找华盛顿特区 Pennsylvania Avenue 1600 号的坐标。",
expected_output="该地址的纬度和经度。",
agent=agent,
)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()
print(result)

当你的工具测试完成并准备就绪后:

Terminal window
# 构建包
uv build
# 发布到 PyPI
uv publish

如果这是你第一次发布,你需要一个 PyPI 账户 和一个 API token

用户可以用下面的命令安装你的工具:

Terminal window
pip install crewai-geolocate

或者使用 uv:

Terminal window
uv add crewai-geolocate

然后在他们的 crews 中使用它:

from crewai_geolocate import GeolocateTool
agent = Agent(
role="位置分析师",
tools=[GeolocateTool()],
# ...
)