从零手搓一个 Claude Code：用 Python 打造你自己的终端 AI 编程助手

你：帮我写个 hello world
AI：好的，这是代码 print("hello world")
你：（复制粘贴到编辑器，手动保存，手动运行）

你：帮我写个 hello world
AI：（自动创建 hello.py → 写入代码 → 运行 → 告诉你结果）

# 创建项目目录
mkdir magiccode && cd magiccode

# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install openai rich prompt_toolkit

export OPENAI_API_KEY="sk-your-key-here"

#!/usr/bin/env python3
"""MagicCode v1 — 20 行实现终端 AI 助手"""
from openai import OpenAI

client = OpenAI()  # 自动读取 OPENAI_API_KEY
history = [{"role": "system", "content": "你是 MagicCode，一个终端 AI 编程助手。用中文回答，代码用英文。"}]

print("🪄 MagicCode v1 - 输入 exit 退出")
while True:
    user_input = input("\n你 > ")
    if user_input.strip().lower() in ("exit", "quit"):
        break

    history.append({"role": "user", "content": user_input})

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=history,
    )

    reply = response.choices[0].message.content
    history.append({"role": "assistant", "content": reply})
    print(f"\n🤖 {reply}")

python v1_basic.py

#!/usr/bin/env python3
"""MagicCode v2 — 流式输出版"""
from openai import OpenAI

client = OpenAI()
history = [{"role": "system", "content": "你是 MagicCode，一个终端 AI 编程助手。用中文回答，代码用英文。简洁专业。"}]

print("🪄 MagicCode v2 (streaming) - 输入 exit 退出")
while True:
    user_input = input("\n你 > ")
    if user_input.strip().lower() in ("exit", "quit"):
        break

    history.append({"role": "user", "content": user_input})

    # 关键改动：加上 stream=True
    print("\n🤖 ", end="", flush=True)
    full_reply = ""

    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=history,
        stream=True,  # ← 开启流式输出
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
            full_reply += delta

    print()  # 换行
    history.append({"role": "assistant", "content": full_reply})

#!/usr/bin/env python3
"""MagicCode v3 — Rich Markdown 渲染版"""
from openai import OpenAI
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from rich.live import Live

client = OpenAI()
console = Console()
history = [{"role": "system", "content": "你是 MagicCode，一个终端 AI 编程助手。用 Markdown 格式回答。"}]

console.print(Panel(
    "🪄 [bold cyan]MagicCode v3[/] - 终端 AI 编程助手\n输入 exit 退出",
    border_style="cyan"
))

while True:
    console.print()
    user_input = console.input("[bold green]你 >[/] ")
    if user_input.strip().lower() in ("exit", "quit"):
        break

    history.append({"role": "user", "content": user_input})

    # 流式输出 + 实时 Markdown 渲染
    full_reply = ""
    stream = client.chat.completions.create(
        model="gpt-4o", messages=history, stream=True,
    )
    with Live(console=console, refresh_per_second=8) as live:
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                full_reply += delta
                live.update(Panel(
                    Markdown(full_reply),
                    title="🤖 MagicCode",
                    border_style="blue",
                ))

    history.append({"role": "assistant", "content": full_reply})

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取文件内容。支持任意文本文件。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件路径"
                    }
                },
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "将内容写入文件。文件不存在会自动创建。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"},
                    "content": {"type": "string", "description": "完整文件内容"},
                },
                "required": ["path", "content"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "执行终端命令。有 30 秒超时。",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "shell 命令"}
                },
                "required": ["command"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_files",
            "description": "列出目录结构（自动忽略 node_modules、.git 等）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "目录路径", "default": "."},
                },
                "required": [],
            },
        },
    },
]

import os
import subprocess

def execute_tool(name: str, params: dict) -> str:
    """执行工具调用，返回字符串结果"""
    try:
        if name == "read_file":
            with open(params["path"], "r", encoding="utf-8") as f:
                content = f.read()
            lines = content.split("\n")
            # 加行号，方便 AI 精确定位
            numbered = "\n".join(
                f"{i+1:4d} | {line}" for i, line in enumerate(lines)
            )
            return f"📄 {params['path']}（{len(lines)} 行）\n{numbered}"

        elif name == "write_file":
            path = params["path"]
            os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
            with open(path, "w", encoding="utf-8") as f:
                f.write(params["content"])
            return f"✅ 已写入 {path}（{len(params['content'])} 字符）"

        elif name == "run_command":
            cmd = params["command"]
            # 🛡️ 安全检查：拒绝危险命令
            dangerous = ["rm -rf /", "mkfs", "dd if=", "> /dev/sd"]
            if any(d in cmd for d in dangerous):
                return "❌ 拒绝执行危险命令"
            result = subprocess.run(
                cmd, shell=True, capture_output=True,
                text=True, timeout=30
            )
            output = result.stdout
            if result.stderr:
                output += "\n--- stderr ---\n" + result.stderr
            return output.strip() or "（命令执行成功，无输出）"

        elif name == "list_files":
            path = params.get("path", ".")
            entries = sorted(os.listdir(path))
            result = []
            for entry in entries:
                full = os.path.join(path, entry)
                icon = "📁" if os.path.isdir(full) else "📄"
                result.append(f"{icon} {entry}")
            return "\n".join(result) or "目录为空"

    except Exception as e:
        return f"❌ {type(e).__name__}: {e}"

def chat(user_input: str):
    """Agentic Loop：AI 自主决策循环"""
    history.append({"role": "user", "content": user_input})

    while True:
        # 1️⃣ 调用 LLM（携带工具定义）
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=history,
            tools=TOOLS,          # ← 关键：传入工具列表
        )
        message = response.choices[0].message

        # 2️⃣ 把 AI 的完整回复存入历史
        history.append(message)

        # 3️⃣ 处理文本回复
        if message.content:
            console.print(Panel(Markdown(message.content), title="🤖 MagicCode"))

        # 4️⃣ 如果没有工具调用，说明任务完成，退出循环
        if not message.tool_calls:
            break

        # 5️⃣ 执行每个工具调用，把结果反馈给 AI
        for tool_call in message.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)

            console.print(f"  🔧 {name}({args})")
            result = execute_tool(name, args)

            # 工具结果以 role="tool" 消息反馈
            history.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })
        # → 回到 while 循环顶部，AI 继续思考

AI 第 1 轮回复：
  content: "好的，让我先看看项目结构"
  tool_calls: [list_files("."), read_file("package.json")]

→ 执行工具，把结果反馈给 AI

AI 第 2 轮回复：
  content: "我看到这是个 Node.js 项目，让我修改..."
  tool_calls: [write_file("index.js", ...)]

→ 执行工具，把结果反馈给 AI

AI 第 3 轮回复：
  content: "修改完成，让我运行测试验证"
  tool_calls: [run_command("npm test")]

→ 执行工具，把结果反馈给 AI

AI 第 4 轮回复：
  content: "✅ 所有测试通过！我做了以下改动..."
  tool_calls: null（没有工具调用 → 循环结束）

[
    # 系统消息
    {"role": "system", "content": "你是 MagicCode..."},

    # 用户消息
    {"role": "user", "content": "帮我写个 hello world"},

    # AI 回复（包含工具调用）
    {
        "role": "assistant",
        "content": "好的，我来创建文件",
        "tool_calls": [{
            "id": "call_xxx",
            "type": "function",
            "function": {
                "name": "write_file",
                "arguments": '{"path":"hello.py","content":"print(\'hello\')"}'
            }
        }]
    },

    # 工具执行结果
    {
        "role": "tool",
        "tool_call_id": "call_xxx",
        "content": "✅ 已写入 hello.py（16 字符）"
    },

    # AI 继续回复
    {
        "role": "assistant",
        "content": "文件已创建，现在运行它",
        "tool_calls": [{
            "id": "call_yyy",
            "type": "function",
            "function": {"name": "run_command", "arguments": '{"command":"python hello.py"}'}
        }]
    },

    # ... 循环继续
]

#!/usr/bin/env python3
"""
MagicCode - 从零手搓的终端 AI 编程助手
功能：工具调用 | Markdown 渲染 | Agentic Loop
"""
import os
import json
import glob
import subprocess
from openai import OpenAI
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel

# ========== 配置 ==========
MODEL = os.getenv("MAGIC_MODEL", "gpt-4o")
client = OpenAI()  # 自动读取 OPENAI_API_KEY

SYSTEM_PROMPT = """你是 MagicCode，一个强大的终端 AI 编程助手。

## 核心能力
你拥有以下工具：
- read_file: 读取文件内容（带行号）
- write_file: 写入文件（自动创建目录）
- edit_file: 精确替换文件中的指定内容
- run_command: 执行终端命令（30s 超时）
- list_files: 列出目录结构
- search_code: 在代码中搜索关键词

## 工作原则
1. 修改文件前必须先 read_file 查看当前内容
2. 复杂任务分步完成，每步验证
3. 不执行危险命令（rm -rf、格式化等）
4. 用中文回答，代码用英文，Markdown 格式输出"""

# ========== 工具定义 ==========
def _fn(name, desc, params, required):
    return {"type": "function", "function": {
        "name": name, "description": desc,
        "parameters": {"type": "object", "properties": params, "required": required},
    }}

TOOLS = [
    _fn("read_file", "读取文件内容，返回带行号的文本。",
        {"path": {"type": "string", "description": "文件路径"}}, ["path"]),
    _fn("write_file", "将内容写入文件，自动创建不存在的目录。",
        {"path": {"type": "string", "description": "文件路径"},
         "content": {"type": "string", "description": "完整文件内容"}}, ["path", "content"]),
    _fn("edit_file", "精确编辑：将文件中的 old_text 替换为 new_text。",
        {"path": {"type": "string", "description": "文件路径"},
         "old_text": {"type": "string", "description": "要替换的原文本"},
         "new_text": {"type": "string", "description": "替换后的新文本"}}, ["path", "old_text", "new_text"]),
    _fn("run_command", "执行 shell 命令，30 秒超时。",
        {"command": {"type": "string", "description": "命令"}}, ["command"]),
    _fn("list_files", "递归列出目录结构（最多 3 层，自动忽略 .git 等）。",
        {"path": {"type": "string", "description": "目录路径"}}, []),
    _fn("search_code", "在项目中搜索包含关键词的代码行。",
        {"pattern": {"type": "string", "description": "搜索关键词"},
         "path": {"type": "string", "description": "搜索目录"}}, ["pattern"]),
]

IGNORED_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build"}

# ========== 工具执行 ==========
def execute_tool(name: str, params: dict) -> str:
    try:
        if name == "read_file":
            with open(params["path"], "r", encoding="utf-8", errors="replace") as f:
                content = f.read()
            lines = content.split("\n")
            numbered = "\n".join(f"{i+1:4d} | {line}" for i, line in enumerate(lines))
            return f"📄 {params['path']}（{len(lines)} 行）\n{numbered}"

        elif name == "write_file":
            path = params["path"]
            os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
            with open(path, "w", encoding="utf-8") as f:
                f.write(params["content"])
            return f"✅ 已写入 {path}（{len(params['content'])} 字符）"

        elif name == "edit_file":
            path = params["path"]
            with open(path, "r", encoding="utf-8") as f:
                content = f.read()
            if params["old_text"] not in content:
                return "❌ 未找到要替换的文本"
            new_content = content.replace(params["old_text"], params["new_text"], 1)
            with open(path, "w", encoding="utf-8") as f:
                f.write(new_content)
            return f"✅ 已编辑 {path}"

        elif name == "run_command":
            cmd = params["command"]
            dangerous = ["rm -rf /", "mkfs", "dd if=", "> /dev/sd"]
            if any(d in cmd for d in dangerous):
                return "❌ 拒绝执行危险命令"
            result = subprocess.run(
                cmd, shell=True, capture_output=True, text=True, timeout=30
            )
            output = result.stdout
            if result.stderr:
                output += "\n--- stderr ---\n" + result.stderr
            return output.strip() or "（无输出）"

        elif name == "list_files":
            path = params.get("path", ".")
            lines = []
            def walk(d, prefix="", depth=0):
                if depth >= 3: return
                try: entries = sorted(os.listdir(d))
                except PermissionError: return
                for e in entries:
                    full = os.path.join(d, e)
                    if e in IGNORED_DIRS or e.startswith("."): continue
                    if os.path.isdir(full):
                        lines.append(f"{prefix}📁 {e}/")
                        walk(full, prefix + "  ", depth + 1)
                    else:
                        lines.append(f"{prefix}📄 {e}")
            walk(path)
            return "\n".join(lines[:200]) or "目录为空"

        elif name == "search_code":
            pattern = params["pattern"]
            path = params.get("path", ".")
            matches = []
            for fp in glob.glob(os.path.join(path, "**", "*"), recursive=True):
                if any(d in fp for d in IGNORED_DIRS) or not os.path.isfile(fp):
                    continue
                try:
                    with open(fp, "r", encoding="utf-8", errors="replace") as f:
                        for i, line in enumerate(f, 1):
                            if pattern.lower() in line.lower():
                                matches.append(f"{fp}:{i}: {line.rstrip()}")
                                if len(matches) >= 50: break
                except OSError: continue
                if len(matches) >= 50: break
            return "\n".join(matches) or f"未找到 '{pattern}'"

    except Exception as e:
        return f"❌ {type(e).__name__}: {e}"

# ========== Agentic Loop ==========
class MagicCode:
    def __init__(self):
        self.console = Console()
        self.history = [{"role": "system", "content": SYSTEM_PROMPT}]

    def chat(self, user_input: str):
        self.history.append({"role": "user", "content": user_input})
        tool_count = 0

        while True:
            response = client.chat.completions.create(
                model=MODEL, messages=self.history, tools=TOOLS,
            )
            message = response.choices[0].message
            self.history.append(message)

            # 显示文本回复
            if message.content:
                self.console.print(Panel(
                    Markdown(message.content),
                    title="🤖 MagicCode", border_style="blue", padding=(1, 2),
                ))

            # 没有工具调用 → 任务完成
            if not message.tool_calls:
                break

            # 执行每个工具调用
            for tc in message.tool_calls:
                tool_count += 1
                name = tc.function.name
                args = json.loads(tc.function.arguments)
                info = json.dumps(args, ensure_ascii=False)
                if len(info) > 120: info = info[:120] + "..."
                self.console.print(f"  [yellow]🔧 [{tool_count}] {name}[/] [dim]{info}[/]")

                result = execute_tool(name, args)
                preview = result[:100].replace("\n", " ")
                self.console.print(f"  [green]  ✓[/] [dim]{preview}[/]")

                self.history.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result,
                })

            if tool_count > 20:
                self.console.print("[red]⚠️ 已达到工具调用上限（20次）[/]")
                break

    def run(self):
        self.console.print(Panel(
            "[bold cyan]🪄 MagicCode[/] — 你的终端 AI 编程助手\n\n"
            "  [green]能力[/]：读写文件 | 执行命令 | 搜索代码 | 精确编辑\n"
            "  [green]命令[/]：exit 退出 | clear 清空历史",
            border_style="cyan", padding=(1, 2),
        ))
        self.console.print(f"  [dim]📂 {os.getcwd()}[/]")
        self.console.print(f"  [dim]🧠 {MODEL}[/]\n")

        while True:
            try:
                user_input = self.console.input("[bold green]✦ 你 >[/] ")
                cmd = user_input.strip().lower()
                if cmd in ("exit", "quit"): break
                elif cmd == "clear":
                    self.history = [{"role": "system", "content": SYSTEM_PROMPT}]
                    self.console.print("[dim]🗑️ 历史已清空[/]")
                    continue
                elif not cmd: continue
                self.chat(user_input)
                self.console.print()
            except KeyboardInterrupt:
                self.console.print("\n[cyan]👋 再见！[/]")
                break

if __name__ == "__main__":
    MagicCode().run()

python magic.py

def execute_tool_with_confirm(name, params):
    # 读操作直接执行
    if name in ("read_file", "list_files", "search_code"):
        return execute_tool(name, params)

    # 写操作需要用户确认
    console.print(f"[yellow]⚠️ {name}({params})[/]")
    confirm = console.input("[bold]允许执行？(y/n) [/]")
    if confirm.lower() == "y":
        return execute_tool(name, params)
    return "用户拒绝了此操作"

def load_project_context():
    """读取项目配置文件作为上下文"""
    context = ""
    for name in ["CLAUDE.md", "AGENTS.md", "README.md"]:
        if os.path.exists(name):
            with open(name, "r") as f:
                context += f"\n\n--- {name} ---\n{f.read()}"
    return context

# 在 SYSTEM_PROMPT 后追加项目上下文
project_ctx = load_project_context()
if project_ctx:
    SYSTEM_PROMPT += f"\n\n## 项目上下文\n{project_ctx}"

import json

HISTORY_FILE = ".magiccode_history.json"

def save_history(history):
    with open(HISTORY_FILE, "w") as f:
        json.dump(history, f, ensure_ascii=False, default=str)

def load_history():
    if os.path.exists(HISTORY_FILE):
        with open(HISTORY_FILE, "r") as f:
            return json.load(f)
    return []

from openai import OpenAI

# 使用 DeepSeek
client = OpenAI(api_key="your-key", base_url="https://api.deepseek.com/v1")

# 使用 Qwen（通义千问）
client = OpenAI(api_key="your-key", base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")

# 使用本地 Ollama
client = OpenAI(api_key="ollama", base_url="http://localhost:11434/v1")

total_input_tokens = 0
total_output_tokens = 0

# 在每次 API 调用后累计
total_input_tokens += response.usage.prompt_tokens
total_output_tokens += response.usage.completion_tokens

# 在 exit 时显示
console.print(f"[dim]本次会话 Token 用量：输入 {total_input_tokens} | 输出 {total_output_tokens}[/]")