Shell с безопасностью

Слова, которые встретятся

• Фабрика (factory) — функция, которая собирает инструмент и возвращает готовую функцию/объект. Вы не копируете bash руками каждый раз — вызываете фабрику с нужными настройками.
• Шов / seam — место, где можно подменить реализацию, не ломая остальное. Здесь шов — «кто выполняет exec».
• Protocol (в typing) — «контракт класса» без наследования: «у тебя должен быть метод exec». Утка крякает — значит утка.
• Sandbox — изолированная среда (позже). Пока достаточно знать: это «другой бэкенд для тех же команд».

Быстрый путь

1. Создайте tools_bash.py — Protocol и фабрика create_bash_tool.
2. Создайте local_bash.py — класс LocalBashOps на subprocess.
3. В main.py уберите старый inline-bash и подключите фабрику.

Какие файлы трогаем

Если всё ещё лежит одним файлом — нормально: можете временно держать классы в main.py, но имена ролей те же. Лучше сразу разнести: потом модуль 4 подменит только local_bash.py.

Упражнение 2.2

• Проверка белого списка остаётся в фабрике (tools_bash.py), не в LocalBashOps.
• LocalBashOps.exec всегда возвращает пару stdout + код выхода (и при ошибке тоже).
• read пока не трогаем — фабрика нужна там, где бэкенд реально разъедется.

Где режем

В старом main.py было примерно так (всё в одном месте):

if not is_safe(command):
    return "Заблокировано..."
proc = subprocess.run(command, shell=True, cwd=ctx.deps, ...)

Когда появится песочница, вместо subprocess будет sandbox.exec. Фабрика вводит шов — его объявляем в tools_bash.py:

from typing import Protocol, TypedDict

class ExecResult(TypedDict):
    stdout: str
    exit_code: int

class BashOperations(Protocol):
    async def exec(self, command: str) -> ExecResult: ...

Над швом — то, что видит модель (описание, аргументы, белый список). Под швом — железо или эмулятор.

1. Файл tools_bash.py (создать)

tools_bash.py

from collections.abc import Callable
from typing import Protocol, TypedDict

class ExecResult(TypedDict):
    stdout: str
    exit_code: int

class BashOperations(Protocol):
    async def exec(self, command: str) -> ExecResult: ...


def create_bash_tool(
    operations: BashOperations,
    safe_prefixes: list[str],
) -> Callable[..., str]:
    def is_safe(command: str) -> bool:
        cmd = command.strip()
        return any(cmd.startswith(p) for p in safe_prefixes)

    async def bash(command: str) -> str:
        """Выполнить shell-команду в рабочей директории.

        WHEN TO USE: сборка, тесты, git, список файлов.
        WHEN NOT TO USE: чтение файла (read); поиск (grep).
        DO NOT USE FOR: чтение файлов (read), поиск по коду (grep).
        USAGE: одна строка shell; вне белого списка — понятная блокировка.
        EXAMPLES:
          - ls -la
          - git status
        """
        if not is_safe(command):
            return f'Заблокировано: "{command}" не из белого списка.'
        result = await operations.exec(command)
        return result["stdout"] or "(нет вывода)"

    return bash

Заметьте: внутри фабрики нет subprocess и нет знания про cwd. Есть только «что-то с методом exec».

2. Файл local_bash.py (создать)

Здесь живёт только запуск на вашей машине. Белый список сюда не тащите — иначе шов размажется.

Важно: тип ExecResult объявлен в tools_bash.py. В local_bash.py его нужно импортировать. Если вы временно кладёте LocalBashOps прямо в main.py — либо импортируйте from tools_bash import ExecResult, либо сначала объявите ExecResult выше класса. Иначе будет NameError: name 'ExecResult' is not defined.

local_bash.py

import subprocess
from pathlib import Path

from tools_bash import ExecResult  # без этой строки — NameError


class LocalBashOps:
    def __init__(self, cwd: Path) -> None:
        self.cwd = cwd

    async def exec(self, command: str) -> ExecResult:
        try:
            proc = subprocess.run(
                command,
                shell=True,
                cwd=self.cwd,
                capture_output=True,
                text=True,
                timeout=30,
            )
            out = (proc.stdout or "") + (proc.stderr or "")
            return {"stdout": out.strip(), "exit_code": proc.returncode}
        except subprocess.TimeoutExpired:
            return {"stdout": "Таймаут: больше 30 секунд.", "exit_code": 124}

3. Файл main.py (править)

Удалите целиком старый @agent.tool / функцию с именем bash (ту, где ещё был subprocess). Если оставить её и добавить tool из фабрики — получите:

Дальше соберите tool из фабрики и повесьте на агента один раз:

main.py — фрагмент сборки

import asyncio
import sys
from pathlib import Path

from pydantic_ai import Agent, RunContext, Tool

from local_bash import LocalBashOps
from tools_bash import create_bash_tool


MAX_LINES = 500

# Белый список: команда должна НАЧИНАТЬСЯ с одной из этих строк
SAFE_PREFIXES = [
    "ls", "cat", "echo", "pwd", "which", "find",
    "head", "tail", "wc",
    "git log", "git status", "git diff",
]

cwd = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else Path.cwd()
prompt = " ".join(sys.argv[2:]) if len(sys.argv) > 2 else "Привет!"

local_ops = LocalBashOps(cwd)
bash_fn = create_bash_tool(local_ops, SAFE_PREFIXES)

agent = Agent(
    "deepseek:deepseek-chat",
    deps_type=Path,
    tools=[Tool(bash_fn, takes_ctx=False)],
    instructions=(
        f"Ты coding-агент.\nРабочая директория: {cwd}\n"
        "Используй инструмент read, чтобы открывать файлы. "
        "Не выдумывай содержимое и не имитируй tool-calls текстом."
    ),
)

Проверьте

uv run python main.py . "Перечисли файлы в этой директории"
uv run python main.py . "Выполни: rm -rf .venv"

Поведение как в уроке 1.3: безопасное — работает, опасное — текст блокировки. Модель не обязана заметить рефакторинг. Это и есть комплимент архитектуре.

Готово, если:

• Есть файл tools_bash.py с Protocol и create_bash_tool
• Есть файл local_bash.py с LocalBashOps (subprocess только там)
• В main.py старый inline-bash убран, tool собран через фабрику
• Безопасные команды проходят, опасные по-прежнему блокируются текстом
• (бонус) Набросали MockOps, который всегда отвечает «(pretend output)»