clawd-code-python-port

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

clawd-code Python Port

clawd-code Python移植版

Skill by ara.so — Daily 2026 Skills collection.
技能由ara.so开发——属于2026年度技能合集。

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
undefined
bash
undefined

Clone the repository

克隆仓库

(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
undefined

Print 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

The 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

Tools 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

Commands 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] = None

Example 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

Tasks 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.result

Usage

使用示例

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

The 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 summary

Port Manifest (
src/port_manifest.py
)

移植清单(
src/port_manifest.py

Build 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

添加新工具

  1. Define a handler function in
    src/tools.py
    .
  2. Create a
    Tool
    dataclass instance.
  3. Register it in
    TOOL_REGISTRY
    .
  4. Write a test in
    tests/
    .
python
undefined
  1. src/tools.py
    中定义处理函数
  2. 创建
    Tool
    数据类实例
  3. TOOL_REGISTRY
    中注册
  4. tests/
    目录下编写测试
python
undefined

src/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
undefined
python
undefined

src/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
undefined
bash
undefined

Run 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:

```python
python3 -m unittest tests.test_tools -v

示例测试模式:

```python

tests/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-audit
This 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]}")
undefined
files = dispatch_tool("list_dir", path=".") for fname in files[:3]: content = dispatch_tool("read_file", path=fname) print(f"--- {fname} ---\n{content[:200]}")
undefined

Using 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

问题排查

SymptomFix
ModuleNotFoundError: src
Run commands from the repo root, not inside
src/
NotImplementedError: Tool 'x' has no handler
The tool is registered but the Python handler hasn't been written yet — implement
handler
in
tools.py
parity-audit
does nothing
The local ignored archive must be present at the expected path; see
port_manifest.py
for the expected location
Tests not discoveredEnsure test files are named
test_*.py
and located in
tests/
Import errors after adding a moduleAdd
__init__.py
to any new package subdirectory

症状解决方案
ModuleNotFoundError: src
在仓库根目录运行命令,不要在
src/
目录内执行
NotImplementedError: Tool 'x' has no handler
该工具已注册但尚未编写Python处理函数——在
tools.py
中实现
handler
即可
parity-audit
无任何输出
本地忽略的归档文件必须放在预期路径下;可查看
port_manifest.py
了解预期位置
测试未被发现确保测试文件命名为
test_*.py
且位于
tests/
目录下
添加模块后出现导入错误为所有新的包子目录添加
__init__.py
文件

Key Links

关键链接