python写一个类似claude code和codex的命令行自主智能体代码-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python写一个类似claude code和codex的命令行自主智能体代码 - 真流式输出（SSE），模型思考过程实时打字机效果显示 - 类 Claude Code 的终端 UI：工具调用/结果面板、步骤分隔线、圆角框、旋转动画 - REPL 斜杠命令：/mode /clear /compact /skills /tasks /help /exit - 其余能力保持不变：Plan/Build 模式、文件/命令/Python 执行、技能系统、定时任务调度、自动+手动上下文压缩、系

代码标签： python 类似claude code codex 命令行自主智能体代码

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
codex_agent.py
一个类似 Claude Code / Codex CLI 的命令行自主智能体（单文件实现）。

新特性：
  - 真流式输出（SSE），模型思考过程实时打字机效果显示
  - 类 Claude Code 的终端 UI：工具调用/结果面板、步骤分隔线、圆角框、旋转动画
  - REPL 斜杠命令：/mode /clear /compact /skill /tasks /help /exit
  - 技能系统升级：技能 = 自然语言描述 -> 模型自动生成的"指南文档"（可选内嵌可执行代码），
    支持在 chat 中动态创建 / 修改 / 删除 / 调用技能，无需用户手写代码
  - 其余能力保持不变：Plan/Build 模式、文件/命令/Python 执行、定时任务调度、
    自动+手动上下文压缩、系统环境信息注入

依赖：仅 Python 标准库。

用法示例：
  export OPENAI_API_KEY=sk-xxxx
  export OPENAI_BASE_URL=https://your-endpoint/v1
  export OPENAI_MODEL=gpt-4o-mini

  python codex_agent.py run --goal "统计当前目录下代码行数" --mode build
  python codex_agent.py chat --mode build
  python codex_agent.py skill list
  python codex_agent.py skill create --name deploy_guide --requirement "如何把一个静态网站部署到服务器"
  python codex_agent.py skill update --name deploy_guide --instruction "补充 nginx 配置示例"
  python codex_agent.py schedule add --name daily --when "daily:09:00" --goal "..."
  python codex_agent.py daemon
  python codex_agent.py config show
"""

import os
import sys
import re
import json
import time
import uuid
import base64
import shutil
import socket
import getpass
import platform
import argparse
import textwrap
import traceback
import threading
import unicodedata
import subprocess
import datetime
import urllib.request
import urllib.error

# ============================================================
# 全局路径
# ============================================================
HOME_DIR = os.path.join(os.path.expanduser("~"), ".codex_agent")
SKILLS_DIR = os.path.join(HOME_DIR, "skills")
LOGS_DIR = os.path.join(HOME_DIR, "logs")
TASKS_PATH = os.path.join(HOME_DIR, "tasks.json")
CONFIG_PATH = os.path.join(HOME_DIR, "config.json")


def ensure_dirs():
    os.makedirs(HOME_DIR, exist_ok=True)
    os.makedirs(SKILLS_DIR, exist_ok=True)
    os.makedirs(LOGS_DIR, exist_ok=True)


# ============================================================
# UI / 终端美化模块
# ============================================================
class C:
    RESET = BOLD = DIM = ITALIC = UNDERLINE = ""
    BLACK = RED = GREEN = YELLOW = BLUE = MAGENTA = CYAN = WHITE = GRAY = ""
    BRIGHT_RED = BRIGHT_GREEN = BRIGHT_YELLOW = BRIGHT_BLUE = ""
    BRIGHT_MAGENTA = BRIGHT_CYAN = BRIGHT_WHITE = ""


def _enable_color():
    C.RESET = "\033[0m"
    C.BOLD = "\033[1m"
    C.DIM = "\033[2m"
    C.ITALIC = "\033[3m"
    C.UNDERLINE = "\033[4m"
    C.BLACK = "\033[30m"; C.RED = "\033[31m"; C.GREEN = "\033[32m"
    C.YELLOW = "\033[33m"; C.BLUE = "\033[34m"; C.MAGENTA = "\033[35m"
    C.CYAN = "\033[36m"; C.WHITE = "\033[37m"; C.GRAY = "\033[90m"
    C.BRIGHT_RED = "\033[91m"; C.BRIGHT_GREEN = "\033[92m"
    C.BRIGHT_YELLOW = "\033[93m"; C.BRIGHT_BLUE = "\033[94m"
    C.BRIGHT_MAGENTA = "\033[95m"; C.BRIGHT_CYAN = "\033[96m"
    C.BRIGHT_WHITE = "\033[97m"


def init_color(force=None):
    color_on = force
    if color_on is None:
        color_on = sys.stdout.isatty() and os.environ.get("NO_COLOR") is None
    if color_on:
        _enable_color()
        if platform.system() == "Windows":
            os.system("")  # 开启 Windows 终端 ANSI 转义支持
    return color_on


ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")


def strip_ansi(s):
    return ANSI_RE.sub("", s or "")


def display_width(s):
    s = strip_ansi(s)
    w = 0
    for ch in s:
        w += 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
    return w


def draw_box(lines, color=None, padding=1, min_width=20):
    color = color or ""
    reset = C.RESET if color else ""
    widths = [display_width(l) for l in lines] or [0]
    inner = max(max(widths), min_width) + padding * 2
    top = "╭" + "─" * inner + "╮"
    bottom = "╰" + "─" * inner + "╯"
    out = [f"{color}{top}{reset}"]
    for l, w in zip(lines, widths):
        right_pad = max(inner - w - padding, 0)
        out.append(f"{color}│{reset}" + " " * padding + l + " " * right_pad + f"{color}│{reset}")
    out.append(f"{color}{bottom}{reset}")
    return "\n".join(out)


class Spinner:
    FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]

    def __init__(self, message="Thinking"):
        self.message = message
        self._stop_evt = threading.Event()
        self._thread = None
        self._active = False

    def start(self):
        if self._active:
            return
        self._active = True
        if not sys.stdout.isatty():
            return
        self._stop_evt.clear()
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def _run(self):
        i = 0
        while not self._stop_evt.is_set():
            frame = self.FRAMES[i % len(self.FRAMES)]
            sys.stdout.write(f"\r{C.CYAN}{frame}{C.RESET} {C.DIM}{self.message}…{C.RESET}")
            sys.stdout.flush()
            i += 1
            time.sleep(0.08)
        clear_len = display_width(self.message) + 6
        sys.stdout.write("\r" + " " * clear_len + "\r")
        sys.stdout.flush()

    def stop(self):
        if not self._active:
            return
        self._active = False
        if self._thread:
            self._stop_evt.set()
            self._thread.join(timeout=1)
            self._thread = None


def summarize_action_input(action, action_input):
    action_input = action_input or {}
    try:
        if action == "read_file":
            return action_input.get("path", "")
        if action == "write_file":
            c = action_input.get("content", "") or ""
            tag = "追加" if action_input.get("append") else "写入"
            return f'{action_input.get("path","")} ({tag} {len(c)} 字符)'
        if action == "list_dir":
            return action_input.get("path", ".")
        if action == "run_command":
            return action_input.get("cmd", "")
        if action == "run_python":
            code = action_input.get("code", "") or ""
            n = code.count("\n") + 1
            return f"<{n} 行代码>"
        if action == "create_skill":
            req = (action_input.get("requirement", "") or "")[:60]
            return f'{action_input.get("name","")}: {req}'
        if action == "update_skill":
            ins = (action_input.get("instruction", "") or "")[:60]
            return f'{action_input.get("name","")}: {ins}'
        if action == "delete_skill":
            return action_input.get("name", "")
        if action == "use_skill":
            return f'{action_input.get("name","")} {action_input.get("args", {})}'
        if action in ("list_skills", "list_tasks", "compress_context"):
            return ""
        if action == "create_task":
            return f'{action_input.get("name","")} [{action_input.get("schedule","")}]'
        if action == "cancel_task":
            return action_input.get("name", "")
        if action == "ask_user":
            return (action_input.get("question&quo.........完整代码请登录后点击上方下载按钮下载查看