自定义工具有两条路，先看它们在源码里的样子（base_tool.py:496 的 Tool 就是 @tool 背后的类）：

# 路线A：@tool 装饰器（base_tool.py:676）——包一个已有函数
from crewai.tools import tool
@tool("查天气")
def get_weather(city: str) -> str:
    """查询某城市当前天气。输入城市名，返回温度和天气描述。"""
    return call_weather_api(city)

# 路线B：继承 BaseTool（base_tool.py:103）——写一个有状态的类
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
class WeatherSchema(BaseModel):
    city: str = Field(..., description="城市名，如'上海'")
class WeatherTool(BaseTool):
    name: str = "查天气"
    description: str = "查询某城市当前天气，返回温度和描述"
    args_schema: type[BaseModel] = WeatherSchema
    def _run(self, city: str) -> str:
        return call_weather_api(city)

源码自带的 ReadFileTool（read_file_tool.py:24）是路线B 的典范：

# read_file_tool.py:18
class ReadFileToolSchema(BaseModel):
    file_name: str = Field(..., description="The name of the input file to read")

# read_file_tool.py:24
class ReadFileTool(BaseTool):
    name: str = "read_file"
    description: str = (
        "Read content from an input file by name. "
        "Returns file content as text for text files, or base64 for binary files.")
    args_schema: type[BaseModel] = ReadFileToolSchema
    _files: dict[str, FileInput] | None = PrivateAttr(default=None)   # ★私有状态

    def set_files(self, files) -> None:      # 框架在 kickoff 时注入文件
        self._files = files

    def _run(self, file_name: str, **kwargs) -> str:
        if not self._files:
            return "No input files available."                   # ★① 无文件：说人话
        if file_name not in self._files:
            available = ", ".join(self._files.keys())
            return f"File '{file_name}' not found. Available files: {available}"  # ★② 找不到：给可选项
        file_input = self._files[file_name]
        content = file_input.read()
        content_type = file_input.content_type
        ...
        if any(content_type.startswith(t) for t in text_types):
            return content.decode("utf-8")                       # 文本：直接返回
        encoded = base64.b64encode(content).decode("ascii")
        return f"[Binary file: {filename} ({content_type})]\nBase64: {encoded}"  # 二进制：base64

AddImageTool（add_image_tool.py:16）展示工具也能返回非字符串：

# add_image_tool.py:9
class AddImageToolSchema(BaseModel):
    image_url: str = Field(..., description="The URL or path of the image to add")
    action: str | None = Field(default=None, description="Optional context or question about the image")

# add_image_tool.py:16
class AddImageTool(BaseTool):
    name: str = Field(default_factory=lambda: I18N_DEFAULT.tools("add_image")["name"])
    description: str = Field(default_factory=lambda: I18N_DEFAULT.tools("add_image")["description"])
    args_schema: type[BaseModel] = AddImageToolSchema

    def _run(self, image_url: str, action: str | None = None, **kwargs) -> dict:
        action = action or I18N_DEFAULT.tools("add_image")["default_action"]
        content = [
            {"type": "text", "text": action},
            {"type": "image_url", "image_url": {"url": image_url}},
        ]
        return {"role": "user", "content": content}     # ★返回一条"消息"结构，而非普通文本

回顾 Day 27 的 _generate_description（base_tool.py:482），它把你的 schema 拼进最终描述：

# base_tool.py:482
def _generate_description(self) -> None:
    schema = generate_model_description(self.args_schema)
    args_json = json.dumps(schema["json_schema"]["schema"], indent=2)
    self.description = (
        f"Tool Name: {sanitize_tool_name(self.name)}\n"
        f"Tool Arguments: {args_json}\n"       # 你 Field 里的 description 会出现在这
        f"Tool Description: {self.description}")

Day 27 见过的三个字段，这里讲实战怎么用（base_tool.py:176）：

# base_tool.py:176（回顾三个开关）
cache_function: SerializableCallable = Field(default=_default_cache_function, ...)
result_as_answer: bool = Field(default=False, ...)
max_usage_count: int | None = Field(default=None, ...)

# 实战：一个"生成最终报告"的工具，结果就是交付物
@tool(result_as_answer=True)          # ★一返回就当最终答案，Agent 不再多想
def write_report(findings: str) -> str:
    """把调研发现整理成最终报告"""
    return format_report(findings)

# 实战：一个昂贵的付费搜索，限制最多用 3 次
@tool(max_usage_count=3)              # ★超 3 次返回"用到上限"，逼模型收敛
def premium_search(query: str) -> str:
    """付费深度搜索（调用受限）"""
    return paid_api(query)

# 实战：实时数据工具，关掉缓存
weather_tool.cache_function = lambda args, result: False   # ★永不缓存，避免过期

已有 langchain 工具不用重写，from_langchain（base_tool.py:410）能转进来：

# base_tool.py:410
@classmethod
def from_langchain(cls, tool: Any) -> BaseTool:
    if not hasattr(tool, "func") or not callable(tool.func):
        raise ValueError("The provided tool must have a callable 'func' attribute.")
    args_schema = getattr(tool, "args_schema", None)
    result_schema = getattr(tool, "result_schema", None)
    if result_schema is None:
        result_schema = _infer_result_schema_from_callable(tool.func)
    if args_schema is None:                      # 没 schema → 从 func 签名推断
        func_signature = signature(tool.func)
        fields = {}
        for name, param in func_signature.parameters.items():
            ...                                  # 同 Day 27 的签名推断套路
        args_schema = create_model(f"{sanitize_tool_name(tool.name)}_input", **fields)
    return cls(name=getattr(tool, "name", "Unnamed Tool"),
               description=getattr(tool, "description", ""),
               func=tool.func, args_schema=args_schema, result_schema=result_schema)

反向，把一批工具统一成执行态用 to_langchain（base_tool.py:641）：

# base_tool.py:641
def to_langchain(tools: list[BaseTool | CrewStructuredTool]) -> list[CrewStructuredTool]:
    return [t.to_structured_tool() if isinstance(t, BaseTool) else t for t in tools]