第 10 章:实战 — 工具生态
前两章我们造了 MiniClaw 的"骨架":认证、配置、引擎、记忆、命令行界面。 但说白了,它现在就是一个"能聊天的终端"——你说一句,它回一句,仅此而已。
这一章,我们给 MiniClaw 装技能。让它不只是聊天,还能做事。
前提条件
- 已完成前面章节的学习,了解
@tool、create_sdk_mcp_server的用法- 已安装 SDK:
uv add claude-agent-sdk
1. 本章目标
想象一下你手机刚买来的时候——只能打电话发短信。装上微信就能聊天,装上高德就能导航, 装上支付宝就能付款。工具就是 MiniClaw 的"App Store"。
本章结束后,MiniClaw 将拥有 5 个内置技能:
| 技能 | 说明 | 举例 |
|---|---|---|
current_time |
查看当前时间 | "现在几点了?" |
system_info |
查看系统信息 | "我的电脑配置怎么样?" |
calculator |
做数学计算 | "帮我算 sin(45) + log(100)" |
note_save |
保存笔记 | "帮我记一下:明天下午开会" |
note_list |
查看所有笔记 | "我之前记了哪些笔记?" |
更重要的是,我们会建一套工具注册机制——以后想加新技能, 写个函数、注册一下就完事了,不用改引擎代码。
2. 整体架构
加完工具后,MiniClaw 的文件结构:
miniclaw/
__init__.py # 包初始化
__main__.py # 入口:python -m miniclaw
auth.py # 认证管理
config.py # 配置管理
engine.py # 对话引擎(本章更新)
memory.py # 对话记忆
cli.py # 命令行界面
tools/ # 新增:工具系统
__init__.py # 工具包入口
registry.py # 工具注册中心
builtin.py # 5 个内置工具
工作流程一目了然:
启动 MiniClaw
↓
创建 ToolRegistry(注册中心)
↓
register_builtin_tools() ← 把 5 个内置工具注册进去
↓
registry.create_server() ← 打包成 SDK MCP 服务器
↓
注入 engine.py ← 引擎带着工具启动
↓
Claude 自动使用工具 ← 用户问"现在几点",Claude 调用 current_time
3. 工具注册中心 (tools/registry.py)
为什么需要注册中心?
你可能会想:我直接在 engine.py 里写死工具列表不行吗?
行,但不好。原因有三:
- 解耦 — 引擎不应该知道具体有哪些工具,它只需要知道"有工具可用"
- 可扩展 — 未来加新工具,只要往注册中心注册就行,不用改引擎
- 白名单管理 — 注册中心自动生成
allowed_tools,不用手动拼字符串
核心代码
完整代码见
miniclaw/tools/registry.py
ToolRegistry 只有 4 个方法,每个都很直白:
class ToolRegistry:
def __init__(self, server_name: str = "miniclaw"):
self.server_name = server_name
self._tools: list[SdkMcpTool] = []
def register(self, tool_def: SdkMcpTool) -> None:
"""注册一个工具。"""
self._tools.append(tool_def)
def create_server(self) -> McpSdkServerConfig:
"""打包成 SDK MCP 服务器。"""
return create_sdk_mcp_server(name=self.server_name, tools=self._tools)
def get_allowed_tools(self) -> list[str]:
"""获取工具白名单,格式为 mcp__<服务器名>__<工具名>。"""
return [f"mcp__{self.server_name}__{t.name}" for t in self._tools]
说白了就是一个列表的封装。它的价值不在于代码复杂度,而在于建立了一个清晰的边界:
工具的定义、注册、打包都在 tools/ 包里搞定,引擎只管用。
4. 内置工具集 (tools/builtin.py)
完整代码见
miniclaw/tools/builtin.py
5 个工具,逐个讲解核心设计思路。
工具 1:current_time — 查看当前时间
@tool(
"current_time",
"获取当前日期和时间,包含年月日、时分秒、星期几",
{"timezone": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
async def current_time(args: dict) -> dict:
...
设计要点:
- readOnlyHint=True:查时间是纯读操作,不改任何东西
- 参数 timezone 是字符串,Claude 会根据用户的问法自动传合适的值
- 用标准库 timedelta 做时区偏移,不依赖 pytz
工具 2:system_info — 查看系统信息
@tool(
"system_info",
"获取当前系统信息,包括操作系统、CPU、内存、磁盘使用情况",
{"detail_level": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
设计要点:
- 同样 readOnlyHint=True——只读不写
- detail_level 参数让 Claude 选择 "basic" 或 "full" 模式
- 只用 platform、os、shutil 标准库,零额外依赖
工具 3:calculator — 数学计算器
@tool(
"calculator",
"计算数学表达式。支持加减乘除、幂运算、三角函数、对数等",
{"expression": str},
)
设计要点:
- 没标注 readOnlyHint——有意不加,让你对比"有注解"和"没注解"的区别
- 安全措施:eval 的 __builtins__ 设为空字典,只暴露白名单里的数学函数。
即使表达式里写了 os.system("rm -rf /"),也会报错而不是真的执行
- 用 try/except 兜住所有异常,不让工具崩掉
工具 4:note_save — 保存笔记
@tool(
"note_save",
"保存一条笔记到本地。笔记会存储在 ~/.miniclaw/notes/ 目录下",
{"title": str, "content": str},
)
设计要点:
- 这个工具会写文件,所以没标 readOnlyHint
- 笔记存到 ~/.miniclaw/notes/,自动建目录
- 文件名用"时间戳 + 标题",不会覆盖旧笔记
- 标题做了特殊字符清理,避免文件名问题
工具 5:note_list — 列出所有笔记
@tool(
"note_list",
"列出所有已保存的笔记。可以按关键词筛选",
{"keyword": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
设计要点:
- readOnlyHint=True——只读不写
- keyword 参数支持按关键词筛选
- 笔记按时间倒序排列(最新的在前面)
ToolAnnotations 小结
5 个工具中,3 个标了 readOnlyHint=True,2 个没标。规律很简单:
| 工具 | 是否只读 | 有无注解 |
|---|---|---|
current_time |
只读 | readOnlyHint=True |
system_info |
只读 | readOnlyHint=True |
calculator |
只读(但我们故意不标) | 无 |
note_save |
会写文件 | 无 |
note_list |
只读 | readOnlyHint=True |
建议: 只读工具都标上 readOnlyHint=True,这是个好习惯。
注册函数
一个 register_builtin_tools() 函数,一键把 5 个工具注册进去:
def register_builtin_tools(registry: ToolRegistry) -> None:
registry.register(current_time)
registry.register(system_info)
registry.register(calculator)
registry.register(note_save)
registry.register(note_list)
调用方只需要两行:
registry = ToolRegistry()
register_builtin_tools(registry)
# 5 个工具就绑好了
5. 集成到引擎
engine.py 的变化
之前创建 ClaudeAgentOptions 时没有工具,现在加两行:
options = ClaudeAgentOptions(
system_prompt=system_prompt,
permission_mode="bypassPermissions",
max_turns=10,
# ---- 新增 ----
mcp_servers=mcp_servers, # 工具服务器
allowed_tools=allowed_tools, # 工具白名单
)
引擎通过构造函数接收这两个参数,完全不需要知道具体有哪些工具。
main.py 的变化
入口文件负责"组装":
from miniclaw.tools import ToolRegistry, register_builtin_tools
registry = ToolRegistry()
register_builtin_tools(registry)
engine = ChatEngine(
config=config,
memory=memory,
mcp_servers={"miniclaw": registry.create_server()},
allowed_tools=registry.get_allowed_tools(),
)
注意 mcp_servers 的 key 必须和 ToolRegistry 的 server_name 一致(默认都是 "miniclaw")。
不一致的话 allowed_tools 里的名字就对不上,工具就用不了。
6. 用户自定义工具(扩展机制)
想给 MiniClaw 加新技能?比如查天气。两步搞定:
Step 1:写工具函数
# miniclaw/tools/weather.py
@tool("get_weather", "查询指定城市的天气信息", {"city": str},
annotations=ToolAnnotations(readOnlyHint=True))
async def get_weather(args: dict) -> dict:
city = args["city"]
# 这里调用真正的天气 API
return {"content": [{"type": "text", "text": f"{city}:晴,25°C"}]}
Step 2:注册一下
from miniclaw.tools.weather import get_weather
registry.register(get_weather) # 加这一行
不用改引擎,不用改 CLI,不用改任何其他文件。
get_allowed_tools() 自动包含新工具。这就是注册中心的价值。
7. 运行效果
python -m miniclaw
查时间:
你: 现在几点了?
MiniClaw: 当前时间是 2026-02-25 14:32:18,星期三,时区 Asia/Shanghai (UTC+8)。
查系统:
你: 看看我电脑的配置
MiniClaw: 操作系统 Darwin 25.3.0,架构 arm64,Python 3.12.1,CPU 10 核心。
算数学:
你: 帮我算 sin(pi/4) + log10(1000)
MiniClaw: sin(pi/4) + log10(1000) = 0.7071 + 3 = 3.7071
记笔记 + 查笔记:
你: 帮我记一下,明天下午三点和老王开会
MiniClaw: 笔记已保存到 ~/.miniclaw/notes/20260225_143518_明天下午三点和老王开会.md
你: 我之前记了哪些笔记?
MiniClaw: 共 3 条笔记:
- 明天下午三点和老王开会
- 买菜清单
- 项目 TODO
组合使用(最酷的部分):
你: 现在几点了?顺便帮我记一下,明天要交报告
MiniClaw: 现在是 14:35:22。已帮你记下"明天要交报告",保存到
~/.miniclaw/notes/20260225_143522_明天要交报告.md。
Claude 自动先调 current_time,再调 note_save,把结果整合成一句话。
你不需要告诉它"先查时间再记笔记"——它自己判断该用什么工具、什么顺序。
8. 小结
这一章做了三件事:
- 建了工具注册中心 —
ToolRegistry统一管理注册、打包、白名单 - 写了 5 个内置工具 — 时间、系统信息、计算器、保存笔记、查看笔记
- 集成到引擎 — 在
ClaudeAgentOptions中注入mcp_servers和allowed_tools
核心收获:
@tool装饰器把 Python 函数变成 Claude 可调用的技能ToolAnnotations给工具标上"只读"等标签,帮助 Claude 安全决策ToolRegistry让扩展新能力的成本极低——写个函数、注册一下、完事了- Claude 自动判断用什么工具、什么顺序,你不需要手动编排
下一章,我们将为 MiniClaw 添加钩子系统——在 Claude 调用工具前后插入安全检查。
完整代码
以下是本章所有文件的完整可运行代码。在 miniclaw/ 目录的上级目录执行 python -m miniclaw 即可运行。
miniclaw/__init__.py
"""MiniClaw - 一个用 Claude Agent SDK 构建的个人助手。"""
__version__ = "0.3.0"
miniclaw/auth.py
"""MiniClaw 认证模块。
检查用户是否已配置 Claude 的认证信息。
支持两种方式:
1. 环境变量 ANTHROPIC_API_KEY
2. Claude CLI 的 OAuth 登录状态
"""
import os
import subprocess
def ensure_authenticated() -> bool:
"""检查是否已认证。
优先检查环境变量,其次检查 CLI 登录状态。
返回 True 表示认证通过。
"""
# 方式 1:检查 API Key 环境变量
if os.environ.get("ANTHROPIC_API_KEY"):
return True
# 方式 2:检查 Claude CLI 是否已登录
try:
result = subprocess.run(
["claude", "--version"],
capture_output=True,
text=True,
timeout=5,
)
return result.returncode == 0
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
miniclaw/config.py
"""MiniClaw 配置模块。
管理所有可配置项,支持从配置文件和环境变量加载。
"""
import json
from dataclasses import dataclass, field
from pathlib import Path
# 配置文件路径
CONFIG_DIR = Path.home() / ".miniclaw"
CONFIG_FILE = CONFIG_DIR / "config.json"
@dataclass
class Config:
"""MiniClaw 配置。"""
# 模型设置
model: str | None = None # None 表示使用 SDK 默认模型
max_turns: int = 10 # 单次对话最大轮数
# 系统提示词
system_prompt: str = (
"你是 MiniClaw,一个友好的个人助手。"
"你说中文,回答简洁明了。"
"你有多种工具可以使用,包括查看时间、系统信息、计算器和笔记功能。"
"当用户的问题可以用工具解决时,请主动使用工具。"
)
# 记忆设置
max_history: int = 50 # 最多保存多少条对话历史
# 显示设置
show_cost: bool = True # 是否显示花费
show_model: bool = False # 是否显示使用的模型
@classmethod
def load(cls) -> "Config":
"""从配置文件加载。文件不存在则用默认值。"""
if CONFIG_FILE.exists():
try:
data = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
return cls(**{
k: v for k, v in data.items()
if k in cls.__dataclass_fields__
})
except (json.JSONDecodeError, TypeError):
pass # 文件损坏,用默认值
return cls()
def save(self) -> None:
"""保存到配置文件。"""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
data = {
"model": self.model,
"max_turns": self.max_turns,
"system_prompt": self.system_prompt,
"max_history": self.max_history,
"show_cost": self.show_cost,
"show_model": self.show_model,
}
CONFIG_FILE.write_text(
json.dumps(data, ensure_ascii=False, indent=2),
encoding="utf-8",
)
miniclaw/engine.py
"""MiniClaw 对话引擎。
负责和 Claude 通信,管理对话生命周期。
本章更新:支持工具注入。
"""
from typing import Any, AsyncIterator
from claude_agent_sdk import (
AssistantMessage,
ClaudeAgentOptions,
ClaudeSDKClient,
ResultMessage,
TextBlock,
)
from claude_agent_sdk.types import McpServerConfig
from miniclaw.config import Config
from miniclaw.memory import Memory
class ChatEngine:
"""MiniClaw 对话引擎。
参数:
config: 配置对象
memory: 记忆对象
mcp_servers: MCP 服务器字典(工具注入)
allowed_tools: 允许使用的工具白名单
"""
def __init__(
self,
config: Config,
memory: Memory,
mcp_servers: dict[str, McpServerConfig] | None = None,
allowed_tools: list[str] | None = None,
) -> None:
self.config = config
self.memory = memory
self.mcp_servers = mcp_servers or {}
self.allowed_tools = allowed_tools or []
self._client: ClaudeSDKClient | None = None
def _build_options(self) -> ClaudeAgentOptions:
"""构建 ClaudeAgentOptions。"""
return ClaudeAgentOptions(
system_prompt=self.config.system_prompt,
model=self.config.model,
permission_mode="bypassPermissions",
max_turns=self.config.max_turns,
# 工具支持
mcp_servers=self.mcp_servers,
allowed_tools=self.allowed_tools,
)
async def start(self) -> None:
"""启动引擎,建立与 Claude 的连接。"""
options = self._build_options()
self._client = ClaudeSDKClient(options)
await self._client.connect()
async def stop(self) -> None:
"""停止引擎,断开连接。"""
if self._client:
await self._client.disconnect()
self._client = None
async def chat(self, user_input: str) -> AsyncIterator[str]:
"""发送消息并流式返回回复。
参数:
user_input: 用户输入
Yields:
Claude 的回复文本(可能分多次返回)
"""
if not self._client:
raise RuntimeError("引擎未启动,请先调用 start()")
# 记录用户输入
self.memory.add_user_message(user_input)
# 发送消息
await self._client.query(user_input)
# 收集完整回复
full_response = ""
cost: float | None = None
model: str = ""
async for msg in self._client.receive_response():
if isinstance(msg, AssistantMessage):
model = msg.model
for block in msg.content:
if isinstance(block, TextBlock):
full_response += block.text
yield block.text
elif isinstance(msg, ResultMessage):
cost = msg.total_cost_usd
# 输出费用信息
if self.config.show_cost and cost is not None:
yield f"\n[费用: ${cost:.6f}]"
if self.config.show_model and model:
yield f" [模型: {model}]"
# 记录助手回复
if full_response:
self.memory.add_assistant_message(full_response)
async def __aenter__(self) -> "ChatEngine":
await self.start()
return self
async def __aexit__(self, *args: Any) -> None:
await self.stop()
miniclaw/memory.py
"""MiniClaw 记忆模块。
管理对话历史,支持保存和加载。
"""
import json
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
MEMORY_DIR = Path.home() / ".miniclaw" / "history"
@dataclass
class MemoryEntry:
"""一条记忆。"""
role: str # "user" 或 "assistant"
content: str # 消息内容
timestamp: str = "" # 时间戳
def __post_init__(self) -> None:
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
@dataclass
class Memory:
"""对话记忆管理。"""
max_history: int = 50
_entries: list[MemoryEntry] = field(default_factory=list)
def add_user_message(self, content: str) -> None:
"""记录用户消息。"""
self._entries.append(MemoryEntry(role="user", content=content))
self._trim()
def add_assistant_message(self, content: str) -> None:
"""记录助手消息。"""
self._entries.append(MemoryEntry(role="assistant", content=content))
self._trim()
def _trim(self) -> None:
"""超出限制时,裁剪旧记录。"""
if len(self._entries) > self.max_history:
self._entries = self._entries[-self.max_history:]
def get_recent(self, n: int = 10) -> list[MemoryEntry]:
"""获取最近 n 条记录。"""
return self._entries[-n:]
def save(self, session_id: str = "default") -> None:
"""保存记忆到文件。"""
MEMORY_DIR.mkdir(parents=True, exist_ok=True)
filepath = MEMORY_DIR / f"{session_id}.json"
data = [
{"role": e.role, "content": e.content, "timestamp": e.timestamp}
for e in self._entries
]
filepath.write_text(
json.dumps(data, ensure_ascii=False, indent=2),
encoding="utf-8",
)
def load(self, session_id: str = "default") -> None:
"""从文件加载记忆。"""
filepath = MEMORY_DIR / f"{session_id}.json"
if not filepath.exists():
return
try:
data = json.loads(filepath.read_text(encoding="utf-8"))
self._entries = [
MemoryEntry(
role=item["role"],
content=item["content"],
timestamp=item.get("timestamp", ""),
)
for item in data
]
except (json.JSONDecodeError, KeyError):
pass # 文件损坏,忽略
@property
def count(self) -> int:
"""记录总数。"""
return len(self._entries)
def clear(self) -> None:
"""清空记忆。"""
self._entries.clear()
miniclaw/cli.py
"""MiniClaw 命令行界面。
处理用户输入、输出格式化、斜杠命令等。
"""
from miniclaw.engine import ChatEngine
# 斜杠命令
COMMANDS = {
"/help": "显示帮助信息",
"/quit": "退出 MiniClaw",
"/clear": "清空对话历史",
"/save": "保存对话历史",
"/info": "显示当前配置",
}
class CLI:
"""MiniClaw 命令行界面。"""
def __init__(self, engine: ChatEngine) -> None:
self.engine = engine
async def run(self) -> None:
"""主循环。"""
self._print_banner()
async with self.engine:
while True:
try:
user_input = input("\n你: ").strip()
except (EOFError, KeyboardInterrupt):
print("\n再见!")
break
if not user_input:
continue
# 处理斜杠命令
if user_input.startswith("/"):
should_continue = await self._handle_command(user_input)
if not should_continue:
break
continue
# 发送给引擎,流式输出回复
print("MiniClaw: ", end="", flush=True)
async for chunk in self.engine.chat(user_input):
print(chunk, end="", flush=True)
print() # 换行
async def _handle_command(self, command: str) -> bool:
"""处理斜杠命令。返回 False 表示应该退出。"""
cmd = command.lower().split()[0]
if cmd in ("/quit", "/exit", "/q"):
print("再见!")
return False
elif cmd == "/help":
print("\n可用命令:")
for name, desc in COMMANDS.items():
print(f" {name:10s} {desc}")
elif cmd == "/clear":
self.engine.memory.clear()
print("对话历史已清空。")
elif cmd == "/save":
self.engine.memory.save()
print("对话历史已保存。")
elif cmd == "/info":
config = self.engine.config
tool_count = len(self.engine.allowed_tools)
print(f"\n当前配置:")
print(f" 模型: {config.model or '默认'}")
print(f" 最大轮数: {config.max_turns}")
print(f" 对话记录: {self.engine.memory.count} 条")
print(f" 已加载工具: {tool_count} 个")
else:
print(f"未知命令: {command}。输入 /help 查看帮助。")
return True
def _print_banner(self) -> None:
"""打印欢迎横幅。"""
print("=" * 50)
print(" MiniClaw v0.3 - 你的个人 AI 助手")
print(" 输入 /help 查看帮助,/quit 退出")
print("=" * 50)
miniclaw/tools/__init__.py
"""MiniClaw 工具系统。"""
from .registry import ToolRegistry
from .builtin import register_builtin_tools
__all__ = ["ToolRegistry", "register_builtin_tools"]
miniclaw/tools/registry.py
"""工具注册中心 - 管理所有自定义工具。"""
from claude_agent_sdk import SdkMcpTool, create_sdk_mcp_server
from claude_agent_sdk.types import McpSdkServerConfig
class ToolRegistry:
"""工具注册中心:集中管理所有 MiniClaw 工具。
用法:
registry = ToolRegistry()
registry.register(my_tool) # 注册工具
server = registry.create_server() # 打包成 MCP 服务器
tools = registry.get_allowed_tools() # 获取白名单
"""
def __init__(self, server_name: str = "miniclaw") -> None:
self.server_name = server_name
self._tools: list[SdkMcpTool] = [] # type: ignore[type-arg]
def register(self, tool_def: SdkMcpTool) -> None: # type: ignore[type-arg]
"""注册一个工具。
参数:
tool_def: 用 @tool 装饰器创建的工具定义
"""
self._tools.append(tool_def)
def create_server(self) -> McpSdkServerConfig:
"""把所有已注册的工具打包成 SDK MCP 服务器。
返回值可以直接传给 ClaudeAgentOptions.mcp_servers。
"""
return create_sdk_mcp_server(
name=self.server_name,
tools=self._tools,
)
def get_allowed_tools(self) -> list[str]:
"""获取所有已注册工具的白名单列表。
返回值可以直接传给 ClaudeAgentOptions.allowed_tools。
名称格式为 mcp__<服务器名>__<工具名>。
"""
return [
f"mcp__{self.server_name}__{t.name}"
for t in self._tools
]
@property
def tool_count(self) -> int:
"""已注册的工具数量。"""
return len(self._tools)
miniclaw/tools/builtin.py
"""MiniClaw 内置工具集。
包含 5 个开箱即用的工具:
- current_time — 获取当前时间
- system_info — 查看系统信息
- calculator — 数学计算器
- note_save — 保存笔记
- note_list — 列出所有笔记
"""
from typing import Any
from claude_agent_sdk import tool
from mcp.types import ToolAnnotations
from .registry import ToolRegistry
# ============================================================
# 工具 1:当前时间
# ============================================================
@tool(
"current_time",
"获取当前日期和时间,包含年月日、时分秒、星期几",
{"timezone": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
async def current_time(args: dict[str, Any]) -> dict[str, Any]:
"""获取当前时间。
参数:
timezone: 时区名称,如 "Asia/Shanghai"、"UTC"
"""
from datetime import datetime, timezone, timedelta
tz_name = args.get("timezone", "Asia/Shanghai")
# 常见时区偏移量映射(简单实现,不依赖第三方库)
tz_offsets: dict[str, int] = {
"Asia/Shanghai": 8,
"Asia/Tokyo": 9,
"Asia/Seoul": 9,
"UTC": 0,
"US/Eastern": -5,
"US/Pacific": -8,
"Europe/London": 0,
"Europe/Paris": 1,
"Europe/Berlin": 1,
}
offset_hours = tz_offsets.get(tz_name, 8) # 默认东八区
tz = timezone(timedelta(hours=offset_hours))
now = datetime.now(tz)
# 中文星期
weekdays = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]
result = (
f"当前时间:{now.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"星期:{weekdays[now.weekday()]}\n"
f"时区:{tz_name} (UTC{offset_hours:+d})"
)
return {"content": [{"type": "text", "text": result}]}
# ============================================================
# 工具 2:系统信息
# ============================================================
@tool(
"system_info",
"获取当前系统信息,包括操作系统、CPU、内存、磁盘使用情况",
{"detail_level": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
async def system_info(args: dict[str, Any]) -> dict[str, Any]:
"""获取系统信息。
参数:
detail_level: "basic" 或 "full",full 会额外包含磁盘信息
"""
import os
import platform
info_lines = [
f"操作系统: {platform.system()} {platform.release()}",
f"架构: {platform.machine()}",
f"Python: {platform.python_version()}",
f"CPU 核心数: {os.cpu_count()}",
f"主机名: {platform.node()}",
]
# 详细模式:额外获取磁盘信息
detail = args.get("detail_level", "basic")
if detail == "full":
try:
import shutil
total, used, free = shutil.disk_usage("/")
info_lines.extend([
f"磁盘总量: {total // (1024**3)} GB",
f"磁盘已用: {used // (1024**3)} GB",
f"磁盘可用: {free // (1024**3)} GB",
])
except Exception:
info_lines.append("磁盘信息: 获取失败")
return {"content": [{"type": "text", "text": "\n".join(info_lines)}]}
# ============================================================
# 工具 3:计算器
# ============================================================
@tool(
"calculator",
"计算数学表达式。支持加减乘除、幂运算、三角函数(sin/cos/tan)、"
"对数(log/log10)、平方根(sqrt)等",
{"expression": str},
)
async def calculator(args: dict[str, Any]) -> dict[str, Any]:
"""安全地计算数学表达式。
参数:
expression: 数学表达式字符串,如 "sin(pi/4) + log10(1000)"
"""
import math
expression = args["expression"]
# 安全白名单:只允许这些函数和常量
safe_names: dict[str, Any] = {
# 内置函数
"abs": abs,
"round": round,
"min": min,
"max": max,
"pow": pow,
# math 函数
"sin": math.sin,
"cos": math.cos,
"tan": math.tan,
"asin": math.asin,
"acos": math.acos,
"atan": math.atan,
"sqrt": math.sqrt,
"log": math.log,
"log2": math.log2,
"log10": math.log10,
"exp": math.exp,
"ceil": math.ceil,
"floor": math.floor,
"factorial": math.factorial,
# 常量
"pi": math.pi,
"e": math.e,
}
try:
# __builtins__ 设为空字典,阻止访问任何内置函数
# 只有 safe_names 里的函数才能使用
result = eval(expression, {"__builtins__": {}}, safe_names)
return {
"content": [{"type": "text", "text": f"计算结果:{result}"}]
}
except ZeroDivisionError:
return {
"content": [{"type": "text", "text": "错误:除数不能为零"}],
"is_error": True,
}
except Exception as e:
return {
"content": [{"type": "text", "text": f"计算出错:{e}"}],
"is_error": True,
}
# ============================================================
# 工具 4:保存笔记
# ============================================================
@tool(
"note_save",
"保存一条笔记到本地。笔记会存储在 ~/.miniclaw/notes/ 目录下",
{"title": str, "content": str},
)
async def note_save(args: dict[str, Any]) -> dict[str, Any]:
"""保存笔记。
参数:
title: 笔记标题
content: 笔记内容
"""
from datetime import datetime
from pathlib import Path
title = args["title"]
content = args["content"]
# 确保笔记目录存在
notes_dir = Path.home() / ".miniclaw" / "notes"
notes_dir.mkdir(parents=True, exist_ok=True)
# 用时间戳 + 标题生成文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# 清理标题中不适合做文件名的字符
safe_title = "".join(
c if c.isalnum() or c in "-_" else "_"
for c in title
)
filename = f"{timestamp}_{safe_title}.md"
filepath = notes_dir / filename
# 写入 Markdown 格式的笔记
note_text = (
f"# {title}\n"
f"\n"
f"{content}\n"
f"\n"
f"---\n"
f"创建时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
)
filepath.write_text(note_text, encoding="utf-8")
return {
"content": [{"type": "text", "text": f"笔记已保存:{filepath}"}]
}
# ============================================================
# 工具 5:列出笔记
# ============================================================
@tool(
"note_list",
"列出所有已保存的笔记。可以按关键词筛选",
{"keyword": str},
annotations=ToolAnnotations(readOnlyHint=True),
)
async def note_list(args: dict[str, Any]) -> dict[str, Any]:
"""列出所有笔记。
参数:
keyword: 可选的关键词过滤,为空字符串时返回所有笔记
"""
from pathlib import Path
notes_dir = Path.home() / ".miniclaw" / "notes"
# 目录不存在,说明还没保存过笔记
if not notes_dir.exists():
return {
"content": [{"type": "text", "text": "还没有任何笔记。"}]
}
# 按修改时间倒序排列
notes = sorted(notes_dir.glob("*.md"), reverse=True)
if not notes:
return {
"content": [{"type": "text", "text": "还没有任何笔记。"}]
}
keyword = args.get("keyword", "")
lines = []
for note_path in notes:
# 读取第一行作为标题
try:
first_line = note_path.read_text(encoding="utf-8").split("\n")[0]
title = first_line.lstrip("# ").strip() or note_path.stem
except Exception:
title = note_path.stem
# 关键词筛选
if keyword and keyword.lower() not in title.lower():
continue
lines.append(f" - {title} ({note_path.name})")
if not lines:
if keyword:
text = f"没有找到包含 '{keyword}' 的笔记。"
else:
text = "还没有任何笔记。"
return {"content": [{"type": "text", "text": text}]}
header = f"共 {len(lines)} 条笔记:\n"
return {
"content": [{"type": "text", "text": header + "\n".join(lines)}]
}
# ============================================================
# 注册函数:一键注册所有内置工具
# ============================================================
def register_builtin_tools(registry: ToolRegistry) -> None:
"""把所有内置工具注册到注册中心。
参数:
registry: ToolRegistry 实例
"""
registry.register(current_time)
registry.register(system_info)
registry.register(calculator)
registry.register(note_save)
registry.register(note_list)
miniclaw/__main__.py
"""MiniClaw 入口文件。
用法:python -m miniclaw
"""
import asyncio
import sys
from miniclaw.auth import ensure_authenticated
from miniclaw.config import Config
from miniclaw.engine import ChatEngine
from miniclaw.memory import Memory
from miniclaw.cli import CLI
from miniclaw.tools import ToolRegistry, register_builtin_tools
async def main() -> None:
"""MiniClaw 主流程。"""
# 1. 检查认证
if not ensure_authenticated():
print("认证失败,请先配置 API Key 或登录。")
sys.exit(1)
# 2. 加载配置
config = Config.load()
# 3. 初始化记忆
memory = Memory(max_history=config.max_history)
# 4. 初始化工具系统
registry = ToolRegistry()
register_builtin_tools(registry)
print(f"已加载 {registry.tool_count} 个工具")
# 5. 创建引擎(注入工具)
engine = ChatEngine(
config=config,
memory=memory,
mcp_servers={"miniclaw": registry.create_server()},
allowed_tools=registry.get_allowed_tools(),
)
# 6. 启动 CLI
cli = CLI(engine=engine)
await cli.run()
if __name__ == "__main__":
asyncio.run(main())
本章文件清单
10-实战-工具生态/
README.md # 你正在读的这个文件
miniclaw/
__init__.py # 包初始化
__main__.py # 入口文件(集成工具系统)
auth.py # 认证管理
config.py # 配置管理
engine.py # 对话引擎(已更新,支持工具注入)
memory.py # 对话记忆
cli.py # 命令行界面
tools/
__init__.py # 工具包入口
registry.py # 工具注册中心
builtin.py # 5 个内置工具