按 Claude Code CLI 源码架构逐步搭建的最简可运行 Agent —— Python 3 版。
纯标准库,零第三方依赖(asyncio 驱动循环、http.client 做流式 API、http.server 做 Web)。Python ≥ 3.11。
与
myagent(Node/TypeScript 版)一一对应的姊妹项目。 核心认知不变:整个 Agent 框架只是一个递归的 async generator 循环 —— 装配消息 → 调模型 → 流式解析 → 执行工具 → 结果回灌 → 递归。
# CLI(未设 ANTHROPIC_API_KEY 时用确定性 mock 模型,可跑通全部机制)
python -m pyagent.cli
python -m pyagent.cli --mode=plan # 权限模式:default|acceptEdits|plan|bypassPermissions
python -m pyagent.cli --show-system # 查看装配后的系统提示
# Web(SSE 流式,浏览器打开 http://localhost:3000)
python -m pyagent.server.server
# 接真实模型
$env:ANTHROPIC_API_KEY = "sk-ant-..."
$env:MYAGENT_MODEL = "claude-sonnet-5" # 可选core/deps.py 内置提供方注册表(anthropic / deepseek),按
显式参数 > 环境变量 > 自动探测(看哪个 key 在) 选择。DeepSeek 走其官方
Anthropic 兼容端点(api.deepseek.com/anthropic),复用同一套流式解析:
# macOS / Linux(zsh/bash)
export DEEPSEEK_API_KEY="sk-..."
python3 -m pyagent.cli # 自动探测到 deepseek
python3 -m pyagent.cli --provider=deepseek --model=deepseek-v4-pro # 显式指定# Windows(PowerShell)
$env:DEEPSEEK_API_KEY = "sk-..."
python -m pyagent.cli可用模型:deepseek-v4-flash(默认)/ deepseek-v4-pro。
新增其他兼容端点只需在 deps.py 的 _PROVIDERS 注册表加一项(host/path/key_env/默认模型)。
mock 模型可识别的演示指令(与 myagent 完全一致):
| 输入 | 演示的机制 |
|---|---|
读取 pyproject.toml |
工具调用 → 结果回灌 → 第二轮总结 |
读 a.txt 读 b.txt 然后往 c.txt 写入 hello |
并发分区:2 个 Read 并行 ∥ + Write 串行 → |
运行 rm -rf temp |
deny 规则拦截(任何模式下) |
用子代理去读 a.txt 并总结 |
子代理递归(独立上下文 + 事件上抛) |
用只读子代理探索:... |
explore 代理:工具池过滤(无 Write)+ plan 权限双重防线 |
两轮对话 + MYAGENT_COMPACT_THRESHOLD=200 |
自动压缩(⧉ 事件) |
自顶向下四层,依赖方向单向(上层依赖下层,内核不依赖任何入口):
┌─ 入口层 ──────────────────────────────────────────────────────┐
│ cli.py(REPL) server/server.py(HTTP/SSE)+ web/index.html │
│ 职责:持有会话历史、构造权限决策器、消费事件流做渲染 │
└──────────────────────────┬────────────────────────────────────┘
│ query(QueryParams) → AsyncGenerator[AgentEvent]
┌─ 内核层 core/ ────────────▼────────────────────────────────────┐
│ query.py 主循环(压缩→调模型→收集tool_use→执行→递归) │
│ orchestration.py 并发分区 + 单工具执行(查找→校验→call) │
│ permissions.py 决策链:deny→bypass→allow→只读→模式兜底 │
│ compact.py 超阈值时把历史头部总结成摘要,保留尾段 │
│ deps.py 提供方注册表 + 依赖注入(mock/deepseek/anthropic)│
│ types.py Message/Block/AgentEvent 均为 dict(API 形状) │
└───────────┬──────────────────────────────┬─────────────────────┘
│ deps.call_model(...) │ tool.call(input, ctx)
┌─ 模型层 ──▼──────────────┐ ┌─ 工具层 ────▼──────────────────────┐
│ model/anthropic.py │ │ tools/base.py Tool 抽象基类 │
│ (http.client + SSE 解析)│ │ registry.py 工具注册表 │
│ model/mock.py │ │ read/write/bash 三个基础工具 │
│ (正则意图识别,确定性) │ │ agent_tool.py 子代理→递归query │
└──────────────────────────┘ └────────────────────────────────────┘
关键点:内核对 I/O 和 UI 一无所知。模型调用经 deps.call_model 注入,
权限询问经 can_use_tool 回调注入,事件渲染由入口层消费 generator 完成。
所以同一个 query() 能同时服务 CLI、Web、以及子代理(第三个"入口"就是 AgentTool)。
以 CLI 输入"读 a.txt 读 b.txt 然后往 c.txt 写入 hello"为例:
cli.py: messages.append(user 输入) → 调 query()
│
▼ query.py 主循环 —— 第 1 轮
⓪ auto_compact_if_needed:估算 token(len(json)/4),未超阈值 → 跳过
① deps.call_model(messages, system_prompt, tools 的 API schema)
└ mock/anthropic 流式产出:text_delta* → assistant(完整消息)
assistant.content = [text, tool_use(Read a), tool_use(Read b), tool_use(Write c)]
② 从 assistant.content 过滤出 tool_use 块(不信 stop_reason,以内容块为准)
③ 有 tool_use → 不出口,进入 ④
④ run_tools(orchestration.py):
a. partition_tool_calls 按 is_concurrency_safe 分区:
[Read a, Read b]∥(相邻安全块并成一批) → [Write c](写操作独立串行批)
b. 每批先【串行】过 can_use_tool 权限(询问用户必须串行);
被拒的直接生成 is_error 的 tool_result,模型下一轮能看到拒绝原因
c. 获批的并行批用 asyncio.as_completed 并发执行、按完成顺序 yield;
串行批逐个执行
d. execute_tool_use:查找 → validate_input(JSON Schema)→ tool.call()
全程不抛错 —— 错误文本包成 is_error 结果回给模型,让它自我修正
e. 所有 tool_result 块累积进 results 列表
⑤ results 包成一条 user 消息追加进历史;检查 max_turns 护栏
⑥ messages 整体重写,continue —— "递归"下一轮
│
▼ 第 2 轮
① 调模型,模型看到 tool_result → 产出纯文本总结
②③ 无 tool_use → yield {"type":"terminal","reason":"completed","messages":[...]}
│
▼ cli.py: 用 terminal 事件带回的 messages 更新会话历史
内核到 UI 的唯一通信方式是 query() yield 的事件流,全部为带 "type" 的 dict
(可直接 JSON 序列化 → 天然适配 SSE):
| 事件 | 产生位置 | 含义 |
|---|---|---|
turn_start |
query 循环顶部 | 新一轮开始(turn 序号) |
text_delta |
模型层 | 流式文本增量,CLI 直接打字机输出 |
assistant |
模型层 | 一轮的完整 assistant 消息(含 tool_use 块) |
tool_start |
orchestration | 某工具开始执行(parallel 标记 ∥/→) |
tool_result |
orchestration | 工具结果(content/is_error/durationMs) |
permission_denied |
orchestration | 权限拒绝(名字+原因) |
compact |
query ⓪ 步 | 发生了上下文压缩(前后 token 数) |
subagent |
入口层包装 | 子代理内部事件经 on_subagent_event 回调上抛 |
terminal |
query 出口 | 终止事件,内含最终 messages 历史(替代 return 值) |
terminal 是核心约定:Python async 生成器不能 return value,所以调用方
(CLI/Web/AgentTool)都从 terminal 事件里读回完整历史与终止原因
(completed / max_turns / aborted / error)。
create_can_use_tool(mode, allow_rules, deny_rules, ask_user) 返回闭包,
每次工具执行前被 orchestration 调用,决策顺序固定:
① deny 规则(如 "Bash(rm *)")→ 拒绝 ← 最高优先级,bypass 也不能越过
② mode == bypassPermissions → 放行
③ allow 规则(如 "Bash(git status)")→ 放行
④ tool.is_read_only(input) → 放行 ← plan 模式仍可读,立身之本
⑤ mode == plan → 拒绝一切写
mode == acceptEdits 且 Write/Edit → 放行
⑥ 兜底:有 ask_user 通道就问用户;没有(Web/子代理)→ 拒绝并说明
交互方式由入口层决定:CLI 传 input() 问答做 ask_user;Web 与子代理不传
(ask 情形自动拒绝,理由作为 tool_result 回给模型)。
子代理不是第二套引擎,而是同一个 query() 换参数再跑一遍:
- 独立 system_prompt(explore = 只读探索;general = 通用)
- 过滤后的工具池:剔除 AgentTool 自身防套娃;explore 只留 Read/Bash
- 独立权限:explore 用 plan 模式(工具池过滤 + 权限双重防线),无 ask 通道
depth + 1,超过_MAX_DEPTH=2拒绝派生- 过程事件经
ctx.on_subagent_event(agent_id, ev)回调直通 UI; 最终报告(最后一条 assistant 消息的文本)作为 tool_result 回给父级
循环依赖 registry → agent_tool → query → orchestration → registry
在 AgentTool.call() 内用延迟 import 打破。
每个工具只需实现五件事,其中三件是元数据:
| 成员 | 用途 | 消费方 |
|---|---|---|
name / description / input_schema |
告诉模型"我是谁、怎么调" | to_api_schema() → API tools 数组 |
input_schema(同一份) |
执行前本地校验 | validate_input()(极简 JSON Schema) |
is_read_only(input) |
权限判定④步依据 | permissions |
is_concurrency_safe(input) |
并发分区依据 | orchestration.partition |
call(input, ctx) |
真正执行 | execute_tool_use |
默认哲学 fail-closed:is_read_only / is_concurrency_safe 默认 False,
判定抛错也按不安全 —— 宁可串行,不可写坏。四个内置工具正好覆盖四个象限:
Read(只读+并发安全)、Write(写+串行)、Bash(按命令动态判定,只读白名单
正则)、Agent(explore 可并行,general 串行)。
QueryDeps.call_model 是统一签名 (messages, system_prompt, tools) → AsyncGenerator[ModelStreamEvent],
产出 text_delta* + 最终 assistant 两种事件:
- mock.py:无任何 API key 时启用。正则从用户文本解析意图(读 X / 写 X / 运行 X /
用子代理…),确定性地产出 tool_use 或文本,零成本跑通全部机制(含压缩:
靠系统提示里的
SUMMARIZE_CONTEXT标记识别压缩请求)。 - anthropic.py:标准库
http.client直连 Anthropic Messages 兼容端点 (官方 / DeepSeek,由ProviderConfig决定 host/path/key/model),手工解析 SSE (content_block_start/delta/stop,tool_use 参数由input_json_delta分片 拼接后json.loads)。阻塞读取放后台线程,经asyncio.Queue+call_soon_threadsafe桥接回 async generator。429/5xx 指数退避重试 3 次。
- 装配(每次输入时重建):静态块(身份 + 工具纪律)+ 缓存边界注释 +
动态块(环境信息 +
AGENT.md项目记忆)。静态在前是为了提示词缓存命中。 - 压缩(每轮调模型前检查):token 估算超阈值(
MYAGENT_COMPACT_THRESHOLD, 默认 30000)时,找到最后一条纯文本用户消息作为分界 —— 之后的尾段原样保留 (保证 tool_use/tool_result 配对不被拆散),之前的历史交给模型总结成一条 摘要消息替换。
pyagent/
├── core/
│ ├── query.py ← query.ts:241 queryLoop() 主循环内核
│ ├── orchestration.py ← toolOrchestration.ts 并发分区 + 单工具执行
│ ├── permissions.py ← useCanUseTool.ts + permissions 权限决策
│ ├── compact.py ← services/compact/autoCompact.ts 自动压缩
│ ├── deps.py ← query/deps.ts I/O 依赖注入
│ └── types.py ← types/message.ts 消息/事件类型
├── model/
│ ├── mock.py ← (无 key 时的确定性模型)
│ └── anthropic.py ← services/api/claude.ts 真实 API(http.client + SSE)
├── tools/
│ ├── base.py ← Tool.ts:362 Tool 抽象基类 + 校验
│ ├── registry.py ← tools.ts:193 getAllBaseTools() 工具注册表
│ ├── read_tool.py / write_tool.py / bash_tool.py
│ └── agent_tool.py ← tools/AgentTool/runAgent.ts 子代理(递归 query)
├── context/
│ └── system_prompt.py ← constants/prompts.ts:444 系统提示分层装配 + AGENT.md
├── server/
│ └── server.py ← QueryEngine.ts + bridge/ HTTP/SSE 后端
├── web/index.html 聊天前端
└── cli.py ← screens/REPL.tsx CLI 入口
| 步骤 | 文件 | 核心机制 |
|---|---|---|
| 1 核心循环 + 模型层 | core/query.py core/deps.py model/* |
while 生成器;tool_use 检测;依赖注入;SSE 流解析 |
| 2 工具系统 | tools/base.py registry.py |
统一接口;JSON Schema 双用(API 描述+校验);错误即结果 |
| 3 并发编排 | core/orchestration.py |
partition:只读并行/写串行;判定抛错按不安全(fail-closed) |
| 4 权限系统 | core/permissions.py |
can_use_tool 由 UI 层构造注入;deny→bypass→allow→模式兜底 |
| 5 上下文装配 | context/system_prompt.py |
静态块+缓存边界+动态块;AGENT.md ←→ CLAUDE.md |
| 6 自动压缩 | core/compact.py |
循环顶部、调模型前;保留尾段防拆散 tool_use/result 配对 |
| 7 子代理 | tools/agent_tool.py |
递归调用同一个 query();工具池过滤+独立权限;事件经回调上抛 |
| 8 Web 前后端 | server/server.py web/index.html |
会话 dict ←→ mutableMessages;AgentEvent → SSE |
TS 与 Python 的语言差异,在这三处做了对应处理(都在源码注释里标注):
-
async 生成器不能
return valueTS 的query()用return {reason, messages}返回终值,runTools用yield*拿返回值。 Python 的 async 生成器禁止带值 return,所以:query()改为 yield 一个{"type":"terminal", ...,"messages":[...]}事件,调用方从中读结果(这也正是 SSE 需要的形状);run_tools()把结果块累积进传入的results列表,而非用返回值。
-
阻塞 I/O ↔ async 桥接
http.client是阻塞的。model/anthropic.py把 SSE 读取丢到后台线程,经asyncio.Queue+loop.call_soon_threadsafe回灌成 async generator —— 对应源码用裸流避免 SDK 累积器 O(n²) 解析的效果。 -
Windows 控制台编码 Windows 默认 GBK,emoji(⏹⚙✓)会编码崩溃、管道 UTF-8 中文会被误解析。
cli.py/server.py启动时sys.std*.reconfigure(encoding="utf-8")强制 UTF-8。
ThreadingHTTPServer 每个 do_POST 在独立线程里 asyncio.run(...) 起一个事件循环,
消费 query() 生成器并把事件逐条写成 SSE(边产生边 flush)。子代理事件经
on_subagent_event 回调直接写入同一个 SSE 流。会话历史存在进程内 dict 里
(对应 QueryEngine.mutableMessages)。
与 myagent 相同:StreamingToolExecutor、microcompact/snip/context-collapse、 413/max-tokens 分层恢复、模型降级、Hooks、Skills/Commands、MCP、任务系统 (后台/远程/teammate/worktree)、转写持久化、遥测、提示词缓存分段。 每一项在源码解读文档里都有对应章节,可按需继续往本项目里加。