{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load env variables and create client\n",
    "from dotenv import load_dotenv\n",
    "from anthropic import Anthropic\n",
    "\n",
    "load_dotenv()\n",
    "\n",
    "client = Anthropic()\n",
    "model = \"claude-3-7-sonnet-latest\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Helper functions\n",
    "from anthropic.types import Message\n",
    "\n",
    "\n",
    "def add_user_message(messages, message):\n",
    "    user_message = {\n",
    "        \"role\": \"user\",\n",
    "        \"content\": message.content if isinstance(message, Message) else message,\n",
    "    }\n",
    "    messages.append(user_message)\n",
    "\n",
    "\n",
    "def add_assistant_message(messages, message):\n",
    "    assistant_message = {\n",
    "        \"role\": \"assistant\",\n",
    "        \"content\": message.content if isinstance(message, Message) else message,\n",
    "    }\n",
    "    messages.append(assistant_message)\n",
    "\n",
    "\n",
    "def chat(messages, system=None, temperature=1.0, stop_sequences=[], tools=None):\n",
    "    params = {\n",
    "        \"model\": model,\n",
    "        \"max_tokens\": 1000,\n",
    "        \"messages\": messages,\n",
    "        \"temperature\": temperature,\n",
    "        \"stop_sequences\": stop_sequences,\n",
    "    }\n",
    "\n",
    "    if tools:\n",
    "        params[\"tools\"] = tools\n",
    "\n",
    "    if system:\n",
    "        params[\"system\"] = system\n",
    "\n",
    "    message = client.messages.create(**params)\n",
    "    return message\n",
    "\n",
    "\n",
    "def text_from_message(message):\n",
    "    return \"\\n\".join(\n",
    "        [block.text for block in message.content if block.type == \"text\"]\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Implementation of the TextEditorTool\n",
    "import os\n",
    "import shutil\n",
    "from typing import Optional, List\n",
    "\n",
    "\n",
    "class TextEditorTool:\n",
    "    def __init__(self, base_dir: str = \"\", backup_dir: str = \"\"):\n",
    "        self.base_dir = base_dir or os.getcwd()\n",
    "        self.backup_dir = backup_dir or os.path.join(self.base_dir, \".backups\")\n",
    "        os.makedirs(self.backup_dir, exist_ok=True)\n",
    "\n",
    "    def _validate_path(self, file_path: str) -> str:\n",
    "        abs_path = os.path.normpath(os.path.join(self.base_dir, file_path))\n",
    "        if not abs_path.startswith(self.base_dir):\n",
    "            raise ValueError(\n",
    "                f\"Access denied: Path '{file_path}' is outside the allowed directory\"\n",
    "            )\n",
    "        return abs_path\n",
    "\n",
    "    def _backup_file(self, file_path: str) -> str:\n",
    "        if not os.path.exists(file_path):\n",
    "            return \"\"\n",
    "        file_name = os.path.basename(file_path)\n",
    "        backup_path = os.path.join(\n",
    "            self.backup_dir, f\"{file_name}.{os.path.getmtime(file_path):.0f}\"\n",
    "        )\n",
    "        shutil.copy2(file_path, backup_path)\n",
    "        return backup_path\n",
    "\n",
    "    def _restore_backup(self, file_path: str) -> str:\n",
    "        file_name = os.path.basename(file_path)\n",
    "        backups = [\n",
    "            f\n",
    "            for f in os.listdir(self.backup_dir)\n",
    "            if f.startswith(file_name + \".\")\n",
    "        ]\n",
    "        if not backups:\n",
    "            raise FileNotFoundError(f\"No backups found for {file_path}\")\n",
    "\n",
    "        latest_backup = sorted(backups, reverse=True)[0]\n",
    "        backup_path = os.path.join(self.backup_dir, latest_backup)\n",
    "\n",
    "        shutil.copy2(backup_path, file_path)\n",
    "        return f\"Successfully restored {file_path} from backup\"\n",
    "\n",
    "    def _count_matches(self, content: str, old_str: str) -> int:\n",
    "        return content.count(old_str)\n",
    "\n",
    "    def view(\n",
    "        self, file_path: str, view_range: Optional[List[int]] = None\n",
    "    ) -> str:\n",
    "        try:\n",
    "            abs_path = self._validate_path(file_path)\n",
    "\n",
    "            if os.path.isdir(abs_path):\n",
    "                try:\n",
    "                    return \"\\n\".join(os.listdir(abs_path))\n",
    "                except PermissionError:\n",
    "                    raise PermissionError(\n",
    "                        \"Permission denied. Cannot list directory contents.\"\n",
    "                    )\n",
    "\n",
    "            if not os.path.exists(abs_path):\n",
    "                raise FileNotFoundError(\"File not found\")\n",
    "\n",
    "            with open(abs_path, \"r\", encoding=\"utf-8\") as f:\n",
    "                content = f.read()\n",
    "\n",
    "            if view_range:\n",
    "                start, end = view_range\n",
    "                lines = content.split(\"\\n\")\n",
    "\n",
    "                if end == -1:\n",
    "                    end = len(lines)\n",
    "\n",
    "                selected_lines = lines[start - 1 : end]\n",
    "\n",
    "                result = []\n",
    "                for i, line in enumerate(selected_lines, start):\n",
    "                    result.append(f\"{i}: {line}\")\n",
    "\n",
    "                return \"\\n\".join(result)\n",
    "            else:\n",
    "                lines = content.split(\"\\n\")\n",
    "                result = []\n",
    "                for i, line in enumerate(lines, 1):\n",
    "                    result.append(f\"{i}: {line}\")\n",
    "\n",
    "                return \"\\n\".join(result)\n",
    "\n",
    "        except UnicodeDecodeError:\n",
    "            raise UnicodeDecodeError(\n",
    "                \"utf-8\",\n",
    "                b\"\",\n",
    "                0,\n",
    "                1,\n",
    "                \"File contains non-text content and cannot be displayed.\",\n",
    "            )\n",
    "        except ValueError as e:\n",
    "            raise ValueError(str(e))\n",
    "        except PermissionError:\n",
    "            raise PermissionError(\"Permission denied. Cannot access file.\")\n",
    "        except Exception as e:\n",
    "            raise type(e)(str(e))\n",
    "\n",
    "    def str_replace(self, file_path: str, old_str: str, new_str: str) -> str:\n",
    "        try:\n",
    "            abs_path = self._validate_path(file_path)\n",
    "\n",
    "            if not os.path.exists(abs_path):\n",
    "                raise FileNotFoundError(\"File not found\")\n",
    "\n",
    "            with open(abs_path, \"r\", encoding=\"utf-8\") as f:\n",
    "                content = f.read()\n",
    "\n",
    "            match_count = self._count_matches(content, old_str)\n",
    "\n",
    "            if match_count == 0:\n",
    "                raise ValueError(\n",
    "                    \"No match found for replacement. Please check your text and try again.\"\n",
    "                )\n",
    "            elif match_count > 1:\n",
    "                raise ValueError(\n",
    "                    f\"Found {match_count} matches for replacement text. Please provide more context to make a unique match.\"\n",
    "                )\n",
    "\n",
    "            # Create backup before modifying\n",
    "            self._backup_file(abs_path)\n",
    "\n",
    "            # Perform the replacement\n",
    "            new_content = content.replace(old_str, new_str)\n",
    "\n",
    "            with open(abs_path, \"w\", encoding=\"utf-8\") as f:\n",
    "                f.write(new_content)\n",
    "\n",
    "            return \"Successfully replaced text at exactly one location.\"\n",
    "\n",
    "        except ValueError as e:\n",
    "            raise ValueError(str(e))\n",
    "        except PermissionError:\n",
    "            raise PermissionError(\"Permission denied. Cannot modify file.\")\n",
    "        except Exception as e:\n",
    "            raise type(e)(str(e))\n",
    "\n",
    "    def create(self, file_path: str, file_text: str) -> str:\n",
    "        try:\n",
    "            abs_path = self._validate_path(file_path)\n",
    "\n",
    "            # Check if file already exists\n",
    "            if os.path.exists(abs_path):\n",
    "                raise FileExistsError(\n",
    "                    \"File already exists. Use str_replace to modify it.\"\n",
    "                )\n",
    "\n",
    "            # Create parent directories if they don't exist\n",
    "            os.makedirs(os.path.dirname(abs_path), exist_ok=True)\n",
    "\n",
    "            # Create the file\n",
    "            with open(abs_path, \"w\", encoding=\"utf-8\") as f:\n",
    "                f.write(file_text)\n",
    "\n",
    "            return f\"Successfully created {file_path}\"\n",
    "\n",
    "        except ValueError as e:\n",
    "            raise ValueError(str(e))\n",
    "        except PermissionError:\n",
    "            raise PermissionError(\"Permission denied. Cannot create file.\")\n",
    "        except Exception as e:\n",
    "            raise type(e)(str(e))\n",
    "\n",
    "    def insert(self, file_path: str, insert_line: int, new_str: str) -> str:\n",
    "        try:\n",
    "            abs_path = self._validate_path(file_path)\n",
    "\n",
    "            if not os.path.exists(abs_path):\n",
    "                raise FileNotFoundError(\"File not found\")\n",
    "\n",
    "            # Create backup before modifying\n",
    "            self._backup_file(abs_path)\n",
    "\n",
    "            with open(abs_path, \"r\", encoding=\"utf-8\") as f:\n",
    "                lines = f.readlines()\n",
    "\n",
    "            # Handle line endings\n",
    "            if lines and not lines[-1].endswith(\"\\n\"):\n",
    "                new_str = \"\\n\" + new_str\n",
    "\n",
    "            # Insert at the beginning if insert_line is 0\n",
    "            if insert_line == 0:\n",
    "                lines.insert(0, new_str + \"\\n\")\n",
    "            # Insert after the specified line\n",
    "            elif insert_line > 0 and insert_line <= len(lines):\n",
    "                lines.insert(insert_line, new_str + \"\\n\")\n",
    "            else:\n",
    "                raise IndexError(\n",
    "                    f\"Line number {insert_line} is out of range. File has {len(lines)} lines.\"\n",
    "                )\n",
    "\n",
    "            with open(abs_path, \"w\", encoding=\"utf-8\") as f:\n",
    "                f.writelines(lines)\n",
    "\n",
    "            return f\"Successfully inserted text after line {insert_line}\"\n",
    "\n",
    "        except ValueError as e:\n",
    "            raise ValueError(str(e))\n",
    "        except PermissionError:\n",
    "            raise PermissionError(\"Permission denied. Cannot modify file.\")\n",
    "        except Exception as e:\n",
    "            raise type(e)(str(e))\n",
    "\n",
    "    def undo_edit(self, file_path: str) -> str:\n",
    "        try:\n",
    "            abs_path = self._validate_path(file_path)\n",
    "\n",
    "            if not os.path.exists(abs_path):\n",
    "                raise FileNotFoundError(\"File not found\")\n",
    "\n",
    "            return self._restore_backup(abs_path)\n",
    "\n",
    "        except ValueError as e:\n",
    "            raise ValueError(str(e))\n",
    "        except FileNotFoundError:\n",
    "            raise FileNotFoundError(\"No previous edits to undo\")\n",
    "        except PermissionError:\n",
    "            raise PermissionError(\"Permission denied. Cannot restore file.\")\n",
    "        except Exception as e:\n",
    "            raise type(e)(str(e))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Process Tool Call Requests\n",
    "import json\n",
    "\n",
    "text_editor_tool = TextEditorTool()\n",
    "\n",
    "\n",
    "def run_tool(tool_name, tool_input):\n",
    "    if tool_name == \"str_replace_editor\":\n",
    "        command = tool_input[\"command\"]\n",
    "        if command == \"view\":\n",
    "            return text_editor_tool.view(\n",
    "                tool_input[\"path\"], tool_input.get(\"view_range\")\n",
    "            )\n",
    "        elif command == \"str_replace\":\n",
    "            return text_editor_tool.str_replace(\n",
    "                tool_input[\"path\"], tool_input[\"old_str\"], tool_input[\"new_str\"]\n",
    "            )\n",
    "        elif command == \"create\":\n",
    "            return text_editor_tool.create(\n",
    "                tool_input[\"path\"], tool_input[\"file_text\"]\n",
    "            )\n",
    "        elif command == \"insert\":\n",
    "            return text_editor_tool.insert(\n",
    "                tool_input[\"path\"],\n",
    "                tool_input[\"insert_line\"],\n",
    "                tool_input[\"new_str\"],\n",
    "            )\n",
    "        elif command == \"undo_edit\":\n",
    "            return text_editor_tool.undo_edit(tool_input[\"path\"])\n",
    "        else:\n",
    "            raise Exception(f\"Unknown text editor command: {command}\")\n",
    "    else:\n",
    "        raise Exception(f\"Unknown tool name: {tool_name}\")\n",
    "\n",
    "\n",
    "def run_tools(message):\n",
    "    tool_requests = [\n",
    "        block for block in message.content if block.type == \"tool_use\"\n",
    "    ]\n",
    "    tool_result_blocks = []\n",
    "\n",
    "    for tool_request in tool_requests:\n",
    "        try:\n",
    "            tool_output = run_tool(tool_request.name, tool_request.input)\n",
    "            tool_result_block = {\n",
    "                \"type\": \"tool_result\",\n",
    "                \"tool_use_id\": tool_request.id,\n",
    "                \"content\": json.dumps(tool_output),\n",
    "                \"is_error\": False,\n",
    "            }\n",
    "        except Exception as e:\n",
    "            tool_result_block = {\n",
    "                \"type\": \"tool_result\",\n",
    "                \"tool_use_id\": tool_request.id,\n",
    "                \"content\": f\"Error: {e}\",\n",
    "                \"is_error\": True,\n",
    "            }\n",
    "\n",
    "        tool_result_blocks.append(tool_result_block)\n",
    "\n",
    "    return tool_result_blocks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make the text edit schema based on the model version being used\n",
    "def get_text_edit_schema(model):\n",
    "    if model.startswith(\"claude-3-7-sonnet\"):\n",
    "        return {\n",
    "            \"type\": \"text_editor_20250124\",\n",
    "            \"name\": \"str_replace_editor\",\n",
    "        }\n",
    "    elif model.startswith(\"claude-3-5-sonnet\"):\n",
    "        return {\n",
    "            \"type\": \"text_editor_20241022\",\n",
    "            \"name\": \"str_replace_editor\",\n",
    "        }\n",
    "    else:\n",
    "        raise ValueError(\n",
    "            f\"Editor schema version not known for model: {model}. Reference Anthropic docs for the correct schema version.\"\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run the conversation in a loop until the model doesn't ask for a tool use\n",
    "def run_conversation(messages):\n",
    "    while True:\n",
    "        response = chat(\n",
    "            messages,\n",
    "            tools=[get_text_edit_schema(model)],\n",
    "        )\n",
    "\n",
    "        add_assistant_message(messages, response)\n",
    "        print(text_from_message(response))\n",
    "\n",
    "        if response.stop_reason != \"tool_use\":\n",
    "            break\n",
    "\n",
    "        tool_results = run_tools(response)\n",
    "        add_user_message(messages, tool_results)\n",
    "\n",
    "    return messages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "messages = []\n",
    "\n",
    "add_user_message(\n",
    "    messages,\n",
    "    \"\"\"\n",
    "\n",
    "    \"\"\",\n",
    ")\n",
    "\n",
    "run_conversation(messages)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}