Mini Vibe : un agent de codage dans votre terminal et votre navigateur

Mini Vibe est un agent de codage en environ 350 lignes de Python, inspiré de Mistral Vibe dans le terminal et de Vibe Code dans le navigateur. Lancez-le dans votre dépôt et saisissez une tâche : l'agent lit le code, modifie les fichiers et exécute des commandes. Les modifications s'appliquent immédiatement, sauf celles des paramètres de git, et git diff les affiche ; toute commande qui ne figure pas dans une courte liste de commandes en lecture seule demande d'abord votre autorisation. La même session est accessible dans un onglet du navigateur, qui affiche le travail au fil de l'exécution et permet aussi de répondre aux demandes d'approbation. Pendant que l'agent travaille, saisissez une ligne pour l'orienter ou appuyez sur Ctrl-C pour l'arrêter.

L'enregistrement : Mini Vibe corrige l'issue #1336 de more-itertools, où replace() renvoie None pour un itérable vide lorsque window_size > 1. L'onglet du navigateur répond à la deuxième commande, une ligne saisie pendant l'exécution oriente l'agent vers le correctif déjà utilisé par locate(), et a autorise toutes les commandes suivantes.

Cet exemple montre : une politique d'autorisation qui exécute les commandes en lecture seule et demande une autorisation pour toutes les autres, une sandbox sur la machine hôte dans votre copie de travail, deux vues suivant les événements d'une même session, la réponse à une demande d'approbation depuis l'une ou l'autre vue, et l'orientation du tour en cours.

Exécutez cette commande depuis le dépôt sur lequel vous voulez travailler, avec MISTRAL_API_KEY défini et MINI_VIBE désignant le dossier de l'exemple : ALLOW_DANGEROUS_LOCAL_HOST=1 uv run --project "$MINI_VIBE" python "$MINI_VIBE/mini_vibe.py"

L'agent travaille dans votre copie de travail via DangerousLocalSandbox(workspace=os.getcwd()) : ses outils de shell et de gestion des fichiers s'exécutent sur votre machine, sans conteneur. La politique détermine quels appels attendent votre réponse :

# The patterns read the call's arguments as JSON text, where a newline is the two characters \n.
# A command that could run a second command or write a file asks: a separator, a newline, a
# redirection, an expansion ($), a backtick, or git's --output option.
ASK = ("*;*", "*&*", "*|*", "*\\n*", "*>*", "*<(*", "*$*", "*`*", "*--output*")
# Git's settings can name a program that `git diff` or `git log` runs, so an edit to them asks: a
# path with .git in it (.git/, .gitattributes, .gitconfig), in any case, since macOS disks ignore
# case. `path` matches both write_file's `path` and edit's `file_path`.
GIT_SETTINGS = ('*path": "*[.][gG][iI][tT]*',)
READ_ONLY = (
    "ls",
    "ls *",
    "cat *",
    "head *",
    "tail *",
    "wc *",
    "grep *",
    "git status",
    "git status *",
    "git diff",
    "git diff *",  # with the space: `git difftool` runs a program of its choosing
    "git log",
    "git log *",
    "git show",
    "git show *",
)

coder = agents.Agent(
    model="mistral-large-4",
    name="mini-vibe",
    instructions=(
        "You are a coding agent working in the user's repository, your current directory. Read the "
        "code before you change it, make the smallest change that does the task, then run the "
        "project's tests. End with two or three lines: what you changed and what the tests said."
    ),
    environment=agents.DangerousLocalSandbox(workspace=os.getcwd()),  # your checkout, no container
    permission_policy=agents.PermissionPolicy(
        default="ask",
        rules=[
            agents.PermissionRule(tools=["file_system.read_file"], decision="allow"),
            agents.PermissionRule(
                tools=["file_system.write_file", "file_system.edit"],
                decision="allow",  # like Vibe: edits apply, `git diff` shows them
                sensitive=GIT_SETTINGS,
            ),
            agents.PermissionRule(
                tools=["file_system.bash"],
                decision="ask",
                sensitive=ASK,
                allowlist=[f'*"command": "{command}"*' for command in READ_ONLY],
            ),
        ],
    ),
)

# The patterns read the call's arguments as JSON text, where a newline is the two characters \n.
# A command that could run a second command or write a file asks: a separator, a newline, a
# redirection, an expansion ($), a backtick, or git's --output option.
ASK = ("*;*", "*&*", "*|*", "*\\n*", "*>*", "*<(*", "*$*", "*`*", "*--output*")
# Git's settings can name a program that `git diff` or `git log` runs, so an edit to them asks: a
# path with .git in it (.git/, .gitattributes, .gitconfig), in any case, since macOS disks ignore
# case. `path` matches both write_file's `path` and edit's `file_path`.
GIT_SETTINGS = ('*path": "*[.][gG][iI][tT]*',)
READ_ONLY = (
    "ls",
    "ls *",
    "cat *",
    "head *",
    "tail *",
    "wc *",
    "grep *",
    "git status",
    "git status *",
    "git diff",
    "git diff *",  # with the space: `git difftool` runs a program of its choosing
    "git log",
    "git log *",
    "git show",
    "git show *",
)

coder = agents.Agent(
    model="mistral-large-4",
    name="mini-vibe",
    instructions=(
        "You are a coding agent working in the user's repository, your current directory. Read the "
        "code before you change it, make the smallest change that does the task, then run the "
        "project's tests. End with two or three lines: what you changed and what the tests said."
    ),
    environment=agents.DangerousLocalSandbox(workspace=os.getcwd()),  # your checkout, no container
    permission_policy=agents.PermissionPolicy(
        default="ask",
        rules=[
            agents.PermissionRule(tools=["file_system.read_file"], decision="allow"),
            agents.PermissionRule(
                tools=["file_system.write_file", "file_system.edit"],
                decision="allow",  # like Vibe: edits apply, `git diff` shows them
                sensitive=GIT_SETTINGS,
            ),
            agents.PermissionRule(
                tools=["file_system.bash"],
                decision="ask",
                sensitive=ASK,
                allowlist=[f'*"command": "{command}"*' for command in READ_ONLY],
            ),
        ],
    ),
)

• La lecture et la modification des fichiers sont autorisées. Vous vérifiez le résultat avec git diff, comme dans Vibe.
• Une modification des paramètres de git demande une autorisation. .git/config peut désigner un programme qu'un simple git diff exécute, et .gitattributes sélectionne les fichiers sur lesquels il s'exécute. Le motif reconnaît .git quelle que soit la casse : les disques macOS ignorent la casse, donc .GIT/config désigne le même fichier. Le motif reconnaît aussi .github/, .gitignore et une modification dont le texte contient .git : ces opérations demandent également une autorisation.
• Une commande en lecture seule est exécutée. grep -n window more.py correspond à la liste d'autorisation.
• Toute autre commande demande une autorisation. python -m unittest, find . -delete et git difftool ne correspondent à aucun motif de la liste d'autorisation : git diff * comporte un espace après diff.
• Une commande susceptible d'en exécuter une seconde ou d'écrire dans un fichier demande une autorisation, même si elle commence comme une commande en lecture seule. Les motifs reconnaissent les arguments de l'appel sous forme de texte JSON, où un saut de ligne est représenté par les deux caractères \n : cat a.py; rm -rf . et cat a.py suivi d'un saut de ligne puis de rm -rf . correspondent tous deux à un motif sensitive, et sensitive est vérifié avant la liste d'autorisation.

Les motifs analysent le texte de la commande : ils détectent une commande imprudente, pas une commande malveillante. Une modification des paramètres utilisateur de git hors du dépôt, dans ~/.config/git/config, s'applique sans demander d'autorisation. Toute autre modification s'applique également sans autorisation. Avant d'autoriser une commande, examinez donc les modifications du tour : un fichier de test modifié change ce qu'exécute python -m unittest. test_mini_vibe.py teste des commandes et des modifications de ce type, une par cas, en les soumettant à la politique avec un modèle piloté par un script ; depuis le dossier de l'exemple, lancez-le avec ALLOW_DANGEROUS_LOCAL_HOST=1 uv run pytest -o asyncio_mode=auto test_mini_vibe.py.

Le programme nécessite ALLOW_DANGEROUS_LOCAL_HOST=1, et la création de la sandbox affiche un UserWarning : les commandes de l'agent s'exécutent avec vos droits, vos fichiers et votre accès réseau. Lancez-le dans une copie de travail que vous pouvez réinitialiser.

Le terminal et l'onglet du navigateur lisent chacun session.iter_events() de leur côté. Aucune des deux vues ne pilote le tour : l'envoi d'un message depuis l'une ou l'autre le démarre, et les deux vues reçoivent tous les événements. Le terminal affiche chaque événement dès son arrivée :

async def show(self) -> None:
    async for event in self.session.iter_events():
        match event:
            case agents.TurnStarted():
                self.turn_id = event.turn_id
            case agents.TextDelta(entry_id=entry_id, text=text):
                self.streamed.add(entry_id)
                self.out(text, end="")
            case agents.EntryCompleted(entry=agents.MessageEntry(role="assistant") as message):
                self.out("" if message.id in self.streamed else message.text)
            case agents.EntryCompleted(entry=agents.ToolCallEntry(name=name, arguments=args)):
                self.out(f"{ORANGE}⏺{RESET} {BOLD}{name.split('.')[-1]}{RESET} {web.summary(args)}")
            case agents.EntryCompleted(entry=agents.ToolResultEntry() as result):
                lines = web.result_text(result).strip().splitlines() or [""]
                more = f" (+{len(lines) - 1} lines)" if len(lines) > 1 else ""
                self.out(f"  {DIM}⎿ {lines[0][:100]}{more}{RESET}")
            case agents.CallbackOpened(callback=agents.OpenApproval() as approval):
                await self.ask(approval)
            case agents.CallbackResolved(callback_id=callback_id):
                for approval in [a for a in self.waiting if a.id == callback_id]:
                    self.waiting.remove(approval)
                    self.out(f"  {DIM}answered in the web view: {web.summary(approval.request.input)}{RESET}")
            case agents.TurnEnded(state=state):
                self.turn_id = None
                if not isinstance(state, agents.Completed):
                    self.out(f"{DIM}[{type(state).__name__}]{RESET}")
                self.out(f"{BOLD}>{RESET} ", end="")
        if self.waiting and not self.prompting and not isinstance(event, agents.TextDelta):
            self.prompt()  # ask again, below the new output

async def show(self) -> None:
    async for event in self.session.iter_events():
        match event:
            case agents.TurnStarted():
                self.turn_id = event.turn_id
            case agents.TextDelta(entry_id=entry_id, text=text):
                self.streamed.add(entry_id)
                self.out(text, end="")
            case agents.EntryCompleted(entry=agents.MessageEntry(role="assistant") as message):
                self.out("" if message.id in self.streamed else message.text)
            case agents.EntryCompleted(entry=agents.ToolCallEntry(name=name, arguments=args)):
                self.out(f"{ORANGE}⏺{RESET} {BOLD}{name.split('.')[-1]}{RESET} {web.summary(args)}")
            case agents.EntryCompleted(entry=agents.ToolResultEntry() as result):
                lines = web.result_text(result).strip().splitlines() or [""]
                more = f" (+{len(lines) - 1} lines)" if len(lines) > 1 else ""
                self.out(f"  {DIM}⎿ {lines[0][:100]}{more}{RESET}")
            case agents.CallbackOpened(callback=agents.OpenApproval() as approval):
                await self.ask(approval)
            case agents.CallbackResolved(callback_id=callback_id):
                for approval in [a for a in self.waiting if a.id == callback_id]:
                    self.waiting.remove(approval)
                    self.out(f"  {DIM}answered in the web view: {web.summary(approval.request.input)}{RESET}")
            case agents.TurnEnded(state=state):
                self.turn_id = None
                if not isinstance(state, agents.Completed):
                    self.out(f"{DIM}[{type(state).__name__}]{RESET}")
                self.out(f"{BOLD}>{RESET} ", end="")
        if self.waiting and not self.prompting and not isinstance(event, agents.TextDelta):
            self.prompt()  # ask again, below the new output

La vue web envoie les mêmes événements à la page sous forme de server-sent events, chacun avec son id. Lorsqu'un onglet se recharge, il renvoie le dernier id reçu, et le flux reprend à partir de là, sans afficher de doublon :

async def events(request: Request) -> StreamingResponse:
    from_the_page(request)
    # A browser that reconnects sends the id of the last event it got: nothing shows twice.
    after = request.headers.get("last-event-id")

    async def stream():
        async for event in session.iter_events(after=after):
            if (data := as_json(event)) is not None:
                yield f"id: {event.id}\ndata: {json.dumps(data)}\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

async def events(request: Request) -> StreamingResponse:
    from_the_page(request)
    # A browser that reconnects sends the id of the last event it got: nothing shows twice.
    after = request.headers.get("last-event-id")

    async def stream():
        async for event in session.iter_events(after=after):
            if (data := as_json(event)) is not None:
                yield f"id: {event.id}\ndata: {json.dumps(data)}\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

Une demande d'approbation reste en attente jusqu'à ce qu'une vue y réponde, avec son identifiant et celui de son tour. Si les deux répondent, la première réponse est retenue. La seconde reçoit AlreadySettled, ou Answered avec la réponse de la première vue, qui diffère de la sienne si les deux vues ont donné des réponses différentes. L'autre vue reçoit alors CallbackResolved et retire sa question :

async def answer(self, approval: agents.OpenApproval, decision: agents.AllowApproval | agents.DenyApproval) -> None:
    match await self.session.turn(approval.turn_id).answer(approval.id, decision):
        case agents.Answered(answer=applied) if applied != decision:
            self.out(f"  {DIM}the web view answered first{RESET}")
        case agents.AlreadySettled():
            self.out(f"  {DIM}the web view answered first{RESET}")

async def answer(self, approval: agents.OpenApproval, decision: agents.AllowApproval | agents.DenyApproval) -> None:
    match await self.session.turn(approval.turn_id).answer(approval.id, decision):
        case agents.Answered(answer=applied) if applied != decision:
            self.out(f"  {DIM}the web view answered first{RESET}")
        case agents.AlreadySettled():
            self.out(f"  {DIM}the web view answered first{RESET}")

async def answer(request: Request) -> JSONResponse:
    from_the_page(request)
    body = await request.json()
    decision = agents.AllowApproval() if body["allow"] else agents.DenyApproval()
    match await session.turn(body["turn"]).answer(body["id"], decision):
        case agents.Answered(answer=applied) if applied != decision:
            return JSONResponse({"result": "the terminal answered first"})
        case agents.AlreadySettled():
            return JSONResponse({"result": "the terminal answered first"})
    return JSONResponse({"result": "done"})

async def answer(request: Request) -> JSONResponse:
    from_the_page(request)
    body = await request.json()
    decision = agents.AllowApproval() if body["allow"] else agents.DenyApproval()
    match await session.turn(body["turn"]).answer(body["id"], decision):
        case agents.Answered(answer=applied) if applied != decision:
            return JSONResponse({"result": "the terminal answered first"})
        case agents.AlreadySettled():
            return JSONResponse({"result": "the terminal answered first"})
    return JSONResponse({"result": "done"})

Le modèle peut appeler plusieurs outils à la fois, donc plusieurs demandes d'approbation peuvent être en attente simultanément. Le terminal les présente une par une. a (toujours) autorise l'outil pour le reste de la session, ici file_system.bash : toutes les commandes suivantes s'exécutent sans autorisation, y compris les commandes enchaînées, tout comme les autres appels en attente de cet outil. a sur une modification autorise cet outil de la même façon : les modifications ultérieures des paramètres de git s'appliquent elles aussi sans autorisation.

Chaque route, sauf celle de la page elle-même, vérifie d'abord deux éléments. Sans ces vérifications, une page d'un autre site web ouverte dans le même navigateur pourrait lire votre code depuis 127.0.0.1:8765 ou envoyer une tâche qui modifie vos fichiers :

• Le token de l'exécution. Chaque exécution en crée un nouveau et affiche dans le terminal le lien qui le contient. La page le renvoie avec chaque requête.
• L'Origin de la requête. Une requête envoyée par une page d'un autre site web porte l'origine de ce site.

def from_the_page(request: Request) -> None:
    """Refuse a request without this run's token, or sent by another website's page: it could read
    your code or start a turn that edits your files."""
    token = request.headers.get("x-token") or request.query_params.get("token", "")
    if not secrets.compare_digest(token, TOKEN) or request.headers.get("origin", URL) != URL:
        raise HTTPException(403, "only the web view's own page can send this")

def from_the_page(request: Request) -> None:
    """Refuse a request without this run's token, or sent by another website's page: it could read
    your code or start a turn that edits your files."""
    token = request.headers.get("x-token") or request.query_params.get("token", "")
    if not secrets.compare_digest(token, TOKEN) or request.headers.get("origin", URL) != URL:
        raise HTTPException(403, "only the web view's own page can send this")

Une ligne saisie pendant que l'agent travaille n'attend pas la fin du tour. send_message() démarre un tour si aucun n'est en cours ; sinon, il ajoute la ligne au tour en cours, qui la lit à l'étape suivante. Ctrl-C interrompt le tour ; la session reste ouverte pour la tâche suivante :

elif line.strip():
    self.prompting = False
    if self.turn_id is not None:
        self.out(f"  {DIM}↳ steering the running turn{RESET}")
    await self.session.send_message(line)  # starts a turn, or steers the running one
    self.prompt()  # an open approval still waits for y/n/a

elif line.strip():
    self.prompting = False
    if self.turn_id is not None:
        self.out(f"  {DIM}↳ steering the running turn{RESET}")
    await self.session.send_message(line)  # starts a turn, or steers the running one
    self.prompt()  # an open approval still waits for y/n/a

async def stop(self) -> None:
    if self.turn_id is not None:
        await self.session.turn(self.turn_id).interrupt()

async def stop(self) -> None:
    if self.turn_id is not None:
        await self.session.turn(self.turn_id).interrupt()

Le formulaire et le bouton Arrêter de la vue web appellent les deux mêmes méthodes.

Une session démarrée dans votre processus n'est accessible que depuis la boucle d'événements qui l'a démarrée. Le serveur web s'exécute donc comme une tâche dans cette boucle, aux côtés du terminal, plutôt que dans un thread :

async def main() -> None:
    session = await coder.start_session()
    terminal = Terminal(session)
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.ensure_future(terminal.stop()))  # Ctrl-C stops the turn
    print(
        f"{BOLD}mini-vibe{RESET} {DIM}· {os.path.basename(os.getcwd())} · {coder.model} · web view: {web.LINK}{RESET}"
    )
    print(f"{BOLD}>{RESET} ", end="", flush=True)
    browser = web.server(session)
    tasks = [asyncio.create_task(terminal.show()), asyncio.create_task(browser.serve())]
    try:
        await read_lines(terminal)
    finally:
        browser.should_exit = browser.force_exit = True  # an open browser tab does not hold the exit
        tasks[0].cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        await session.cancel()

async def main() -> None:
    session = await coder.start_session()
    terminal = Terminal(session)
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.ensure_future(terminal.stop()))  # Ctrl-C stops the turn
    print(
        f"{BOLD}mini-vibe{RESET} {DIM}· {os.path.basename(os.getcwd())} · {coder.model} · web view: {web.LINK}{RESET}"
    )
    print(f"{BOLD}>{RESET} ", end="", flush=True)
    browser = web.server(session)
    tasks = [asyncio.create_task(terminal.show()), asyncio.create_task(browser.serve())]
    try:
        await read_lines(terminal)
    finally:
        browser.should_exit = browser.force_exit = True  # an open browser tab does not hold the exit
        tasks[0].cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        await session.cancel()

web.server renvoie un uvicorn.Server dont capture_signals ne fait rien. Ctrl-C atteint donc le gestionnaire du terminal et arrête le tour, pas la vue web.

L'exemple est un projet avec son propre pyproject.toml :

Installez l'exemple une seule fois, depuis son dossier :

uv sync

uv sync

Puis, depuis le dépôt sur lequel vous voulez travailler :

export MISTRAL_API_KEY=...
export MINI_VIBE=~/path/to/mini_vibe  # the example's folder
ALLOW_DANGEROUS_LOCAL_HOST=1 uv run --project "$MINI_VIBE" python "$MINI_VIBE/mini_vibe.py"

export MISTRAL_API_KEY=...
export MINI_VIBE=~/path/to/mini_vibe  # the example's folder
ALLOW_DANGEROUS_LOCAL_HOST=1 uv run --project "$MINI_VIBE" python "$MINI_VIBE/mini_vibe.py"

Ouvrez le lien de la vue web affiché par le terminal, http://127.0.0.1:8765/?token=…, pour suivre la session dans un navigateur. Saisissez y, n ou a pour répondre à une question dans le terminal, ou cliquez sur Autoriser ou Refuser dans l'onglet. Ctrl-D quitte le programme.

Lors de l'exécution ci-dessus, l'onglet du navigateur affichait chaque appel sur une ligne, ainsi que la commande en attente d'une réponse :

À la fin du tour, l'onglet affichait le résumé de l'agent :

La modification apportée au dépôt, affichée par git diff :

• Exécutez les commandes dans un conteneur plutôt que sur votre machine : fournissez à l'Agent une LocalDockerSandbox.
• Demandez aussi une autorisation avant les modifications : remplacez la valeur de decision dans la première règle par "ask".
• Autorisez davantage de commandes sans confirmation, par exemple votre outil d'exécution des tests : ajoutez-les à READ_ONLY.