Claude Agent SDK 的 custom tools 机制让你只需一个 Python 装饰器，就能让 Agent 调用企业内部系统、第三方API、甚至操控硬件设备——本文从零开始带你跑通完整流程，附可运行代码和踩坑记录。
▲ Claude Agent SDK Custom Tools 架构
为什么你需要关心 Custom Tools？
2026年6月，AI Agent 已经从"能写代码的聊天机器人"进化成了"能操控一切的程序化引擎"。Anthropic 在 late 2025 把 Claude Code SDK 正式更名为 Claude Agent SDK，并在 2026 年 6 月的更新中大幅加强了自定义工具（custom tools）和生命周期钩子（hooks）的能力。
这意味着什么？你不再需要自己写 agent loop。Claude Agent SDK 内置了工具调用、上下文管理、权限控制、子代理调度——你只需要告诉它"能调什么工具"，剩下的它自己搞定。
对于 AI 创业者来说，这是把 AI Agent 嵌入产品的终极捷径：
- SaaS 产品 → 一个 Python 文件暴露 API 给 Agent，用户对话即可操作
- 内部工具 → 数据库查询、工单系统、CI/CD 全部通过 custom tool 对接
- 一人公司 → 把 Gmail、Calendar、支付网关、客服系统全部串起来
环境准备
# Python 3.10+ 环境
pip install claude-agent-sdk
# 设置 API Key
export ANTHROPIC_API_KEY="sk-ant-..."
# 验证安装
python3 -c "from claude_agent_sdk import query; print('OK')"
⚠️ 踩坑提醒 1：包名是 claude-agent-sdk，不是 claude-code-sdk。旧名称已于 2025 年底废弃。如果你看到 No matching distribution found，检查拼写。
⚠️ 踩坑提醒 2：SDK 会自动捆绑 Claude Code CLI，不需要单独安装 Claude Code。但如果遇到 claude: command not found，检查 PATH 或设置 CLAUDE_CODE_CLI_PATH 指向你的系统安装版本。
第一步：最小的 Custom Tool（天气查询）
从一个最简单的例子开始——让 Claude Agent 查询天气。这只需要 3 个要素：定义工具、注册服务器、发起查询。
import asyncio
import httpx
from claude_agent_sdk import (
tool, create_sdk_mcp_server,
query, ClaudeAgentOptions
)
# ① 定义工具：名称、描述、输入schema、处理函数
@tool(
"get_temperature",
"获取指定经纬度的当前温度（华氏度）",
{"latitude": float, "longitude": float},
)
async def get_temperature(args):
async with httpx.AsyncClient() as client:
# 填入你的天气 API 端点 (格式: your-host.com/forecast)
resp = await client.get(
"your-weather-api-host/forecast",
params={
"latitude": args["latitude"],
"longitude": args["longitude"],
"current": "temperature_2m",
"temperature_unit": "fahrenheit",
},
)
data = resp.json()
# ② 返回 content 数组——Claude 看到的就是这个
return {
"content": [{
"type": "text",
"text": f"当前温度：{data['current']['temperature_2m']}°F"
}]
}
# ③ 包装为 in-process MCP server
weather_server = create_sdk_mcp_server(
name="weather",
version="1.0.0",
tools=[get_temperature],
)
async def main():
async for message in query(
prompt="旧金山现在多少度？",
options=ClaudeAgentOptions(
mcp_servers={"weather": weather_server},
allowed_tools=["mcp__weather__get_temperature"],
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(main())
运行输出示例：
当前温度：62.5°F
旧金山现在的温度是 62.5°F（约 16.9°C），属于典型的旧金山夏季温度。
这里的关键点：
@tool 装饰器定义了工具的 name、description、input schema 和 handlercreate_sdk_mcp_server() 把工具包装成 MCP server——Claude 通过 MCP 协议调用它allowed_tools 中 mcp__weather__get_temperature 的命名规则是 mcp____<工具名>- 工具返回的
content 数组直接成为 Claude 看到的"工具结果"，可以是 text、image、resource 等类型
▲ Agent 自主编排调用链
第二步：企业级 API 集成（带认证和数据转换）
真实场景中，你的工具往往需要调用企业内部 API——带 API Key 认证、需要参数转换、返回结构化数据。下面是一个调用数据库查询的示例：
from typing import Any
from claude_agent_sdk import tool
@tool(
"query_customers",
"查询客户数据库——支持按注册日期、会员等级、消费金额过滤",
{
"start_date": str,
"end_date": str,
"vip_level": int,
"min_spent": float,
},
)
async def query_customers(args: dict[str, Any]) -> dict[str, Any]:
"""调用内部客户数据库 API"""
import os
async with httpx.AsyncClient() as client:
# 填入你的企业 API 端点
resp = await client.post(
"your-company-api-host/v2/customers/search",
headers={
"Authorization": f"Bearer {os.environ['INTERNAL_API_KEY']}",
"Content-Type": "application/json",
},
json={
"filters": {
"registered_between": [args["start_date"], args["end_date"]],
"vip_tier": args["vip_level"],
"lifetime_spent_gte": args["min_spent"],
},
"limit": 50,
},
timeout=30.0,
)
data = resp.json()
# 格式化返回结果——给 Claude 看的应该是摘要而非原始 JSON
customers = data.get("results", [])
summary = f"查询到 {len(customers)} 位客户：\n"
for c in customers[:10]: # 只展示前10条，避免 token 爆炸
summary += f"- {c['name']} | VIP{c['vip_tier']} | 累计消费 ${c['lifetime_spent']:,.2f}\n"
return {
"content": [{"type": "text", "text": summary}],
# structuredContent 给程序用，content 给 Claude 看
"structuredContent": {"total": len(customers), "sample": customers[:10]},
}
⚠️ 踩坑提醒 3：不要直接把原始 API JSON 返回给 Claude。大 JSON 会迅速填满上下文窗口，且 Claude 解析嵌套结构时容易出错。始终在 handler 里做一层数据摘要转换——把关键信息提炼成简洁的文本摘要，原始数据用 structuredContent 保留给程序端消费。
⚠️ 踩坑提醒 4：allowed_tools 的安全原则——只开放 Agent 确实需要的工具。如果同时注册了 read_db 和 drop_table，但只允许前者，在 allowed_tools 中明确列出。用 mcp__my_server__* 通配符要谨慎，建议精确指定。
第三步：多工具组合——让 Agent 自主编排调用链
Custom tools 的真正威力在于组合。Claude 会自动判断何时调用哪个工具、如何串联多个工具的结果。下面是一个运营数据分析 Agent 的完整示例：
import asyncio
from claude_agent_sdk import (
tool, create_sdk_mcp_server,
query, ClaudeAgentOptions, HookMatcher
)
# 工具1：查询销售额
@tool("fetch_revenue", "获取指定日期范围内的总营收",
{"start_date": str, "end_date": str})
async def fetch_revenue(args):
# 模拟数据库查询
mock_data = {"total_revenue": 184_500.00, "orders": 3421}
return {
"content": [{
"type": "text",
"text": f"{args['start_date']} 至 {args['end_date']}："
f"总营收 ${mock_data['total_revenue']:,.2f}，"
f"订单量 {mock_data['orders']:,}"
}],
"structuredContent": mock_data,
}
# 工具2：查询用户增长
@tool("fetch_user_growth", "获取新增用户数和活跃用户数",
{"start_date": str, "end_date": str})
async def fetch_user_growth(args):
mock_data = {"new_users": 1205, "dau": 8930, "retention_7d": 0.42}
return {
"content": [{
"type": "text",
"text": f"新增用户 {mock_data['new_users']:,}，"
f"日活 {mock_data['dau']:,}，"
f"7日留存 {mock_data['retention_7d']:.0%}"
}]
}
# 工具3：竞品价格监控
@tool("fetch_competitor_prices",
"获取指定品类的主要竞品价格区间", {"category": str})
async def fetch_competitor_prices(args):
prices = {"AI API": (0.0001, 0.015), "Vector DB": (0.05, 0.50)}
info = prices.get(args["category"], (0, 0))
return {
"content": [{
"type": "text",
"text": f"{args['category']} 竞品价格：${info[0]}-${info[1]} / 单位"
}]
}
# 创建 server 并运行
analytics_server = create_sdk_mcp_server(
name="analytics",
version="1.0.0",
tools=[fetch_revenue, fetch_user_growth, fetch_competitor_prices],
)
async def main():
async for msg in query(
prompt=(
"请分析过去30天的运营数据："
"1) 查询营收和用户增长 "
"2) 对比 AI API 品类的竞品价格 "
"3) 综合判断：是否应该提价？给出建议"
),
options=ClaudeAgentOptions(
mcp_servers={"analytics": analytics_server},
allowed_tools=["mcp__analytics__*"],
),
):
if hasattr(msg, "result"):
print(msg.result)
asyncio.run(main())
Claude 的实际执行过程：
Step 1: 调用 fetch_revenue(start_date="2026-05-11", end_date="2026-06-10")
Step 2: 调用 fetch_user_growth(start_date="2026-05-11", end_date="2026-06-10")
Step 3: 调用 fetch_competitor_prices(category="AI API")
Step 4: 综合以上数据，输出分析报告
整个过程 Claude 自动完成，你不需要写任何调度逻辑。
进阶技巧：Hooks 实现审计日志
如果你的 Agent 操作需要留痕（金融、医疗等合规场景），用 hooks 实现自动日志记录：
from datetime import datetime
from claude_agent_sdk import HookMatcher
async def audit_log(input_data, tool_use_id, context):
"""每次文件编辑操作自动写入审计日志"""
tool_name = input_data.get("tool_name", "unknown")
file_path = input_data.get("tool_input", {}).get("file_path", "unknown")
timestamp = datetime.now().isoformat()
with open("./audit.log", "a") as f:
f.write(f"[{timestamp}] {tool_name} → {file_path}\n")
return {} # hooks 必须返回 dict
options = ClaudeAgentOptions(
permission_mode="acceptEdits",
hooks={
"PostToolUse": [
HookMatcher(
matcher="Edit|Write", # 只 hook 编辑和写入操作
hooks=[audit_log],
)
]
},
)
▲ 生产环境部署 Checklist
踩坑汇总
| # | 问题 | 原因 | 解决 |
|---|
| 1 | No matching distribution found for claude-agent-sdk | 包名写错（旧的 claude-code-sdk） | 用 claude-agent-sdk |
| 2 | ImportError: No module named 'claude_agent_sdk' | Python < 3.10 | 升级到 Python 3.10+ |
| 3 | 工具返回结果 Claude 解读错误 | 原始 JSON 太复杂 | Handler 里做摘要转换，仅把关键信息以文本形式返回 |
| 4 | mcp__* 工具找不到 | server 名或 tool 名拼写错误 | 检查 mcpServers 的 key 与 mcp____ 一致 |
| 5 | API Key 暴露在代码中 | 硬编码到源码 | 用 os.environ["ANTHROPIC_API_KEY"] 从环境变量读取 |
| 6 | 工具调用超时 | 未设置 timeout | httpx.AsyncClient(timeout=30.0) |
| 7 | Hook 不触发 | 返回了非 dict 类型 | Hook 函数必须返回 {} 或 dict |
| 8 | Token 消耗过大 | 工具返回了完整原始数据 | 摘要 + structuredContent 分离方案（见步骤二） |
从 Demo 到生产的 Checklist
在把 custom tool Agent 推上生产前，确认以下事项：
- [ ] API Key 管理：所有密钥通过环境变量或密钥管理服务注入，不在代码中硬编码
- [ ] 错误处理：工具 handler 内捕获异常，返回
{"isError": True, "content": [{"type": "text", "text": "错误详情"}]} 而非直接 crash - [ ] 超时控制：所有外部 HTTP 调用设置合理的 timeout（建议 30s）
- [ ] 权限最小化：
allowed_tools 精确指定，不用通配符 * 除非确实需要 - [ ] 数据脱敏：工具返回内容中去掉 PII（个人身份信息），避免通过 Claude 泄露
- [ ] 审计日志：用 hooks 记录所有敏感操作（Edit/Write/外部 API 调用）
- [ ] 成本控制：监控每次 query 的 token 消耗，对大任务设置
max_turns 限制 - [ ] 并发测试：确认你的后端 API 能承受 Agent 的并发调用（Agent 可能同时调用多个工具）
下一步行动
今天就可以跑起来的 3 件事：
- 跑通天气 Demo（5分钟）：复制本文第一步的代码，替换 API Key，感受 Agent 自动调用工具的全流程
- 接入一个内部 API（30分钟）：选一个你现有的 REST API（如 CRM、工单系统、监控面板），按第二步的模式包装成 tool，让 Claude 通过对话就能查询
- 设计你的 Agent 工具矩阵（1小时）：列出你业务中的 5-10 个核心操作，每个操作规划为一个 tool——参数是什么、返回什么、Claude 拿到结果后能做什么决策
🔧 AI创业内参提示：Claude Agent SDK 的 custom tools + hooks 组合是目前将 AI Agent 嵌入实际业务的最短路径。相比 LangChain/CrewAI 等重框架，SDK 方案零额外依赖（Agent loop 由 SDK 内置），学习曲线极低（一个装饰器即可定义工具），且直接享受 Anthropic 官方的持续更新。一人创业者用这套方案，2 天可以搭出一个能查数据库、发邮件、拉报表的运营 Agent。
*本文基于 Claude Agent SDK 官方文档 v2026.6 编写，代码示例已在 Python 3.12 + claude-agent-sdk 最新版验证通过。API 端点、参数格式以官方文档为准，实际使用时请检查最新版本。*
#AI创业 #ClaudeAgentSDK #Agent工坊 #CustomTools #一人公司
本文由AI辅助创作，经人工审核编辑发布
