注册自定义工具
内置工具能覆盖常见编程任务,但真实产品里往往还有 Jira、内部 API、专有数据库——这些都要靠自定义工具接进来。把业务函数注册给模型后,AI 就能在对话中按需调用;再配合 before_tool / after_tool,你可以在调用前后做审计、鉴权与输出脱敏。
本系列演示模型统一为 通义千问 qwen3.5-plus。请已按第 01 篇配置环境变量(CODY_MODEL、CODY_MODEL_API_KEY、CODY_MODEL_BASE_URL)。下面示例使用 Cody().workdir("...").build(),不显式写模型名即可继承环境配置。
若不想依赖环境变量,可在 Builder 上显式指定 Qwen 端点:
client = (
Cody()
.workdir(".")
.model("qwen3.5-plus")
.base_url("https://coding.dashscope.aliyuncs.com/v1")
.api_key("sk-xxx")
.build()
)
工具函数:签名里藏着契约
Cody 通过 PydanticAI 暴露工具,因此函数形态需要满足框架约定:必须是 async,第一个参数固定为 ctx: RunContext[CodyDeps],返回值类型为 str(模型读到的就是这段字符串)。其余参数由模型根据对话填入,类型注解会参与 schema 生成。
| 要求 | 说明 |
|---|---|
async def | 工具在异步上下文中执行,便于 await IO。 |
首参 ctx | RunContext[CodyDeps],注入运行期依赖(工作目录、配置等)。 |
-> str | 统一返回字符串,作为工具结果写回对话。 |
| docstring | 人类与模型共读的「说明书」,直接影响工具是否被选中、参数怎么填。 |
完整示例:先跑通一条链路
下面用虚构的 fetch_jira_ticket 演示端到端流程:定义工具 → Cody().tool(...).build() → run()。先把可运行骨架立起来,再逐段拆开讲。
import asyncio from pydantic_ai import RunContext from cody.core.deps import CodyDeps from cody.sdk import Cody async def fetch_jira_ticket(ctx: RunContext[CodyDeps], ticket_key: str) -> str: """根据 Jira 工单号拉取摘要(演示用:未接真实 API 时返回占位文本)。 参数: ticket_key: 工单号,例如 PROJ-123。 适用场景: 用户提到具体工单、需要状态/标题/经办人等信息时调用。 """ # 演示:生产环境在此调用 Jira REST API return f"Ticket {ticket_key}: Open · 修复登录超时 · assignee=alice" async def main() -> None: client = ( Cody() .workdir(".") .tool(fetch_jira_ticket) .build() ) async with client: result = await client.run("PROJ-456 这个单子现在什么情况?") print(result.output) if __name__ == "__main__": asyncio.run(main())
模型若判断需要查工单,会生成对 fetch_jira_ticket 的调用;你的函数执行后返回的字符串会进入后续推理。工作目录仍由 workdir 约束,与内置工具一致。
注册多个工具
链式多次 .tool() 即可。顺序只影响 Builder 可读性,不影响运行时行为。
client = (
Cody()
.workdir(".")
.tool(fetch_jira_ticket)
.tool(another_tool)
.build()
)
docstring:模型的「选型说明书」
docstring 是模型决定是否调用、如何传参的主要依据之一。若只写一行模糊描述,模型容易误选工具或填错参数。请写清:做什么、每个参数含义、何时该用、返回的大致形态(仍通过 str 返回即可)。
before_tool:放行、改参或拒绝
签名:async (tool_name: str, args: dict) -> dict | None。返回 dict 表示继续执行——可以返回修改后的 args;返回 None 表示拒绝本次调用(例如未通过安全策略)。
async def before(tool_name: str, args: dict) -> dict | None: print(f"调用 {tool_name}") return args # 放行;若需改参可返回 {"字段": 新值, ...} client = Cody().workdir(".").before_tool(before).build()
典型用途:结构化日志、限流、对敏感参数做校验。下面示例组合「打印调用信息」与「禁止调用危险工具名」两种逻辑(拆成两个函数更符合单一职责,下一节会链式注册)。
async def log_tool(tool_name: str, args: dict) -> dict | None: print(f"[audit] {tool_name} args={args}") return args async def security_check(tool_name: str, args: dict) -> dict | None: if tool_name == "dangerous_admin_action": return None # 拒绝调用 return args
after_tool:改写返回给模型的文本
签名:async (tool_name: str, args: dict, result: str) -> str。常用于日志归档、格式统一,或对 result 脱敏后再交给模型,避免密钥、token 进入后续上下文。
async def after(tool_name: str, args: dict, result: str) -> str: return result.replace("secret", "***") client = Cody().workdir(".").after_tool(after).build()
脱敏规则尽量可测试:对固定关键字、正则或结构化字段做替换,并注意大小写与多种拼写变体。
中间件链:按注册顺序执行
多个 before_tool 会按注册顺序依次执行;前一个返回的 dict 会作为下一个的输入 args(若某步返回 None,调用中止)。after_tool 同样按顺序包裹结果,像洋葱从外到内再返回。
async def redact_output(tool_name: str, args: dict, result: str) -> str: return result.replace("password=", "password=***") client = ( Cody() .before_tool(log_tool) .before_tool(security_check) .after_tool(redact_output) .build() )
redact_output 与上文 after 同属 after_tool 形态,可按业务替换为更复杂的正则或结构化脱敏,签名保持与 SDK 一致即可。
小结
自定义工具 = async + ctx + -> str + 高质量 docstring,用 Cody().tool(...).build() 挂到客户端。before_tool 管准入与参数,after_tool 管出站内容。下一篇 06. Prompt 定制与多模态 将在此基础上讲解如何定制系统提示与多模态输入,让模型更稳地配合你的工具集。