类的骨架（structured_tool.py:115）：

# structured_tool.py:115
class CrewStructuredTool(BaseModel):
    name: str = Field(default="")
    description: str = Field(default="")
    args_schema: ... = Field(default=None)       # 参数 schema（带序列化器）
    result_schema: ... = Field(default=None)     # 输出 schema
    func: Any = Field(default=None, exclude=True) # ★被包裹的真正函数
    result_as_answer: bool = Field(default=False)
    max_usage_count: int | None = Field(default=None)
    current_usage_count: int = Field(default=0)
    cache_function: Any = Field(default=None, exclude=True)
    _original_tool: Any = PrivateAttr(default=None)  # 指回原 BaseTool

    def invoke(self, input, config=None, **kwargs) -> Any: ...   # :340 主入口
    async def ainvoke(self, input, config=None, **kwargs): ...   # :296 异步入口
    def _parse_args(self, raw_args) -> dict: ...                 # :272 解析参数

回看 Day 27 的 to_structured_tool（base_tool.py:393）——就是这座桥：

# base_tool.py:393
def to_structured_tool(self) -> CrewStructuredTool:
    self._set_args_schema()                 # 确保 args_schema 就绪
    structured_tool = CrewStructuredTool(
        name=self.name, description=self.description,
        args_schema=self.args_schema, result_schema=self.result_schema,
        func=self._run,                      # ★把 _run 作为要调的函数
        result_as_answer=self.result_as_answer,
        max_usage_count=self.max_usage_count,
        current_usage_count=self.current_usage_count,
        cache_function=self.cache_function)
    structured_tool._original_tool = self    # ★留一根指针指回自己
    return structured_tool

除了从 BaseTool 转，也能直接从一个函数造（structured_tool.py:151）：

# structured_tool.py:151
@classmethod
def from_function(cls, func, name=None, description=None,
                  return_direct=False, args_schema=None,
                  result_schema=None, infer_schema=True, **kwargs):
    name = name or func.__name__                    # 没名字 → 用函数名
    description = description or inspect.getdoc(func) # 没描述 → 用 docstring
    if description is None:
        raise ValueError(f"Function {name} must have a docstring ...")
    description = textwrap.dedent(description).strip()

    if args_schema is not None:
        schema = args_schema
    elif infer_schema:
        schema = cls._create_schema_from_function(name, func)  # ★从签名推断
    else:
        raise ValueError("Either args_schema must be provided or infer_schema must be True.")

    return cls(name=name, description=description, args_schema=schema,
               result_schema=result_schema or _infer_result_schema_from_callable(func),
               func=func, result_as_answer=return_direct, **kwargs)

推断 schema 的 _create_schema_from_function（structured_tool.py:217）和 Day 27 思路一致，但用 get_type_hints：

# structured_tool.py:217
@staticmethod
def _create_schema_from_function(name, func):
    sig = inspect.signature(func)
    type_hints = get_type_hints(func)         # 解析"字符串形式的类型标注"
    fields = {}
    for param_name, param in sig.parameters.items():
        if param_name in ("self", "cls"):
            continue
        annotation = type_hints.get(param_name, Any)
        default = ... if param.default == param.empty else param.default
        fields[param_name] = (annotation, Field(default=default))
    schema_name = f"{name.title()}Schema"
    return create_model(schema_name, **fields)

_parse_args（structured_tool.py:272）就干这个：

# structured_tool.py:272
def _parse_args(self, raw_args: str | dict) -> dict:
    if isinstance(raw_args, str):
        try:
            raw_args = json.loads(raw_args)          # ① 字符串 → dict
        except json.JSONDecodeError as e:
            raise ValueError(f"Failed to parse arguments as JSON: {e}") from e
    if not self.args_schema:
        return raw_args if isinstance(raw_args, dict) else {}
    try:
        validated_args = self.args_schema.model_validate(raw_args)  # ② schema 校验
        return dict(validated_args.model_dump())                    # ③ 返回清洗后的 dict
    except Exception as e:
        hint = build_schema_hint(self.args_schema)   # ④ 校验失败 → 附上期望参数
        raise ValueError(f"Arguments validation failed: {e}{hint}") from e

同步主入口 invoke（structured_tool.py:340）：

# structured_tool.py:340
def invoke(self, input, config=None, **kwargs) -> Any:
    parsed_args = self._parse_args(input)            # ① 解析参数

    if self.has_reached_max_usage_count():           # ② 查用量上限
        raise ToolUsageLimitExceededError(
            f"Tool '...' has reached its maximum usage limit of {self.max_usage_count}. ...")
    self._increment_usage_count()                    # ③ 计数 +1

    if inspect.iscoroutinefunction(self.func):       # ④ func 本身是 async？
        return asyncio.run(self.func(**parsed_args, **kwargs))
    result = self.func(**parsed_args, **kwargs)      # ⑤ 普通函数：直接调
    if asyncio.iscoroutine(result):                  # ⑥ 返回了协程？跑完它
        return asyncio.run(result)
    return result

用量检查 has_reached_max_usage_count 与计数同步（structured_tool.py:366）：

# structured_tool.py:366
def has_reached_max_usage_count(self) -> bool:
    return (self.max_usage_count is not None
            and self.current_usage_count >= self.max_usage_count)

def _increment_usage_count(self) -> None:
    self.current_usage_count += 1
    if self._original_tool is not None:
        self._original_tool.current_usage_count = self.current_usage_count  # ★同步回原工具

异步入口 ainvoke（structured_tool.py:296）：

# structured_tool.py:296
async def ainvoke(self, input, config=None, **kwargs) -> Any:
    parsed_args = self._parse_args(input)
    if self.has_reached_max_usage_count():
        raise ToolUsageLimitExceededError(...)
    self._increment_usage_count()
    try:
        if inspect.iscoroutinefunction(self.func):
            return await self.func(**parsed_args, **kwargs)     # ① async 函数 → 直接 await
        import asyncio
        return await asyncio.get_event_loop().run_in_executor(  # ② 同步函数 → 丢线程池
            None, lambda: self.func(**parsed_args, **kwargs))
    except Exception:
        raise

工具返回的可能是对象、字典、任意东西，但喂给 LLM 的必须是字符串。_format_tool_output_for_agent（structured_tool.py:54）负责这一步：

# structured_tool.py:54
def _format_tool_output_for_agent(tool, raw_result):
    original_tool = getattr(tool, "_original_tool", None)
    if original_tool is not None:
        return original_tool.format_output_for_agent(raw_result)  # ① 有原工具 → 用它的逻辑

    result_schema = getattr(tool, "result_schema", None)
    if not (isinstance(result_schema, type) and issubclass(result_schema, BaseModel)):
        return str(raw_result)                    # ② 没 result_schema → 直接 str()

    try:
        validation_input = raw_result
        if isinstance(raw_result, BaseModel) and not isinstance(raw_result, result_schema):
            validation_input = raw_result.model_dump()
        validated = result_schema.model_validate(validation_input)
        return validated.model_dump_json()        # ③ 有 schema → 校验后转 JSON
    except Exception as exc:
        warnings.warn(f"Failed to validate ... Falling back to str(raw_result).", ...)
        return str(raw_result)                    # ④ 校验失败 → 优雅降级为 str()