clawd-code-python-port
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
Chineseclawd-code Python Port
clawd-code Python移植版
What This Project Does
项目功能
clawd-code is an independent Python rewrite of the Claude Code agent harness, built from scratch for educational purposes. It captures the architectural patterns of Claude Code — tool wiring, command dispatch, task orchestration, and agent runtime context — in clean Python, without copying any proprietary TypeScript source.
The project is orchestrated end-to-end using oh-my-codex (OmX), a workflow layer on top of OpenAI Codex. It is not affiliated with or endorsed by Anthropic.
clawd-code是Claude Code Agent框架的独立Python重写版本,出于教育目的从零开始构建。它用简洁的Python复现了Claude Code的架构模式——工具连接、命令分发、任务编排和Agent运行时上下文,没有复制任何专有TypeScript源代码。
该项目全流程由oh-my-codex (OmX)编排,OmX是基于OpenAI Codex的工作流层。本项目与Anthropic无任何关联,也未获得其官方认可。
Installation
安装
bash
undefinedbash
undefinedClone the repository
克隆仓库
git clone https://github.com/instructkr/clawd-code.git
cd clawd-code
git clone https://github.com/instructkr/clawd-code.git
cd clawd-code
(Optional but recommended) Create a virtual environment
(可选但推荐)创建虚拟环境
python3 -m venv .venv
source .venv/bin/activate
python3 -m venv .venv
source .venv/bin/activate
Install dependencies (if a requirements.txt or pyproject.toml is present)
安装依赖(如果存在requirements.txt或pyproject.toml)
pip install -r requirements.txt
pip install -r requirements.txt
or
或者
pip install -e .
No API keys are needed for the manifest/summary/CLI commands. If you extend the query engine to call a live model, set your key via environment variable:
```bash
export ANTHROPIC_API_KEY="your-key-here"
export OPENAI_API_KEY="your-key-here"pip install -e .
清单/摘要/CLI命令无需API密钥。如果你扩展查询引擎来调用在线模型,可以通过环境变量设置密钥:
```bash
export ANTHROPIC_API_KEY="your-key-here"
export OPENAI_API_KEY="your-key-here"Repository Layout
仓库结构
.
├── src/
│ ├── __init__.py
│ ├── commands.py # Command port metadata
│ ├── main.py # CLI entrypoint
│ ├── models.py # Dataclasses: subsystems, modules, backlog
│ ├── port_manifest.py # Python workspace structure summary
│ ├── query_engine.py # Renders porting summary from active workspace
│ ├── task.py # Task orchestration primitives
│ └── tools.py # Tool port metadata
├── tests/ # unittest-based verification
└── assets/.
├── src/
│ ├── __init__.py
│ ├── commands.py # 命令移植元数据
│ ├── main.py # CLI入口
│ ├── models.py # 数据类:子系统、模块、待办项
│ ├── port_manifest.py # Python工作区结构摘要
│ ├── query_engine.py # 从活跃工作区生成移植摘要
│ ├── task.py # 任务编排原语
│ └── tools.py # 工具移植元数据
├── tests/ # 基于unittest的验证代码
└── assets/Key CLI Commands
核心CLI命令
All commands run via .
python3 -m src.main <subcommand>bash
undefined所有命令通过运行。
python3 -m src.main <子命令>bash
undefinedPrint a human-readable porting summary
打印人类可读的移植摘要
python3 -m src.main summary
python3 -m src.main summary
Print the current Python workspace manifest
打印当前Python工作区清单
python3 -m src.main manifest
python3 -m src.main manifest
List current Python modules/subsystems (paginated)
列出当前Python模块/子系统(分页)
python3 -m src.main subsystems --limit 16
python3 -m src.main subsystems --limit 16
Inspect mirrored command inventory
查看镜像的命令清单
python3 -m src.main commands --limit 10
python3 -m src.main commands --limit 10
Inspect mirrored tool inventory
查看镜像的工具清单
python3 -m src.main tools --limit 10
python3 -m src.main tools --limit 10
Run parity audit against local ignored archive (when present)
针对本地忽略的归档文件运行一致性审计(如果存在)
python3 -m src.main parity-audit
python3 -m src.main parity-audit
Run the full test suite
运行完整测试套件
python3 -m unittest discover -s tests -v
---python3 -m unittest discover -s tests -v
---Core Data Models (src/models.py
)
src/models.py核心数据模型(src/models.py
)
src/models.pyThe dataclasses define the shape of the porting workspace:
python
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class Module:
name: str
status: str # e.g. "ported", "stub", "backlog"
source_path: str
notes: Optional[str] = None
@dataclass
class Subsystem:
name: str
modules: List[Module] = field(default_factory=list)
description: Optional[str] = None
@dataclass
class PortManifest:
subsystems: List[Subsystem] = field(default_factory=list)
backlog: List[str] = field(default_factory=list)
version: str = "0.1.0"数据类定义了移植工作区的结构:
python
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class Module:
name: str
status: str # 例如 "ported", "stub", "backlog"
source_path: str
notes: Optional[str] = None
@dataclass
class Subsystem:
name: str
modules: List[Module] = field(default_factory=list)
description: Optional[str] = None
@dataclass
class PortManifest:
subsystems: List[Subsystem] = field(default_factory=list)
backlog: List[str] = field(default_factory=list)
version: str = "0.1.0"Tools System (src/tools.py
)
src/tools.py工具系统(src/tools.py
)
src/tools.pyTools are the callable units in the agent harness. Each tool entry carries metadata for dispatch:
python
from dataclasses import dataclass
from typing import Callable, Optional, Any, Dict
@dataclass
class Tool:
name: str
description: str
parameters: Dict[str, Any] # JSON-schema style param spec
handler: Optional[Callable] = None # Python callable for this tool工具是Agent框架中的可调用单元,每个工具条目携带分发所需的元数据:
python
from dataclasses import dataclass
from typing import Callable, Optional, Any, Dict
@dataclass
class Tool:
name: str
description: str
parameters: Dict[str, Any] # JSON schema风格的参数规范
handler: Optional[Callable] = None # 该工具对应的Python可调用函数Example: registering a tool
示例:注册一个工具
def read_file_handler(path: str) -> str:
with open(path, "r") as f:
return f.read()
READ_FILE_TOOL = Tool(
name="read_file",
description="Read the contents of a file at the given path.",
parameters={
"path": {"type": "string", "description": "Absolute or relative file path"}
},
handler=read_file_handler,
)
def read_file_handler(path: str) -> str:
with open(path, "r") as f:
return f.read()
READ_FILE_TOOL = Tool(
name="read_file",
description="Read the contents of a file at the given path.",
parameters={
"path": {"type": "string", "description": "Absolute or relative file path"}
},
handler=read_file_handler,
)
Tool registry pattern
工具注册模式
TOOL_REGISTRY: Dict[str, Tool] = {
READ_FILE_TOOL.name: READ_FILE_TOOL,
}
def dispatch_tool(name: str, **kwargs) -> Any:
tool = TOOL_REGISTRY.get(name)
if tool is None:
raise ValueError(f"Unknown tool: {name}")
if tool.handler is None:
raise NotImplementedError(f"Tool '{name}' has no handler yet.")
return tool.handler(**kwargs)
---TOOL_REGISTRY: Dict[str, Tool] = {
READ_FILE_TOOL.name: READ_FILE_TOOL,
}
def dispatch_tool(name: str, **kwargs) -> Any:
tool = TOOL_REGISTRY.get(name)
if tool is None:
raise ValueError(f"Unknown tool: {name}")
if tool.handler is None:
raise NotImplementedError(f"Tool '{name}' has no handler yet.")
return tool.handler(**kwargs)
---Commands System (src/commands.py
)
src/commands.py命令系统(src/commands.py
)
src/commands.pyCommands are higher-level agent actions, distinct from raw tools:
python
from dataclasses import dataclass
from typing import Optional, Callable, Any
@dataclass
class Command:
name: str
description: str
aliases: list
handler: Optional[Callable] = None命令是更高层级的Agent动作,与原始工具不同:
python
from dataclasses import dataclass
from typing import Optional, Callable, Any
@dataclass
class Command:
name: str
description: str
aliases: list
handler: Optional[Callable] = NoneExample command
示例命令
def summarize_handler(context: dict) -> str:
return f"Summarizing {len(context.get('files', []))} files."
SUMMARIZE_COMMAND = Command(
name="summarize",
description="Summarize the current workspace context.",
aliases=["sum", "overview"],
handler=summarize_handler,
)
COMMAND_REGISTRY = {
SUMMARIZE_COMMAND.name: SUMMARIZE_COMMAND,
}
def run_command(name: str, context: dict) -> Any:
cmd = COMMAND_REGISTRY.get(name)
if not cmd:
raise ValueError(f"Unknown command: {name}")
if not cmd.handler:
raise NotImplementedError(f"Command '{name}' not yet implemented.")
return cmd.handler(context)
---def summarize_handler(context: dict) -> str:
return f"Summarizing {len(context.get('files', []))} files."
SUMMARIZE_COMMAND = Command(
name="summarize",
description="Summarize the current workspace context.",
aliases=["sum", "overview"],
handler=summarize_handler,
)
COMMAND_REGISTRY = {
SUMMARIZE_COMMAND.name: SUMMARIZE_COMMAND,
}
def run_command(name: str, context: dict) -> Any:
cmd = COMMAND_REGISTRY.get(name)
if not cmd:
raise ValueError(f"Unknown command: {name}")
if not cmd.handler:
raise NotImplementedError(f"Command '{name}' not yet implemented.")
return cmd.handler(context)
---Task Orchestration (src/task.py
)
src/task.py任务编排(src/task.py
)
src/task.pyTasks wrap a unit of agent work — a goal, a set of tools, and a result:
python
from dataclasses import dataclass, field
from typing import List, Optional, Any
@dataclass
class TaskResult:
success: bool
output: Any
error: Optional[str] = None
@dataclass
class Task:
goal: str
tools: List[str] = field(default_factory=list) # tool names available
context: dict = field(default_factory=dict)
result: Optional[TaskResult] = None
def run(self, dispatcher) -> TaskResult:
"""
dispatcher: callable(tool_name, **kwargs) -> Any
Implement your agent loop here.
"""
try:
# Minimal stub: just report goal received
output = f"Task received: {self.goal}"
self.result = TaskResult(success=True, output=output)
except Exception as e:
self.result = TaskResult(success=False, output=None, error=str(e))
return self.result任务封装了Agent的工作单元:目标、可用工具集和结果:
python
from dataclasses import dataclass, field
from typing import List, Optional, Any
@dataclass
class TaskResult:
success: bool
output: Any
error: Optional[str] = None
@dataclass
class Task:
goal: str
tools: List[str] = field(default_factory=list) # 可用工具名称
context: dict = field(default_factory=dict)
result: Optional[TaskResult] = None
def run(self, dispatcher) -> TaskResult:
"""
dispatcher: callable(tool_name, **kwargs) -> Any
在此处实现你的Agent循环
"""
try:
# 最小存根:仅报告已收到目标
output = f"Task received: {self.goal}"
self.result = TaskResult(success=True, output=output)
except Exception as e:
self.result = TaskResult(success=False, output=None, error=str(e))
return self.resultUsage
使用示例
from src.tools import dispatch_tool
task = Task(
goal="Read README.md and summarize it",
tools=["read_file"],
context={"working_dir": "."},
)
result = task.run(dispatcher=dispatch_tool)
print(result.output)
---from src.tools import dispatch_tool
task = Task(
goal="Read README.md and summarize it",
tools=["read_file"],
context={"working_dir": "."},
)
result = task.run(dispatcher=dispatch_tool)
print(result.output)
---Query Engine (src/query_engine.py
)
src/query_engine.py查询引擎(src/query_engine.py
)
src/query_engine.pyThe query engine renders a porting summary from the active manifest:
python
from src.port_manifest import build_manifest
from src.query_engine import render_summary
manifest = build_manifest()
summary = render_summary(manifest)
print(summary)You can also invoke it from the CLI:
bash
python3 -m src.main summary查询引擎从活跃清单生成移植摘要:
python
from src.port_manifest import build_manifest
from src.query_engine import render_summary
manifest = build_manifest()
summary = render_summary(manifest)
print(summary)你也可以通过CLI调用:
bash
python3 -m src.main summaryPort Manifest (src/port_manifest.py
)
src/port_manifest.py移植清单(src/port_manifest.py
)
src/port_manifest.pyBuild and inspect the current workspace manifest programmatically:
python
from src.port_manifest import build_manifest
manifest = build_manifest()
for subsystem in manifest.subsystems:
print(f"[{subsystem.name}]")
for module in subsystem.modules:
print(f" {module.name}: {module.status}")
print("Backlog:", manifest.backlog)以编程方式构建和查看当前工作区清单:
python
from src.port_manifest import build_manifest
manifest = build_manifest()
for subsystem in manifest.subsystems:
print(f"[{subsystem.name}]")
for module in subsystem.modules:
print(f" {module.name}: {module.status}")
print("Backlog:", manifest.backlog)Adding a New Tool
添加新工具
- Define a handler function in .
src/tools.py - Create a dataclass instance.
Tool - Register it in .
TOOL_REGISTRY - Write a test in .
tests/
python
undefined- 在中定义处理函数
src/tools.py - 创建数据类实例
Tool - 在中注册
TOOL_REGISTRY - 在目录下编写测试
tests/
python
undefinedsrc/tools.py
src/tools.py
def list_dir_handler(path: str):
import os
return os.listdir(path)
LIST_DIR_TOOL = Tool(
name="list_dir",
description="List files in a directory.",
parameters={"path": {"type": "string"}},
handler=list_dir_handler,
)
TOOL_REGISTRY["list_dir"] = LIST_DIR_TOOL
---def list_dir_handler(path: str):
import os
return os.listdir(path)
LIST_DIR_TOOL = Tool(
name="list_dir",
description="List files in a directory.",
parameters={"path": {"type": "string"}},
handler=list_dir_handler,
)
TOOL_REGISTRY["list_dir"] = LIST_DIR_TOOL
---Adding a New Command
添加新命令
python
undefinedpython
undefinedsrc/commands.py
src/commands.py
def lint_handler(context: dict) -> str:
files = context.get("files", [])
return f"Linting {len(files)} files (stub)."
LINT_COMMAND = Command(
name="lint",
description="Lint the current workspace files.",
aliases=["check"],
handler=lint_handler,
)
COMMAND_REGISTRY["lint"] = LINT_COMMAND
---def lint_handler(context: dict) -> str:
files = context.get("files", [])
return f"Linting {len(files)} files (stub)."
LINT_COMMAND = Command(
name="lint",
description="Lint the current workspace files.",
aliases=["check"],
handler=lint_handler,
)
COMMAND_REGISTRY["lint"] = LINT_COMMAND
---Running Tests
运行测试
bash
undefinedbash
undefinedRun all tests with verbose output
运行所有测试并输出详细信息
python3 -m unittest discover -s tests -v
python3 -m unittest discover -s tests -v
Run a specific test file
运行指定测试文件
python3 -m unittest tests.test_tools -v
Example test pattern:
```pythonpython3 -m unittest tests.test_tools -v
示例测试模式:
```pythontests/test_tools.py
tests/test_tools.py
import unittest
from src.tools import dispatch_tool
import tempfile, os
class TestReadFileTool(unittest.TestCase):
def test_read_file(self):
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
f.write("hello clawd")
path = f.name
try:
result = dispatch_tool("read_file", path=path)
self.assertEqual(result, "hello clawd")
finally:
os.unlink(path)
if name == "main":
unittest.main()
---import unittest
from src.tools import dispatch_tool
import tempfile, os
class TestReadFileTool(unittest.TestCase):
def test_read_file(self):
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
f.write("hello clawd")
path = f.name
try:
result = dispatch_tool("read_file", path=path)
self.assertEqual(result, "hello clawd")
finally:
os.unlink(path)
if name == "main":
unittest.main()
---Parity Audit
一致性审计
When a local ignored archive of the original snapshot is present, run:
bash
python3 -m src.main parity-auditThis compares the current Python workspace surface against the archived root-entry file surface, subsystem names, and command/tool inventories, reporting gaps.
如果本地存在原始快照的忽略归档文件,可以运行:
bash
python3 -m src.main parity-audit该命令会将当前Python工作区的对外接口与归档的根入口文件接口、子系统名称、命令/工具清单进行对比,报告缺失项。
Common Patterns
常用模式
Chaining tools in a task loop
在任务循环中链式调用工具
python
from src.tools import dispatch_tool
from src.task import Task
task = Task(
goal="Read and list files",
tools=["read_file", "list_dir"],
context={"working_dir": "."},
)python
from src.tools import dispatch_tool
from src.task import Task
task = Task(
goal="Read and list files",
tools=["read_file", "list_dir"],
context={"working_dir": "."},
)Manual tool chain (before full agent loop is implemented)
手动工具链(完整Agent循环实现前使用)
files = dispatch_tool("list_dir", path=".")
for fname in files[:3]:
content = dispatch_tool("read_file", path=fname)
print(f"--- {fname} ---\n{content[:200]}")
undefinedfiles = dispatch_tool("list_dir", path=".")
for fname in files[:3]:
content = dispatch_tool("read_file", path=fname)
print(f"--- {fname} ---\n{content[:200]}")
undefinedUsing the manifest in automation
在自动化中使用清单
python
from src.port_manifest import build_manifest
def unported_modules():
manifest = build_manifest()
stubs = []
for sub in manifest.subsystems:
for mod in sub.modules:
if mod.status != "ported":
stubs.append((sub.name, mod.name, mod.status))
return stubs
for subsystem, module, status in unported_modules():
print(f"{subsystem}/{module} → {status}")python
from src.port_manifest import build_manifest
def unported_modules():
manifest = build_manifest()
stubs = []
for sub in manifest.subsystems:
for mod in sub.modules:
if mod.status != "ported":
stubs.append((sub.name, mod.name, mod.status))
return stubs
for subsystem, module, status in unported_modules():
print(f"{subsystem}/{module} → {status}")Troubleshooting
问题排查
| Symptom | Fix |
|---|---|
| Run commands from the repo root, not inside |
| The tool is registered but the Python handler hasn't been written yet — implement |
| The local ignored archive must be present at the expected path; see |
| Tests not discovered | Ensure test files are named |
| Import errors after adding a module | Add |
| 症状 | 解决方案 |
|---|---|
| 在仓库根目录运行命令,不要在 |
| 该工具已注册但尚未编写Python处理函数——在 |
| 本地忽略的归档文件必须放在预期路径下;可查看 |
| 测试未被发现 | 确保测试文件命名为 |
| 添加模块后出现导入错误 | 为所有新的包子目录添加 |
Key Links
关键链接
- Repository: https://github.com/instructkr/clawd-code
- oh-my-codex (OmX): https://github.com/Yeachan-Heo/oh-my-codex
- Related essay: Is legal the same as legitimate?
- Not affiliated with Anthropic.
- 仓库地址: https://github.com/instructkr/clawd-code
- oh-my-codex (OmX): https://github.com/Yeachan-Heo/oh-my-codex
- 相关文章: 合法等同于合理吗?
- 与Anthropic无任何关联