第 4 课：知识库与自定义工具
为什么需要自定义工具？
上一课的 AI 只能"空口分类"——它没有你公司的任何数据。
但真正的客服需要： - 查知识库找答案 - 查工单数据库看历史 - 查订单系统核实信息 - 更新工单状态
这些都需要自定义工具。Claude Agent SDK 用 MCP（Model Context Protocol）让你给 AI 接上自己的数据库和 API。
MCP 工具的核心概念
AI 想查知识库
│
├── AI: "我需要调用 search_faq 工具"
│
├── SDK: 调用你写的 search_faq 函数
│
├── 你的函数: 查数据库，返回结果
│
└── AI: 拿到结果，组织回复给客户
AI 自己决定"什么时候"用工具，你负责写"工具能干什么"。
第一个工具：知识库检索
先准备一个简单的 FAQ 知识库（data/faq.json）：
[
{
"id": "faq-001",
"question": "如何重置密码",
"answer": "进入登录页面，点击"忘记密码"，输入注册邮箱，我们会发送重置链接到您的邮箱。链接有效期 24 小时。",
"category": "account",
"keywords": ["密码", "重置", "忘记", "登录不了"]
},
{
"id": "faq-002",
"question": "如何升级到企业版",
"answer": "登录后台，进入"设置-套餐管理"，选择企业版，按提示完成支付。企业版支持年付（8折优惠）和月付。如需定制方案，请联系销售团队。",
"category": "billing",
"keywords": ["企业版", "升级", "套餐", "价格"]
},
{
"id": "faq-003",
"question": "如何导出数据",
"answer": "进入"数据管理"页面，选择要导出的数据范围和格式（CSV/Excel），点击"导出"。大量数据导出可能需要几分钟，导出完成后会发邮件通知。",
"category": "feature",
"keywords": ["导出", "下载", "数据", "Excel", "CSV"]
},
{
"id": "faq-004",
"question": "退款政策",
"answer": "购买后 7 天内可无理由全额退款。超过 7 天但未满 30 天，可退还未使用月份的费用。请联系客服提交退款申请。",
"category": "billing",
"keywords": ["退款", "退钱", "取消", "不想用了"]
},
{
"id": "faq-005",
"question": "API 调用限制",
"answer": "免费版：100次/天，基础版：10000次/天，企业版：无限制。超出限制会返回 429 错误。可在后台查看当前用量。",
"category": "technical",
"keywords": ["API", "限制", "429", "频率", "调用次数"]
}
]
现在写检索工具：
"""知识库检索工具"""
import json
from pathlib import Path
from claude_agent_sdk import tool, create_sdk_mcp_server
# 加载知识库
FAQ_DATA = json.loads(Path("data/faq.json").read_text(encoding="utf-8"))
@tool(
"search_faq",
"在知识库中搜索与客户问题相关的 FAQ 条目",
{"keyword": {"type": "string", "description": "搜索关键词"}}
)
async def search_faq(keyword: str) -> str:
"""关键词搜索 FAQ"""
results = []
keyword_lower = keyword.lower()
for faq in FAQ_DATA:
# 在问题、答案、关键词中搜索
searchable = (
faq["question"] + faq["answer"] + " ".join(faq["keywords"])
).lower()
if keyword_lower in searchable:
results.append(faq)
if not results:
return f"没有找到与 '{keyword}' 相关的 FAQ。"
output = f"找到 {len(results)} 条相关 FAQ：\n\n"
for faq in results:
output += f"📋 {faq['question']}\n"
output += f" {faq['answer']}\n\n"
return output
@tool 装饰器详解
@tool(
"search_faq", # 工具名（AI 看到的名字）
"在知识库中搜索相关的 FAQ 条目", # 工具描述（AI 根据这个决定要不要用）
{ # 参数定义
"keyword": {
"type": "string",
"description": "搜索关键词",
}
}
)
async def search_faq(keyword: str) -> str:
# 函数体：实际的搜索逻辑
...
AI 会看到：
- 名字：search_faq
- 描述：搜索 FAQ
- 参数：需要一个 keyword
然后 AI 自己决定什么时候调用这个工具。
第二个工具：工单数据库
"""工单数据库工具"""
import json
import datetime
from claude_agent_sdk import tool
# 模拟工单数据库
TICKET_DB: dict[str, dict] = {
"TK-001": {
"id": "TK-001",
"customer": "张三",
"email": "zhangsan@example.com",
"subject": "登录页面一直转圈",
"status": "open",
"priority": "high",
"created_at": "2025-03-15 10:30:00",
"history": ["客户提交工单", "已分配给技术团队"],
},
"TK-002": {
"id": "TK-002",
"customer": "李四",
"email": "lisi@example.com",
"subject": "企业版咨询",
"status": "pending",
"priority": "low",
"created_at": "2025-03-15 11:00:00",
"history": ["客户提交工单"],
},
}
@tool(
"get_ticket",
"根据工单号查询工单详情",
{"ticket_id": {"type": "string", "description": "工单号，如 TK-001"}}
)
async def get_ticket(ticket_id: str) -> str:
"""查询工单"""
ticket = TICKET_DB.get(ticket_id.upper())
if not ticket:
return f"未找到工单 {ticket_id}"
return json.dumps(ticket, ensure_ascii=False, indent=2)
@tool(
"update_ticket_status",
"更新工单状态（open/pending/resolved/closed）",
{
"ticket_id": {"type": "string", "description": "工单号"},
"status": {"type": "string", "description": "新状态: open/pending/resolved/closed"},
"note": {"type": "string", "description": "操作备注"},
}
)
async def update_ticket_status(ticket_id: str, status: str, note: str = "") -> str:
"""更新工单状态"""
ticket = TICKET_DB.get(ticket_id.upper())
if not ticket:
return f"未找到工单 {ticket_id}"
valid_statuses = ["open", "pending", "resolved", "closed"]
if status not in valid_statuses:
return f"无效状态: {status}，可选: {valid_statuses}"
old_status = ticket["status"]
ticket["status"] = status
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
ticket["history"].append(f"[{timestamp}] {old_status} → {status}: {note}")
return f"✅ 工单 {ticket_id} 状态已更新: {old_status} → {status}"
@tool(
"search_tickets_by_customer",
"根据客户名字或邮箱搜索工单",
{"query": {"type": "string", "description": "客户名字或邮箱"}}
)
async def search_tickets_by_customer(query: str) -> str:
"""搜索客户的工单"""
results = []
query_lower = query.lower()
for ticket in TICKET_DB.values():
if (query_lower in ticket["customer"].lower()
or query_lower in ticket["email"].lower()):
results.append(ticket)
if not results:
return f"未找到客户 '{query}' 的工单"
output = f"找到 {len(results)} 个工单：\n"
for t in results:
output += f" {t['id']}: {t['subject']} [{t['status']}]\n"
return output
组装工具：创建 MCP 服务器
from claude_agent_sdk import create_sdk_mcp_server
# 创建 MCP 服务器，把所有工具注册进去
support_tools = create_sdk_mcp_server(
name="support-tools",
version="1.0.0",
tools=[
search_faq,
get_ticket,
update_ticket_status,
search_tickets_by_customer,
],
)
完整示例：带工具的智能客服
import anyio
from claude_agent_sdk import (
query, ClaudeAgentOptions,
AssistantMessage, ResultMessage, TextBlock, ToolUseBlock,
tool, create_sdk_mcp_server,
)
# === 工具定义（简化版）===
FAQ_DATA = [
{"question": "如何重置密码", "answer": "进入登录页，点忘记密码，输入邮箱即可。", "keywords": ["密码", "重置"]},
{"question": "退款政策", "answer": "7天内无理由退款，超7天退未使用部分。", "keywords": ["退款", "退钱"]},
{"question": "API 限制", "answer": "免费100次/天，基础1万/天，企业无限。", "keywords": ["API", "限制", "429"]},
]
@tool("search_faq", "搜索常见问题知识库", {"keyword": {"type": "string", "description": "关键词"}})
async def search_faq(keyword: str) -> str:
results = [f for f in FAQ_DATA if any(k in keyword for k in f["keywords"])]
if not results:
return f"知识库中没有找到 '{keyword}' 相关内容。"
return "\n".join(f"Q: {f['question']}\nA: {f['answer']}" for f in results)
TICKETS = {"TK-001": {"id": "TK-001", "customer": "张三", "subject": "登录问题", "status": "open"}}
@tool("get_ticket", "查询工单详情", {"ticket_id": {"type": "string", "description": "工单号"}})
async def get_ticket(ticket_id: str) -> str:
import json
t = TICKETS.get(ticket_id.upper())
return json.dumps(t, ensure_ascii=False) if t else f"未找到 {ticket_id}"
# === 创建 MCP 服务器 ===
support_server = create_sdk_mcp_server(
name="support",
version="1.0.0",
tools=[search_faq, get_ticket],
)
# === 主程序 ===
SYSTEM_PROMPT = """你是智能客服助手。你有以下工具可以使用：
1. search_faq - 搜索知识库回答常见问题
2. get_ticket - 查询工单详情
工作流程：
1. 理解客户问题
2. 如果是常见问题，先搜知识库
3. 如果涉及特定工单，查询工单详情
4. 给客户专业、友好的回复
用中文回复。"""
async def handle_customer(question: str):
options = ClaudeAgentOptions(
system_prompt=SYSTEM_PROMPT,
mcp_servers={"support": support_server},
allowed_tools=[
"mcp__support__search_faq",
"mcp__support__get_ticket",
],
max_turns=5,
)
print(f"\n客户: {question}")
async for msg in query(prompt=question, options=options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"客服: {block.text}")
elif isinstance(block, ToolUseBlock):
print(f" 🔧 [查询 {block.name}]")
elif isinstance(msg, ResultMessage):
if msg.total_cost_usd and msg.total_cost_usd > 0:
print(f" 💰 ${msg.total_cost_usd:.4f}")
async def main():
await handle_customer("我的密码忘了怎么办？")
print("-" * 40)
await handle_customer("帮我查一下工单 TK-001 是什么情况？")
print("-" * 40)
await handle_customer("你们的 API 有调用次数限制吗？")
anyio.run(main)
运行后你会看到 AI 自动决定什么时候搜知识库、什么时候查工单：
客户: 我的密码忘了怎么办？
🔧 [查询 search_faq]
客服: 您好！重置密码很简单：进入登录页，点击"忘记密码"...
客户: 帮我查一下工单 TK-001 是什么情况？
🔧 [查询 get_ticket]
客服: 我查到了您的工单 TK-001，当前状态是"处理中"...
工具命名规则
MCP 工具在 allowed_tools 中的命名规则：
mcp__{服务器名}__{工具名}
比如：
- 服务器名: support，工具名: search_faq → mcp__support__search_faq
- 服务器名: orders，工具名: get_order → mcp__orders__get_order
工具设计最佳实践
1. 描述要写给 AI 看
# ❌ 描述太技术化
@tool("search_faq", "执行 FAQ 表的全文检索查询", ...)
# ✅ 描述像在和 AI 说话
@tool("search_faq", "当客户问常见问题时，在知识库中搜索答案", ...)
2. 返回值要信息丰富
# ❌ 返回太少
return "找到了"
# ✅ 返回有用信息
return f"找到 3 条相关 FAQ：\n1. {faq1}\n2. {faq2}\n3. {faq3}"
3. 错误信息要友好
# ❌ 抛异常
raise ValueError("Not found")
# ✅ 返回友好错误
return "未找到该工单，请确认工单号是否正确（格式：TK-XXXX）"
本课小结
- MCP 自定义工具让 AI 能访问你的数据和系统
@tool装饰器定义工具，create_sdk_mcp_server注册- AI 自动决定什么时候用什么工具
- 工具描述要写给 AI 看，返回值要信息丰富
课后练习
- 给知识库加 5 条新的 FAQ 条目
- 写一个
create_ticket工具，让 AI 能创建新工单 - 写一个
get_order工具，查询订单信息（模拟数据即可）