A tiny codex demo in 200 lines with zero dependencies

nano_codex.py

#!/usr/bin/env python3
# released into the public domain
import os, sys, json, ssl, subprocess, urllib.request, readline

SYSTEM_PROMPT = """You are NanoCodex, an AI assistant running inside a local workspace. Help the user operate their computer, edit code, and complete tasks.
You run in a macOS or Linux environment. Use shell commands exactly as they are used in a terminal. Before running commands, confirm the current working directory. 
For high-risk commands, ask the user for confirmation first.
When working in a codebase, follow common AI coding tool conventions:
    If files such as Agents.md, Claude.md, or Gemini.md exist, prioritize reading one of them fully (reading any one successfully is sufficient).
    If hidden directories such as .cursor, .claude, or .gemini exist and contain rules or skills subdirectories, consult those documents as needed.
Do not modify user files or code unless the user explicitly asks you to do so. For complex requests, first analyze the task, provide a clear plan, then execute step by step.
Think and reply in the user's language. Output plain text only (no Markdown or tables). Keep responses concise and clear."""

def tool_read_file(args: dict) -> dict:
    path = args.get("path")
    if not path:
        return {"ok": False, "error": "path is required"}
    try:
        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()
    except Exception as e:
        return {"ok": False, "error": str(e)}
    if not lines:
        return {"ok": True, "content": f"File: {path}\n(empty file)\n"}
    
    offset = max(1, int(args.get("offset", 1)))
    limit = max(1, int(args.get("limit", 1000)))
    max_len = int(args.get("max_line_length", 500))
    end = min(offset - 1 + limit, len(lines))
    width = len(str(end or 1))
    out = [f"File: {path}\nLines: {offset}-{end} of {len(lines)}\n\n"]
    for i in range(offset - 1, end):
        raw = lines[i].rstrip("\n")
        text = raw[:max_len] + " ...(truncated)" if len(raw) > max_len else raw
        out.append(f"{i + 1:{width}} | {text}\n")
    return {"ok": True, "content": "".join(out)}

def tool_write_file(args: dict) -> dict:
    path = args.get("path")
    if not path:
        return {"ok": False, "error": "path is required"}
    try:
        p = os.path.abspath(os.path.expanduser(path))
        os.makedirs(os.path.dirname(p), exist_ok=True)
        if not args.get("overwrite", True) and os.path.exists(p):
            return {"ok": False, "error": f"file already exists (overwrite=false): {p}"}
        with open(p, "w", encoding="utf-8") as f:
            f.write(args.get("content", ""))
        return {"ok": True, "path": p}
    except Exception as e:
        return {"ok": False, "error": str(e)}

def tool_edit_file(args: dict) -> dict:
    path = args.get("path")
    if not path:
        return {"ok": False, "error": "path is required"}
    try:
        p = os.path.abspath(os.path.expanduser(path))
        with open(p, "r", encoding="utf-8") as f:
            content = f.read()
    except Exception as e:
        return {"ok": False, "error": str(e)}

    old_str = args.get("old_str")
    if old_str is None:
        return {"ok": False, "error": "old_str is required for replace"}
    count = content.count(old_str)
    if count == 0:
        return {"ok": False, "error": "old_str not found in file"}
    if count > 1:
        return {"ok": False, "error": f"old_str matches {count} times; make it more unique"}

    new_content = content.replace(old_str, args.get("new_str", ""), 1)
    try:
        with open(p, "w", encoding="utf-8") as f:
            f.write(new_content)
        return {"ok": True, "path": p}
    except Exception as e:
        return {"ok": False, "error": str(e)}

def tool_run_cmd(args: dict) -> dict:
    cmd = args.get("cmd")
    if not cmd:
        return {"ok": False, "error": "cmd is required"}
    try:
        timeout = max(1, int(args.get("timeout", 60)))
    except (TypeError, ValueError):
        timeout = 60
    try:
        proc = subprocess.run(
            cmd, text=True, cwd=args.get("cwd"), timeout=timeout,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            shell=not isinstance(cmd, list)
        )
        return {
            "ok": True,
            "returncode": proc.returncode,
            "stdout": proc.stdout,
            "stderr": proc.stderr,
        }
    except Exception as e:
        return {"ok": False, "error": str(e)}

def run_tool_call(call: dict) -> dict:
    name = call.get("function", {}).get("name")
    try:
        args = json.loads(call.get("function", {}).get("arguments", "{}"))
    except json.JSONDecodeError:
        args = {}
    tools = {
        "read_file": tool_read_file,
        "write_file": tool_write_file,
        "edit_file": tool_edit_file,
        "run_cmd": tool_run_cmd,
    }
    return tools.get(name, lambda _: {"ok": False, "error": f"unknown tool: {name}"})(args)

def chat_once(messages: list) -> dict:
    body = json.dumps({
        "model": "minimax/minimax-m2.5",
        "messages": messages,
        "tools": [
            {"type": "function", "function": {"name": "read_file", "description": "Read a file with line numbers. offset/limit control range; max_line_length truncates long lines.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "offset": {"type": "integer"}, "limit": {"type": "integer"}, "max_line_length": {"type": "integer"}}, "required": ["path"]}}},
            {"type": "function", "function": {"name": "write_file", "description": "Write full content to a file; creates parent dirs; overwrite=false fails if exists.", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}, "overwrite": {"type": "boolean"}}, "required": ["path", "content"]}}},
            {"type": "function", "function": {"name": "edit_file", "description": "Edit an existing file by replacing a unique old_str with new_str (\"\" to delete).", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "old_str": {"type": "string"}, "new_str": {"type": "string"}}, "required": ["path", "old_str"]}}},
            {"type": "function", "function": {"name": "run_cmd", "description": "Run a shell command (supports pipes/;). timeout is seconds.", "parameters": {"type": "object", "properties": {"cmd": {"type": ["string", "array"]}, "cwd": {"type": "string"}, "timeout": {"type": "number"}}, "required": ["cmd"]}}},
        ],
    }).encode("utf-8")

    req = urllib.request.Request("https://openrouter.ai/api/v1/chat/completions", data=body, method="POST")
    req.add_header("Authorization", f"Bearer {os.environ.get('OPENROUTER_API_KEY', '')}")
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json")

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    with urllib.request.urlopen(req, timeout=60, context=ctx) as resp:
        return json.loads(resp.read().decode("utf-8"))

if __name__ == "__main__":
    COLOR = {"reset": "\033[0m", "orange": "\033[38;5;214m", "purple": "\033[38;5;141m", "green": "\033[38;5;112m", "gray": "\033[38;5;240m"}
    if not os.environ.get("OPENROUTER_API_KEY"):
        print("API key not found, run: export OPENROUTER_API_KEY=your_key_here")
        sys.exit(1)

    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    while True:
        try:
            user = input("you> ")
        except EOFError:
            break
        if not user.strip():
            continue
        messages.append({"role": "user", "content": user})

        for _ in range(50): # max loops in one chat
            print(f"{COLOR['orange']}agent... {COLOR['reset']}", end="", flush=True)
            try:
                resp = chat_once(messages)
            except Exception as e:
                print(f"{COLOR['orange']}\rerror> {e}{COLOR['reset']}")
                break

            msg = resp.get("choices", [{}])[0].get("message", {})
            content = msg.get("content")
            reasoning = msg.get("reasoning")
            tool_calls = msg.get("tool_calls") or []

            if reasoning:
                print(f"{COLOR['gray']}\rreasoning> {reasoning.strip()}{COLOR['reset']}")
            if content:
                print(f"{COLOR['orange']}\ragent> {content}{COLOR['reset']}")
            else:
                print(f"{COLOR['orange']}\ragent> {COLOR['reset']}")

            if tool_calls:
                messages.append({ "role": "assistant", "content": content or "", "tool_calls": tool_calls})
                
                def clip(s: str, n: int = 1000) -> str:
                    return s if len(s) <= n else (s[:n] + "...")
                
                for tc in tool_calls:
                    name = tc.get("function", {}).get("name", "")
                    args_text = tc.get("function", {}).get("arguments", "")
                    print(f"{COLOR['purple']}tool> {name} {clip(args_text)}{COLOR['reset']}")

                    tool_result = run_tool_call(tc)
                    print(f"{COLOR['green']}tool_result> {clip(json.dumps(tool_result, ensure_ascii=False))}{COLOR['reset']}")

                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.get("id") or "",
                        "content": json.dumps(tool_result, ensure_ascii=False),
                    })
                continue

            messages.append({"role": "assistant", "content": content or ""})
            break