第 11 章：实战 — 高级特性
本章是 MiniClaw 实战项目的最后一章。我们要给这个小助手装上"安全锁"、"闹钟"、 "团队协作"和"深度思考"四大能力，让它从一个能聊天的玩具变成一个真正靠谱的 AI 助手。
本章目标
经过前三章的努力，MiniClaw 已经有了对话引擎、记忆系统和工具生态。但还差几块拼图：
| 特性 | 解决什么问题 |
|---|---|
| 安全防护 | 防止 AI 执行危险命令、泄露敏感信息 |
| 定时任务 | 让 MiniClaw 能定期做一些自动化工作 |
| 多 Agent 协作 | 遇到专业问题交给专业的"子 Agent"处理 |
| 扩展思考 | 面对复杂问题时让 Claude 想得更深 |
本章结束后，MiniClaw 就是一个功能完整的项目了。
1. 安全防护（钩子实战）
还记得第 6 章学的钩子系统吗？现在把它用起来。安全防护分三层：
- PreToolUse 钩子 — 工具执行前拦截危险操作
- PostToolUse 钩子 — 工具执行后脱敏敏感信息
- 审计日志 — 记录所有工具调用，出了问题可以回溯
1.1 拦截危险命令
最可怕的事情是 AI 跑了一个 rm -rf /。我们必须在命令执行之前就拦住它：
async def block_dangerous_commands(input_data, tool_use_id, context):
"""PreToolUse 钩子：拦截危险的 Bash 命令。"""
if input_data.get("tool_name") != "Bash":
return {} # 不是 Bash，放行
command = input_data.get("tool_input", {}).get("command", "")
# 危险命令关键词
dangerous_patterns = ["rm -rf", "sudo ", "mkfs", "> /dev/", "dd if="]
for pattern in dangerous_patterns:
if pattern in command:
return {
"decision": "block",
"reason": f"命令包含危险操作: {pattern}",
}
return {} # 安全，放行
原理：在 Bash 工具执行之前检查命令关键词，发现危险就返回 "decision": "block" 拦掉。
1.2 敏感信息脱敏
AI 的回复里可能不小心包含 API Key、密码。PostToolUse 钩子在返回结果前做清洗：
async def redact_sensitive_info(input_data, tool_use_id, context):
"""PostToolUse 钩子：对工具输出中的敏感信息脱敏。"""
response = input_data.get("tool_response", "")
if not isinstance(response, str):
return {}
# 检测 API Key / Token / 密码等模式
patterns = [
r'(?i)(api[_-]?key|token|secret)\s*[:=]\s*["\']?[\w-]{20,}',
r'(?i)(password|passwd)\s*[:=]\s*\S+',
]
for pattern in patterns:
if re.search(pattern, response):
return {
"hookSpecificOutput": {
"hookEventName": "PostToolUse",
"additionalContext": "检测到敏感信息，已提醒注意脱敏。",
}
}
return {}
1.3 审计日志
所有工具调用都记录到 ~/.miniclaw/audit.log，方便事后审计：
async def audit_tool_use(input_data, tool_use_id, context):
"""PostToolUse 钩子：记录工具调用审计日志。"""
log_file = Path.home() / ".miniclaw" / "audit.log"
record = {
"time": datetime.now().isoformat(),
"tool": input_data.get("tool_name", "unknown"),
"input": input_data.get("tool_input", {}),
}
with open(log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
return {} # 不影响正常流程
1.4 在引擎中集成钩子
把三个钩子组装到 ClaudeAgentOptions 里：
hooks = {
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[block_dangerous_commands]),
],
"PostToolUse": [
HookMatcher(matcher=None, hooks=[redact_sensitive_info, audit_tool_use]),
],
}
matcher 字段控制钩子针对哪些工具生效："Bash" 只对 Bash 生效，"Write|Edit" 对 Write 和 Edit 生效，None 对所有工具生效。完整代码在 miniclaw/engine.py。
2. 定时任务 (scheduler.py)
有时我们希望 MiniClaw 能定时做事，比如每小时检查系统状态。用 asyncio 做一个轻量调度器。
2.1 核心设计
@dataclass
class ScheduledTask:
name: str # 任务名称
interval_seconds: float # 执行间隔（秒）
callback: Callable[[], Awaitable[Any]] # 异步回调函数
enabled: bool = True # 是否启用
调度器对每个任务启动一个 asyncio Task，循环执行回调然后睡一段时间。
2.2 使用示例
scheduler = Scheduler()
async def check_status():
print(f"[{datetime.now()}] 系统状态正常")
scheduler.add_task(ScheduledTask(
name="状态检查",
interval_seconds=3600, # 每小时一次
callback=check_status,
))
await scheduler.start()
# ... 程序运行中 ...
await scheduler.stop()
在 CLI 中输入 /schedule 查看任务列表，/schedule toggle 心跳检测 切换任务开关。
3. 多 Agent 协作 (agents.py)
Claude Agent SDK 支持定义多个"子 Agent"，每个有自己的专长。主 Agent 遇到特定任务时会自动把任务交给合适的子 Agent。
3.1 AgentDefinition
from claude_agent_sdk import AgentDefinition
code_reviewer = AgentDefinition(
description="代码审查专家，检查代码质量和潜在问题",
prompt="你是一个代码审查专家。仔细分析代码，找出 bug、性能问题和安全漏洞。",
tools=["Read", "Grep", "Glob"], # 只给读取权限
model="sonnet", # 用 Sonnet，快且便宜
)
| 字段 | 含义 |
|---|---|
description |
子 Agent 的自我介绍，主 Agent 据此判断何时调用 |
prompt |
子 Agent 的系统提示词 |
tools |
可使用的工具列表 |
model |
模型："sonnet"、"opus"、"haiku" 或 "inherit" |
3.2 AgentManager
管理类统一注册所有子 Agent，预定义了三个：代码审查、文档写作、数据分析。
class AgentManager:
def __init__(self):
self._agents = {}
self._register_defaults() # 注册预定义 Agent
def get_all(self):
"""返回所有 Agent，传给 ClaudeAgentOptions.agents"""
return dict(self._agents)
3.3 注入到引擎
agent_manager = AgentManager()
options = ClaudeAgentOptions(
agents=agent_manager.get_all(),
)
就这么简单！主 Agent 会根据对话内容自动决定是否调用子 Agent。输入 /agents 查看所有可用的子 Agent：
MiniClaw> /agents
可用的子 Agent：
code-reviewer 代码审查专家，检查代码质量和潜在问题
writer 文档写作专家，擅长写技术文档和说明
analyst 数据分析专家，擅长分析数据和生成报告
4. 扩展思考模式
有些问题需要 Claude "多想一会儿"——复杂的数学推理、多步骤的逻辑分析。SDK 提供了扩展思考（Extended Thinking）功能。
4.1 三种思考模式
# 自适应 - Claude 自己决定要不要深度思考
thinking = {"type": "adaptive"}
# 启用 - 强制开启，指定思考预算（token 数量）
thinking = {"type": "enabled", "budget_tokens": 10000}
# 禁用 - 关闭思考
thinking = {"type": "disabled"}
4.2 effort 参数
不想操心 token 预算？用更简单的 effort：
options = ClaudeAgentOptions(effort="high")
# 可选值：low / medium / high / max
thinking和effort同时设置时，thinking优先级更高。一般用effort就够了。
4.3 什么时候用
| 场景 | 推荐设置 |
|---|---|
| 日常闲聊 | 不设置或 effort="low" |
| 代码编写 | effort="medium" |
| 复杂 bug 排查 | effort="high" |
| 数学证明、逻辑推理 | effort="max" |
4.4 CLI 命令
MiniClaw> /think on # 开启（默认 effort=high）
MiniClaw> /think off # 关闭
MiniClaw> /think max # 设置强度
MiniClaw> /think status # 查看当前状态
5. 完整项目收尾
5.1 最终目录结构
miniclaw/
__init__.py # 包入口
__main__.py # 启动入口 (python -m miniclaw)
auth.py # 认证管理
config.py # 配置中心
engine.py # 对话引擎（集成钩子、Agent、思考模式）
memory.py # 记忆系统
cli.py # 命令行界面（全部命令）
scheduler.py # 定时任务调度器
agents.py # 多 Agent 管理
tools/
__init__.py # 工具包入口
registry.py # 工具注册表
builtin.py # 内置工具
5.2 启动方式
export ANTHROPIC_API_KEY="你的key"
python -m miniclaw
5.3 命令一览
| 命令 | 功能 |
|---|---|
/help |
显示帮助信息 |
/clear |
清空对话历史 |
/save |
保存当前对话 |
/load |
加载历史对话 |
/tools |
查看可用工具 |
/agents |
查看子 Agent 列表 |
/think on\|off\|max |
切换思考模式 |
/schedule |
查看定时任务 |
/quit |
退出 |
5.4 核心代码串讲
引擎启动时，一次性把所有高级特性组装好：
# engine.py 核心流程
async def connect(self):
hooks = self._build_hooks() # 1. 安全防护
agents = self.agent_manager.get_all() # 2. 多 Agent
thinking = self._build_thinking_config() # 3. 思考模式
options = ClaudeAgentOptions(
system_prompt=self.config.system_prompt,
hooks=hooks,
agents=agents,
thinking=thinking,
permission_mode="bypassPermissions",
)
self._client = ClaudeSDKClient(options)
await self._client.connect()
await self.scheduler.start() # 4. 定时任务
6. 回顾与展望
整个教程学了什么
| 章节 | 内容 | 核心概念 |
|---|---|---|
| 第 0 章 | 环境准备 | SDK 安装、API Key |
| 第 1 章 | 第一次对话 | query() 函数 |
| 第 2 章 | 消息系统 | Message 类型体系 |
| 第 3 章 | 配置选项 | ClaudeAgentOptions |
| 第 4 章 | 多轮对话 | ClaudeSDKClient |
| 第 5 章 | 自定义工具 | MCP 工具 |
| 第 6 章 | 钩子系统 | HookMatcher |
| 第 7 章 | 权限控制 | PermissionMode |
| 第 8 章 | 项目架构 | MiniClaw 设计 |
| 第 9 章 | 核心引擎 | 对话 + 记忆 |
| 第 10 章 | 工具生态 | 工具注册表 |
| 第 11 章 | 高级特性 | 安全、调度、Agent、思考 |
扩展方向
MiniClaw 还有很多可以玩的方向：
- Web 界面 — 用 FastAPI + WebSocket 做一个 Web 版
- 消息平台集成 — 接入飞书、钉钉、Slack
- 更多工具 — 数据库查询、HTTP 请求、文件上传
- RAG 增强 — 接入向量数据库，让记忆系统更强大
- 多用户支持 — 每个用户独立的对话和配置
- 监控面板 — 可视化审计日志和使用统计
MiniClaw 的核心功能到这里就完成了！接下来，我们将探索 Claude Code 的 Skill 扩展体系 — 学会用 Slash Commands、Skills 和 Plugins 为你的 Agent 注入更多能力。
完整代码
以下是本章所有文件的完整可运行代码。在 miniclaw/ 目录的上级目录执行 python -m miniclaw 即可运行。
miniclaw/__init__.py
"""MiniClaw - 基于 Claude Agent SDK 的智能命令行助手（完整版）。
这是教程第 11 章的最终版本，集成了全部功能：
- 对话引擎 + 流式输出
- 记忆系统
- 工具注册 + 内置工具
- 安全防护（钩子）
- 定时任务
- 多 Agent 协作
- 扩展思考模式
"""
__version__ = "0.4.0"
__app_name__ = "MiniClaw"
miniclaw/auth.py
"""认证管理 - 管理 API Key 和认证状态。
从环境变量或配置文件中加载 API Key。
"""
import os
from pathlib import Path
class AuthManager:
"""管理 Claude API 认证。"""
# API Key 环境变量名
ENV_KEY = "ANTHROPIC_API_KEY"
def __init__(self):
self._api_key: str | None = None
def load_api_key(self) -> str | None:
"""加载 API Key，优先级：环境变量 > 配置文件。"""
# 1. 先查环境变量
key = os.environ.get(self.ENV_KEY)
if key:
self._api_key = key
return key
# 2. 再查配置文件 ~/.miniclaw/api_key
key_file = Path.home() / ".miniclaw" / "api_key"
if key_file.exists():
key = key_file.read_text(encoding="utf-8").strip()
if key:
self._api_key = key
# 设置到环境变量，让 SDK 能读到
os.environ[self.ENV_KEY] = key
return key
return None
def save_api_key(self, key: str) -> None:
"""保存 API Key 到配置文件。"""
key_dir = Path.home() / ".miniclaw"
key_dir.mkdir(exist_ok=True)
key_file = key_dir / "api_key"
key_file.write_text(key, encoding="utf-8")
# 设置文件权限为仅所有者可读写
key_file.chmod(0o600)
self._api_key = key
os.environ[self.ENV_KEY] = key
@property
def is_authenticated(self) -> bool:
"""是否已认证。"""
return self._api_key is not None
@property
def api_key_preview(self) -> str:
"""显示 API Key 的前几位（脱敏）。"""
if not self._api_key:
return "(未设置)"
return self._api_key[:12] + "..." + self._api_key[-4:]
miniclaw/config.py
"""配置中心 - 管理 MiniClaw 的全局配置。
所有配置项集中在这里管理，方便修改和扩展。
"""
from dataclasses import dataclass, field
from pathlib import Path
@dataclass
class MiniClawConfig:
"""MiniClaw 全局配置。"""
# 应用信息
app_name: str = "MiniClaw"
# 数据目录
data_dir: Path = field(default_factory=lambda: Path.home() / ".miniclaw")
# 系统提示词
system_prompt: str = (
"你是 MiniClaw，一个友好、专业的命令行智能助手。"
"你擅长编程、系统管理和解答技术问题。"
"回答要简洁明了，代码要附上中文注释。"
"如果用户的问题不够明确，主动追问。"
)
# 对话设置
max_turns: int | None = None # None 表示不限制
max_budget_usd: float | None = None # None 表示不限制
# 记忆系统
memory_file: str = "memory.json"
max_memory_entries: int = 100
# 审计日志
audit_log_file: str = "audit.log"
# 思考模式（默认关闭）
thinking_enabled: bool = False
thinking_effort: str = "high" # low / medium / high / max
# 权限模式
permission_mode: str = "bypassPermissions"
def ensure_data_dir(self) -> None:
"""确保数据目录存在。"""
self.data_dir.mkdir(parents=True, exist_ok=True)
@property
def memory_path(self) -> Path:
"""记忆文件完整路径。"""
return self.data_dir / self.memory_file
@property
def audit_log_path(self) -> Path:
"""审计日志完整路径。"""
return self.data_dir / self.audit_log_file
miniclaw/engine.py
"""对话引擎 - MiniClaw 的核心（完整版）。
集成全部功能：
- 流式对话
- 钩子（安全防护：危险命令拦截、敏感信息脱敏、审计日志）
- 多 Agent 协作
- 扩展思考模式
- MCP 工具
"""
import json
import re
from datetime import datetime
from pathlib import Path
from typing import Any
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
HookMatcher,
AssistantMessage,
ResultMessage,
TextBlock,
ThinkingBlock,
create_sdk_mcp_server,
)
from miniclaw.config import MiniClawConfig
from miniclaw.memory import MemoryManager
from miniclaw.agents import AgentManager
from miniclaw.scheduler import Scheduler, ScheduledTask
from miniclaw.tools import ToolRegistry, register_builtin_tools
# ========== 安全钩子 ==========
async def block_dangerous_commands(
input_data: dict[str, Any],
tool_use_id: str | None,
context: dict[str, Any],
) -> dict[str, Any]:
"""PreToolUse 钩子：拦截危险的 Bash 命令。
在 Bash 工具执行之前检查命令内容，发现危险操作直接拦截。
"""
# 只检查 Bash 工具
if input_data.get("tool_name") != "Bash":
return {}
command = input_data.get("tool_input", {}).get("command", "")
# 危险命令关键词列表
dangerous_patterns = [
"rm -rf /", # 删除根目录
"rm -rf ~", # 删除用户目录
"sudo rm", # sudo 删除
"mkfs", # 格式化磁盘
"> /dev/sd", # 覆盖磁盘设备
"dd if=/dev/zero", # 磁盘写零
":(){:|:&};:", # Fork 炸弹
"chmod -R 777 /", # 危险权限修改
]
for pattern in dangerous_patterns:
if pattern in command:
return {
"decision": "block",
"reason": f"[安全防护] 命令包含危险操作 '{pattern}'，已拦截。",
"systemMessage": f"危险操作被阻止: {pattern}",
}
return {} # 安全，放行
async def redact_sensitive_info(
input_data: dict[str, Any],
tool_use_id: str | None,
context: dict[str, Any],
) -> dict[str, Any]:
"""PostToolUse 钩子：对工具输出中的敏感信息进行脱敏。
检查工具的返回内容，把 API Key、密码之类的敏感信息替换成 ***。
"""
response = input_data.get("tool_response", "")
if not isinstance(response, str):
return {}
# 敏感信息正则模式
sensitive_patterns = [
# API Key / Token / Secret
(r'(?i)(api[_-]?key|token|secret|access[_-]?key)\s*[:=]\s*["\']?[\w\-]{16,}', "***REDACTED***"),
# 密码
(r'(?i)(password|passwd|pwd)\s*[:=]\s*\S+', "password=***REDACTED***"),
# AWS 风格的 Key
(r'(?:AKIA|AGPA|AIDA|AROA|AIPA|ANPA|ANVA|ASIA)[A-Z0-9]{16}', "***AWS_KEY_REDACTED***"),
]
found_sensitive = False
for pattern, _ in sensitive_patterns:
if re.search(pattern, response):
found_sensitive = True
break
if found_sensitive:
return {
"hookSpecificOutput": {
"hookEventName": "PostToolUse",
"additionalContext": "[安全防护] 检测到敏感信息，已提醒 Claude 注意脱敏。",
}
}
return {}
async def audit_tool_use(
input_data: dict[str, Any],
tool_use_id: str | None,
context: dict[str, Any],
) -> dict[str, Any]:
"""PostToolUse 钩子：记录工具调用审计日志。
所有工具调用都会被记录到 ~/.miniclaw/audit.log。
"""
log_dir = Path.home() / ".miniclaw"
log_dir.mkdir(exist_ok=True)
log_file = log_dir / "audit.log"
record = {
"time": datetime.now().isoformat(),
"tool": input_data.get("tool_name", "unknown"),
"input_summary": _summarize_input(input_data.get("tool_input", {})),
"tool_use_id": tool_use_id,
}
try:
with open(log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
except OSError:
pass # 写日志失败不应影响正常流程
return {}
def _summarize_input(tool_input: dict[str, Any]) -> str:
"""生成工具输入的摘要（避免日志太大）。"""
summary = {}
for key, value in tool_input.items():
if isinstance(value, str) and len(value) > 200:
summary[key] = value[:200] + "...(截断)"
else:
summary[key] = value
return json.dumps(summary, ensure_ascii=False)
# ========== 默认定时任务 ==========
async def heartbeat_task() -> None:
"""心跳任务：定期打印系统状态（演示用）。"""
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"\n[{timestamp}] [心跳] MiniClaw 运行正常")
# ========== 对话引擎 ==========
class MiniClawEngine:
"""MiniClaw 对话引擎（完整版）。
这是 MiniClaw 的核心，负责：
1. 管理 ClaudeSDKClient 的生命周期
2. 组装所有配置（钩子、Agent、工具、思考模式）
3. 处理对话的发送和接收
"""
def __init__(self, config: MiniClawConfig | None = None) -> None:
self.config = config or MiniClawConfig()
self.config.ensure_data_dir()
# 记忆系统
self.memory = MemoryManager(self.config.memory_path)
self.memory.load()
# Agent 管理
self.agent_manager = AgentManager()
# 工具注册表
self.tool_registry = ToolRegistry()
register_builtin_tools(self.tool_registry)
# 定时任务调度器
self.scheduler = Scheduler()
# 客户端实例（connect 后创建）
self._client: ClaudeSDKClient | None = None
# 思考模式状态
self._thinking_enabled = self.config.thinking_enabled
self._thinking_effort = self.config.thinking_effort
def _build_hooks(self) -> dict[str, list[HookMatcher]]:
"""构建钩子配置。"""
return {
"PreToolUse": [
HookMatcher(
matcher="Bash", # 只拦截 Bash 工具的危险命令
hooks=[block_dangerous_commands],
),
],
"PostToolUse": [
HookMatcher(
matcher=None, # 对所有工具生效
hooks=[redact_sensitive_info, audit_tool_use],
),
],
}
def _build_thinking_config(self) -> dict[str, Any] | None:
"""构建思考模式配置。"""
if not self._thinking_enabled:
return None
# 使用 enabled 模式，预算 10000 token
return {"type": "enabled", "budget_tokens": 10000}
def _build_effort(self) -> str | None:
"""构建 effort 参数。"""
if self._thinking_enabled:
return self._thinking_effort
return None
async def connect(self) -> None:
"""启动引擎：创建客户端并连接。"""
# 创建 MCP 工具服务器
mcp_tools = self.tool_registry.get_mcp_tools()
mcp_servers = {}
if mcp_tools:
mcp_servers["miniclaw-tools"] = create_sdk_mcp_server(
name="miniclaw-tools",
version="1.0.0",
tools=mcp_tools,
)
# 组装选项
options = ClaudeAgentOptions(
system_prompt=self.config.system_prompt,
permission_mode=self.config.permission_mode, # type: ignore[arg-type]
max_turns=self.config.max_turns,
max_budget_usd=self.config.max_budget_usd,
hooks=self._build_hooks(), # type: ignore[arg-type]
agents=self.agent_manager.get_all(),
thinking=self._build_thinking_config(), # type: ignore[arg-type]
effort=self._build_effort(), # type: ignore[arg-type]
mcp_servers=mcp_servers,
)
# 创建客户端并连接
self._client = ClaudeSDKClient(options)
await self._client.connect()
# 注册默认定时任务（演示：1 小时心跳）
self.scheduler.add_task(ScheduledTask(
name="心跳检测",
interval_seconds=3600,
callback=heartbeat_task,
enabled=False, # 默认不启用，用户通过 /schedule 开启
))
# 启动调度器
await self.scheduler.start()
async def send_message(self, message: str) -> str:
"""发送消息并收集完整回复。
Args:
message: 用户输入的消息。
Returns:
Claude 的完整文本回复。
"""
if not self._client:
raise RuntimeError("引擎未连接，请先调用 connect()")
# 记录用户消息
self.memory.add_entry("user", message)
# 发送消息
await self._client.query(message)
# 收集回复
full_response = ""
async for msg in self._client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
full_response += block.text
elif isinstance(block, ThinkingBlock):
# 思考内容可以选择性展示
pass
elif isinstance(msg, ResultMessage):
# 对话结束，可以获取费用信息
if msg.total_cost_usd:
cost_info = f"\n[费用: ${msg.total_cost_usd:.4f}]"
full_response += cost_info
# 记录助手回复
self.memory.add_entry("assistant", full_response)
return full_response
async def send_message_stream(self, message: str):
"""流式发送消息，逐步 yield 回复片段。
Args:
message: 用户输入的消息。
Yields:
(type, content) 元组：
- ("text", str): 文本片段
- ("thinking", str): 思考内容
- ("result", ResultMessage): 结果消息
"""
if not self._client:
raise RuntimeError("引擎未连接，请先调用 connect()")
self.memory.add_entry("user", message)
await self._client.query(message)
full_response = ""
async for msg in self._client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
full_response += block.text
yield ("text", block.text)
elif isinstance(block, ThinkingBlock):
yield ("thinking", block.thinking)
elif isinstance(msg, ResultMessage):
self.memory.add_entry("assistant", full_response)
yield ("result", msg)
def set_thinking(self, enabled: bool, effort: str = "high") -> None:
"""设置思考模式。
注意：修改后需要重新连接才能生效（因为 options 在连接时就固定了）。
对于 MiniClaw 这种长连接的场景，这个设置在下次重启时生效。
"""
self._thinking_enabled = enabled
self._thinking_effort = effort
@property
def thinking_enabled(self) -> bool:
"""当前是否开启了扩展思考。"""
return self._thinking_enabled
@property
def thinking_effort(self) -> str:
"""当前的思考强度。"""
return self._thinking_effort
async def disconnect(self) -> None:
"""断开连接，释放资源。"""
# 停止定时任务
await self.scheduler.stop()
# 保存记忆
self.memory.save()
# 断开客户端
if self._client:
await self._client.disconnect()
self._client = None
miniclaw/memory.py
"""记忆系统 - 管理对话历史的持久化存储。
把对话记录保存到本地文件，下次启动时可以加载回来。
"""
import json
from dataclasses import dataclass, field, asdict
from datetime import datetime
from pathlib import Path
@dataclass
class MemoryEntry:
"""一条记忆记录。"""
role: str # "user" 或 "assistant"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
@dataclass
class ConversationMemory:
"""一次对话的完整记忆。"""
session_id: str
title: str # 对话标题（取自第一条用户消息）
entries: list[MemoryEntry] = field(default_factory=list)
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
class MemoryManager:
"""记忆管理器 - 负责对话历史的增删查改。"""
def __init__(self, storage_path: Path):
self._path = storage_path
self._conversations: dict[str, ConversationMemory] = {}
self._current_session: str | None = None
def load(self) -> None:
"""从文件加载记忆数据。"""
if not self._path.exists():
return
try:
data = json.loads(self._path.read_text(encoding="utf-8"))
for session_id, conv_data in data.items():
entries = [
MemoryEntry(**entry) for entry in conv_data.get("entries", [])
]
self._conversations[session_id] = ConversationMemory(
session_id=session_id,
title=conv_data.get("title", "未命名对话"),
entries=entries,
created_at=conv_data.get("created_at", ""),
updated_at=conv_data.get("updated_at", ""),
)
except (json.JSONDecodeError, KeyError) as e:
print(f"[记忆系统] 加载记忆数据失败: {e}")
def save(self) -> None:
"""保存记忆数据到文件。"""
self._path.parent.mkdir(parents=True, exist_ok=True)
data = {}
for session_id, conv in self._conversations.items():
data[session_id] = asdict(conv)
self._path.write_text(
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
)
def new_session(self, session_id: str) -> None:
"""开始一个新的对话会话。"""
self._current_session = session_id
self._conversations[session_id] = ConversationMemory(
session_id=session_id,
title="新对话",
)
def add_entry(self, role: str, content: str) -> None:
"""添加一条记忆。"""
if not self._current_session:
return
conv = self._conversations.get(self._current_session)
if not conv:
return
entry = MemoryEntry(role=role, content=content)
conv.entries.append(entry)
conv.updated_at = datetime.now().isoformat()
# 如果是第一条用户消息，用它做对话标题
if role == "user" and len(conv.entries) == 1:
conv.title = content[:50] + ("..." if len(content) > 50 else "")
def get_current_entries(self) -> list[MemoryEntry]:
"""获取当前会话的所有记忆。"""
if not self._current_session:
return []
conv = self._conversations.get(self._current_session)
return conv.entries if conv else []
def list_sessions(self) -> list[tuple[str, str, str]]:
"""列出所有会话：(session_id, title, updated_at)。"""
return [
(sid, conv.title, conv.updated_at)
for sid, conv in self._conversations.items()
]
def clear_current(self) -> None:
"""清空当前会话的记忆。"""
if self._current_session and self._current_session in self._conversations:
self._conversations[self._current_session].entries.clear()
miniclaw/cli.py
"""命令行界面 - MiniClaw 的用户交互层（完整版）。
支持的命令：
/help 显示帮助信息
/clear 清空对话历史
/save 保存当前对话
/load 加载历史对话
/tools 查看可用工具
/agents 查看子 Agent 列表
/think 切换思考模式
/schedule 查看和管理定时任务
/quit 退出
"""
import uuid
from datetime import datetime
from miniclaw import __app_name__, __version__
from miniclaw.auth import AuthManager
from miniclaw.config import MiniClawConfig
from miniclaw.engine import MiniClawEngine
class MiniClawCLI:
"""MiniClaw 命令行界面。
负责用户输入处理、命令分发、输出格式化。
对话逻辑全部交给 MiniClawEngine 处理。
"""
def __init__(self) -> None:
self.config = MiniClawConfig()
self.auth = AuthManager()
self.engine = MiniClawEngine(self.config)
self._running = False
async def run(self) -> None:
"""主运行循环。"""
# 1. 检查认证
if not self._check_auth():
return
# 2. 连接引擎
print(f"正在启动 {__app_name__}...")
try:
await self.engine.connect()
except Exception as e:
print(f"启动失败: {e}")
print("请检查 API Key 是否正确，以及网络连接是否正常。")
return
# 3. 创建新的对话会话
session_id = str(uuid.uuid4())[:8]
self.engine.memory.new_session(session_id)
print(f"就绪！会话 ID: {session_id}")
print("输入 /help 查看帮助，/quit 退出\n")
# 4. 主循环
self._running = True
try:
while self._running:
try:
user_input = input(f"{__app_name__}> ").strip()
except EOFError:
break
if not user_input:
continue
# 判断是命令还是对话
if user_input.startswith("/"):
await self._handle_command(user_input)
else:
await self._handle_chat(user_input)
finally:
# 5. 清理
print("\n正在保存数据并断开连接...")
await self.engine.disconnect()
print("已退出。")
def _check_auth(self) -> bool:
"""检查认证状态。"""
key = self.auth.load_api_key()
if key:
print(f"API Key: {self.auth.api_key_preview}")
return True
print("未找到 API Key！")
print("请通过以下方式之一设置：")
print(" 1. 设置环境变量: export ANTHROPIC_API_KEY='你的key'")
print(" 2. 直接输入（将保存到 ~/.miniclaw/api_key）：")
try:
key_input = input("API Key: ").strip()
except (EOFError, KeyboardInterrupt):
return False
if key_input:
self.auth.save_api_key(key_input)
print("API Key 已保存！")
return True
print("未输入 API Key，退出。")
return False
async def _handle_chat(self, message: str) -> None:
"""处理对话消息。"""
try:
print() # 空行分隔
# 使用流式模式，逐步输出
has_thinking = False
async for msg_type, content in self.engine.send_message_stream(message):
if msg_type == "text":
print(content, end="", flush=True)
elif msg_type == "thinking":
if not has_thinking:
print("[思考中...]", flush=True)
has_thinking = True
elif msg_type == "result":
# result 是 ResultMessage 对象
if hasattr(content, "total_cost_usd") and content.total_cost_usd:
print(f"\n[费用: ${content.total_cost_usd:.4f}]", end="")
print("\n") # 回复结束后换行
except Exception as e:
print(f"\n对话出错: {e}\n")
async def _handle_command(self, command: str) -> None:
"""处理斜杠命令。"""
parts = command.split(maxsplit=1)
cmd = parts[0].lower()
args = parts[1] if len(parts) > 1 else ""
handlers = {
"/help": self._cmd_help,
"/clear": self._cmd_clear,
"/save": self._cmd_save,
"/load": self._cmd_load,
"/tools": self._cmd_tools,
"/agents": self._cmd_agents,
"/think": self._cmd_think,
"/schedule": self._cmd_schedule,
"/quit": self._cmd_quit,
"/exit": self._cmd_quit,
}
handler = handlers.get(cmd)
if handler:
await handler(args)
else:
print(f"未知命令: {cmd}，输入 /help 查看帮助\n")
# ========== 命令处理器 ==========
async def _cmd_help(self, args: str) -> None:
"""显示帮助信息。"""
help_text = f"""
{__app_name__} v{__version__} - 命令帮助
对话命令：
/clear 清空当前对话历史
/save 保存当前对话到记忆系统
/load 查看和加载历史对话
工具与 Agent：
/tools 查看可用的自定义工具
/agents 查看可用的子 Agent
高级功能：
/think on 开启扩展思考模式
/think off 关闭扩展思考模式
/think <level> 设置思考强度 (low/medium/high/max)
/schedule 查看定时任务列表
其他：
/help 显示本帮助
/quit 退出 {__app_name__}
"""
print(help_text)
async def _cmd_clear(self, args: str) -> None:
"""清空对话历史。"""
self.engine.memory.clear_current()
print("对话历史已清空\n")
async def _cmd_save(self, args: str) -> None:
"""保存对话。"""
self.engine.memory.save()
entries = self.engine.memory.get_current_entries()
print(f"已保存！当前对话包含 {len(entries)} 条消息\n")
async def _cmd_load(self, args: str) -> None:
"""列出并加载历史对话。"""
sessions = self.engine.memory.list_sessions()
if not sessions:
print("没有历史对话\n")
return
print("历史对话：")
for i, (sid, title, updated) in enumerate(sessions, 1):
# 格式化时间
try:
dt = datetime.fromisoformat(updated)
time_str = dt.strftime("%m-%d %H:%M")
except ValueError:
time_str = updated[:16]
print(f" {i}. [{sid}] {title} ({time_str})")
print()
async def _cmd_tools(self, args: str) -> None:
"""显示可用工具。"""
categories = self.engine.tool_registry.get_tools_by_category()
if not categories:
print("没有已注册的自定义工具\n")
return
print("可用的自定义工具：\n")
for category, tools in categories.items():
print(f" [{category}]")
for tool_info in tools:
print(f" {tool_info.name:20s} {tool_info.description}")
print()
async def _cmd_agents(self, args: str) -> None:
"""显示可用的子 Agent。"""
agents = self.engine.agent_manager.list_agents()
if not agents:
print("没有可用的子 Agent\n")
return
print("可用的子 Agent：\n")
# 计算名称最大宽度，方便对齐
max_name_len = max(len(name) for name, _ in agents)
for name, description in agents:
print(f" {name:<{max_name_len + 2}} {description}")
print()
print("提示：在对话中提到 Agent 名称，主 Agent 会自动调用它。\n")
async def _cmd_think(self, args: str) -> None:
"""切换思考模式。"""
args = args.strip().lower()
if not args or args == "on":
self.engine.set_thinking(True, "high")
print("已开启扩展思考模式 (effort=high)")
print("注意：修改将在下次重新连接时生效\n")
elif args == "off":
self.engine.set_thinking(False)
print("已关闭扩展思考模式")
print("注意：修改将在下次重新连接时生效\n")
elif args in ("low", "medium", "high", "max"):
self.engine.set_thinking(True, args)
print(f"已设置思考强度为 {args}")
print("注意：修改将在下次重新连接时生效\n")
elif args == "status":
status = "开启" if self.engine.thinking_enabled else "关闭"
print(f"扩展思考: {status}")
if self.engine.thinking_enabled:
print(f"思考强度: {self.engine.thinking_effort}")
print()
else:
print("用法: /think on|off|low|medium|high|max|status\n")
async def _cmd_schedule(self, args: str) -> None:
"""查看和管理定时任务。"""
args = args.strip()
if not args or args == "list":
# 列出所有任务
tasks = self.engine.scheduler.list_tasks()
if not tasks:
print("没有定时任务\n")
return
status_str = "运行中" if self.engine.scheduler.is_running else "已停止"
print(f"定时任务（调度器: {status_str}）：\n")
for name, interval, enabled in tasks:
status = "开" if enabled else "关"
# 友好的时间显示
if interval >= 3600:
interval_str = f"{interval / 3600:.0f} 小时"
elif interval >= 60:
interval_str = f"{interval / 60:.0f} 分钟"
else:
interval_str = f"{interval:.0f} 秒"
print(f" [{status}] {name} - 每 {interval_str} 执行一次")
print()
elif args.startswith("toggle "):
# 切换任务状态
task_name = args[7:].strip()
result = self.engine.scheduler.toggle_task(task_name)
if result is None:
print(f"找不到任务: {task_name}\n")
else:
status = "开启" if result else "关闭"
print(f"任务 '{task_name}' 已{status}\n")
else:
print("用法:")
print(" /schedule 查看所有定时任务")
print(" /schedule toggle <名称> 切换任务开关\n")
async def _cmd_quit(self, args: str) -> None:
"""退出程序。"""
self._running = False
print("正在退出...\n")
miniclaw/scheduler.py
"""定时任务调度器 - 基于 asyncio 的轻量调度。
用法：
scheduler = Scheduler()
scheduler.add_task(ScheduledTask(
name="状态检查",
interval_seconds=3600,
callback=check_status,
))
await scheduler.start()
# ... 程序运行中 ...
await scheduler.stop()
"""
import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import datetime
from typing import Any
@dataclass
class ScheduledTask:
"""一个定时任务。
Attributes:
name: 任务名称，用于显示和管理。
interval_seconds: 执行间隔，单位秒。
callback: 异步回调函数，每次触发时执行。
enabled: 是否启用，禁用的任务不会被调度。
"""
name: str
interval_seconds: float
callback: Callable[[], Awaitable[Any]]
enabled: bool = True
class Scheduler:
"""轻量级异步任务调度器。
基于 asyncio.create_task 实现，每个定时任务对应一个后台协程。
协程在循环中执行回调、等待间隔时间，如此反复。
"""
def __init__(self) -> None:
self._tasks: list[ScheduledTask] = []
self._running_tasks: dict[str, asyncio.Task[None]] = {}
self._running = False
def add_task(self, task: ScheduledTask) -> None:
"""添加一个定时任务。
如果调度器已经在运行，新任务会立即启动。
"""
self._tasks.append(task)
# 如果调度器已经在跑了，立即启动新任务
if self._running and task.enabled:
self._running_tasks[task.name] = asyncio.create_task(
self._run_task(task)
)
def remove_task(self, name: str) -> bool:
"""按名称移除一个定时任务。返回是否成功。"""
# 先取消正在运行的协程
if name in self._running_tasks:
self._running_tasks[name].cancel()
del self._running_tasks[name]
# 从列表中移除
for i, task in enumerate(self._tasks):
if task.name == name:
self._tasks.pop(i)
return True
return False
def toggle_task(self, name: str) -> bool | None:
"""切换任务的启用/禁用状态。返回新状态，None 表示没找到。"""
for task in self._tasks:
if task.name == name:
task.enabled = not task.enabled
# 如果正在运行，需要相应地启动或停止
if self._running:
if task.enabled and task.name not in self._running_tasks:
self._running_tasks[task.name] = asyncio.create_task(
self._run_task(task)
)
elif not task.enabled and task.name in self._running_tasks:
self._running_tasks[task.name].cancel()
del self._running_tasks[task.name]
return task.enabled
return None
async def _run_task(self, task: ScheduledTask) -> None:
"""执行单个定时任务的循环。"""
# 首次执行前先等一个间隔（避免启动时立即触发）
await asyncio.sleep(task.interval_seconds)
while self._running and task.enabled:
try:
await task.callback()
except asyncio.CancelledError:
# 被取消了，安静退出
break
except Exception as e:
# 任务出错不影响调度器，打印错误继续
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"[{timestamp}] [调度器] 任务 '{task.name}' 出错: {e}")
await asyncio.sleep(task.interval_seconds)
async def start(self) -> None:
"""启动调度器，开始执行所有已启用的任务。"""
if self._running:
return # 避免重复启动
self._running = True
for task in self._tasks:
if task.enabled:
self._running_tasks[task.name] = asyncio.create_task(
self._run_task(task)
)
async def stop(self) -> None:
"""停止调度器，取消所有正在运行的任务。"""
self._running = False
for name, t in self._running_tasks.items():
t.cancel()
# 等待任务真正结束，忽略 CancelledError
try:
await t
except asyncio.CancelledError:
pass
self._running_tasks.clear()
def list_tasks(self) -> list[tuple[str, float, bool]]:
"""列出所有任务：(name, interval_seconds, enabled)。"""
return [
(task.name, task.interval_seconds, task.enabled)
for task in self._tasks
]
@property
def is_running(self) -> bool:
"""调度器是否正在运行。"""
return self._running
miniclaw/agents.py
"""多 Agent 管理 - 定义和管理专业子 Agent。
通过 AgentDefinition 定义不同专长的子 Agent，
主 Agent 会根据对话内容自动决定是否交给子 Agent 处理。
用法：
manager = AgentManager()
agents = manager.get_all() # 传给 ClaudeAgentOptions.agents
"""
from claude_agent_sdk import AgentDefinition
class AgentManager:
"""管理多个 Agent 定义。
内置三个预定义的 Agent：
- code-reviewer: 代码审查专家
- writer: 文档写作专家
- analyst: 数据分析专家
也可以通过 register() 方法注册自定义 Agent。
"""
def __init__(self) -> None:
self._agents: dict[str, AgentDefinition] = {}
self._register_defaults()
def _register_defaults(self) -> None:
"""注册默认的专业 Agent。"""
# 代码审查专家：只有读取权限，不能修改代码
self._agents["code-reviewer"] = AgentDefinition(
description="代码审查专家，检查代码质量和潜在问题",
prompt=(
"你是一个代码审查专家。仔细分析代码，找出以下问题：\n"
"1. Bug 和逻辑错误\n"
"2. 性能问题和内存泄漏\n"
"3. 安全漏洞\n"
"4. 代码风格和可读性\n"
"给出建设性的反馈，说明问题在哪里以及如何修复。"
),
tools=["Read", "Grep", "Glob"],
model="sonnet",
)
# 文档写作专家：可以读写文件
self._agents["writer"] = AgentDefinition(
description="文档写作专家，擅长写技术文档和说明",
prompt=(
"你是一个技术文档专家。你的职责：\n"
"1. 写出清晰、准确、易懂的文档\n"
"2. 使用恰当的 Markdown 格式\n"
"3. 包含代码示例和使用说明\n"
"4. 保持结构化的章节组织\n"
"用中文写作，语气专业但不失亲切。"
),
tools=["Read", "Write"],
model="sonnet",
)
# 数据分析专家：可以读取文件和执行命令
self._agents["analyst"] = AgentDefinition(
description="数据分析专家，擅长分析数据和生成报告",
prompt=(
"你是一个数据分析专家。你擅长：\n"
"1. 读取和解析各种数据格式（CSV, JSON, 日志等）\n"
"2. 进行统计分析和趋势发现\n"
"3. 生成清晰的分析报告\n"
"4. 用简单的语言解释复杂的数据发现\n"
"分析时要注意数据质量问题，给出可操作的建议。"
),
tools=["Read", "Grep", "Glob", "Bash"],
model="sonnet",
)
def register(self, name: str, agent: AgentDefinition) -> None:
"""注册一个自定义 Agent。
Args:
name: Agent 的唯一标识符。
agent: Agent 的定义。
"""
self._agents[name] = agent
def unregister(self, name: str) -> bool:
"""取消注册一个 Agent。返回是否成功。"""
if name in self._agents:
del self._agents[name]
return True
return False
def get(self, name: str) -> AgentDefinition | None:
"""按名称获取 Agent 定义。"""
return self._agents.get(name)
def get_all(self) -> dict[str, AgentDefinition]:
"""获取所有 Agent 定义，直接传给 ClaudeAgentOptions.agents。"""
return dict(self._agents)
def list_agents(self) -> list[tuple[str, str]]:
"""列出所有 Agent：(name, description)。"""
return [
(name, agent.description) for name, agent in self._agents.items()
]
miniclaw/tools/__init__.py
"""MiniClaw 工具包 - 提供可扩展的工具系统。
通过 ToolRegistry 注册和管理工具，
通过 builtin 模块提供内置工具。
"""
from miniclaw.tools.registry import ToolRegistry
from miniclaw.tools.builtin import register_builtin_tools
__all__ = ["ToolRegistry", "register_builtin_tools"]
miniclaw/tools/registry.py
"""工具注册表 - 管理 MiniClaw 的工具生态。
工具通过 register() 方法注册到注册表中，
引擎在初始化时读取注册表来配置 MCP 工具。
"""
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any
from claude_agent_sdk import SdkMcpTool
@dataclass
class ToolInfo:
"""工具的描述信息。"""
name: str
description: str
category: str = "通用" # 工具分类
class ToolRegistry:
"""工具注册表。
用法：
registry = ToolRegistry()
registry.register(my_tool, category="开发")
mcp_tools = registry.get_mcp_tools() # 传给 create_sdk_mcp_server
"""
def __init__(self) -> None:
self._tools: dict[str, SdkMcpTool[Any]] = {}
self._info: dict[str, ToolInfo] = {}
def register(
self,
tool: SdkMcpTool[Any],
category: str = "通用",
) -> None:
"""注册一个 MCP 工具。
Args:
tool: 通过 @tool 装饰器创建的 SdkMcpTool。
category: 工具分类，用于展示。
"""
self._tools[tool.name] = tool
self._info[tool.name] = ToolInfo(
name=tool.name,
description=tool.description,
category=category,
)
def unregister(self, name: str) -> bool:
"""取消注册一个工具。"""
if name in self._tools:
del self._tools[name]
del self._info[name]
return True
return False
def get_mcp_tools(self) -> list[SdkMcpTool[Any]]:
"""获取所有已注册的 MCP 工具列表。"""
return list(self._tools.values())
def get_tool_names(self) -> list[str]:
"""获取所有工具名称。"""
return list(self._tools.keys())
def list_tools(self) -> list[ToolInfo]:
"""列出所有工具的描述信息。"""
return list(self._info.values())
def get_tools_by_category(self) -> dict[str, list[ToolInfo]]:
"""按分类返回工具信息。"""
categories: dict[str, list[ToolInfo]] = {}
for info in self._info.values():
if info.category not in categories:
categories[info.category] = []
categories[info.category].append(info)
return categories
miniclaw/tools/builtin.py
"""内置工具 - MiniClaw 自带的工具集。
提供一些常用的自定义工具，通过 SDK MCP Server 暴露给 Claude。
"""
import platform
from datetime import datetime
from typing import Any
from claude_agent_sdk import tool
from miniclaw.tools.registry import ToolRegistry
# ========== 工具定义 ==========
@tool(
name="get_current_time",
description="获取当前时间和日期",
input_schema={
"type": "object",
"properties": {
"format": {
"type": "string",
"description": "时间格式，如 '%Y-%m-%d %H:%M:%S'（可选）",
}
},
"required": [],
},
)
async def get_current_time(args: dict[str, Any]) -> dict[str, Any]:
"""返回当前时间。"""
fmt = args.get("format", "%Y-%m-%d %H:%M:%S")
now = datetime.now().strftime(fmt)
return {"content": [{"type": "text", "text": f"当前时间: {now}"}]}
@tool(
name="get_system_info",
description="获取系统信息（操作系统、Python版本等）",
input_schema={
"type": "object",
"properties": {},
"required": [],
},
)
async def get_system_info(args: dict[str, Any]) -> dict[str, Any]:
"""返回系统基本信息。"""
import sys
info_lines = [
f"操作系统: {platform.system()} {platform.release()}",
f"主机名: {platform.node()}",
f"架构: {platform.machine()}",
f"Python: {sys.version}",
f"平台: {platform.platform()}",
]
return {"content": [{"type": "text", "text": "\n".join(info_lines)}]}
@tool(
name="note_pad",
description="简易记事本，可以保存和读取笔记",
input_schema={
"type": "object",
"properties": {
"action": {
"type": "string",
"description": "操作类型: 'write' 写入, 'read' 读取, 'list' 列表",
},
"title": {
"type": "string",
"description": "笔记标题",
},
"content": {
"type": "string",
"description": "笔记内容（写入时必填）",
},
},
"required": ["action"],
},
)
async def note_pad(args: dict[str, Any]) -> dict[str, Any]:
"""简易记事本工具。"""
import json
from pathlib import Path
notes_file = Path.home() / ".miniclaw" / "notes.json"
notes_file.parent.mkdir(parents=True, exist_ok=True)
# 加载已有笔记
notes: dict[str, str] = {}
if notes_file.exists():
try:
notes = json.loads(notes_file.read_text(encoding="utf-8"))
except json.JSONDecodeError:
notes = {}
action = args.get("action", "list")
if action == "write":
title = args.get("title", "未命名")
content = args.get("content", "")
notes[title] = content
notes_file.write_text(
json.dumps(notes, ensure_ascii=False, indent=2), encoding="utf-8"
)
return {"content": [{"type": "text", "text": f"笔记 '{title}' 已保存"}]}
elif action == "read":
title = args.get("title", "")
if title in notes:
return {
"content": [
{"type": "text", "text": f"# {title}\n\n{notes[title]}"}
]
}
return {
"content": [{"type": "text", "text": f"找不到笔记: {title}"}],
"is_error": True,
}
else: # list
if not notes:
return {"content": [{"type": "text", "text": "还没有任何笔记"}]}
titles = "\n".join(f" - {t}" for t in notes.keys())
return {
"content": [{"type": "text", "text": f"现有笔记:\n{titles}"}]
}
# ========== 注册函数 ==========
def register_builtin_tools(registry: ToolRegistry) -> None:
"""把所有内置工具注册到注册表。"""
registry.register(get_current_time, category="实用工具")
registry.register(get_system_info, category="系统信息")
registry.register(note_pad, category="实用工具")
miniclaw/__main__.py
"""MiniClaw 启动入口 - python -m miniclaw 即可运行。
这是完整版入口文件，集成全部功能：
安全防护、定时任务、多 Agent 协作、扩展思考。
"""
import asyncio
import sys
from miniclaw import __app_name__, __version__
from miniclaw.cli import MiniClawCLI
def print_banner():
"""打印启动横幅。"""
banner = f"""
╔══════════════════════════════════════╗
║ {__app_name__} v{__version__} ║
║ 基于 Claude Agent SDK 的智能助手 ║
╚══════════════════════════════════════╝
功能清单：
- 多轮对话 + 流式输出
- 记忆系统（对话持久化）
- 工具生态（可扩展）
- 安全防护（危险命令拦截 + 敏感信息脱敏）
- 定时任务调度
- 多 Agent 协作
- 扩展思考模式
输入 /help 查看所有命令
"""
print(banner)
def main():
"""主入口函数。"""
print_banner()
cli = MiniClawCLI()
try:
asyncio.run(cli.run())
except KeyboardInterrupt:
print("\n再见！")
sys.exit(0)
if __name__ == "__main__":
main()
下一章： 第 12 章：Skill 与扩展体系