Construire un agent de nettoyage pour un dépôt privé

Construire un agent de codage qui améliore le code d'un dépôt GitHub privé et ouvre une pull request. Vous définissez l'agent dans un projet et l'appelez depuis un projet client distinct.

L'agent travaille dans une nouvelle sandbox gérée. Sandbox Proxy permet à la sandbox d'utiliser votre connecteur GitHub sans y copier les identifiants GitHub. L'agent utilise le modèle mistral-large-4.

Une dernière section facultative montre comment ajouter un workflow personnalisé planifié à l'agent déployé. Le planificateur n'est pas nécessaire pour construire, tester ou déployer l'agent.

Cet exemple présente : les sandbox et Sandbox Proxy, un délai d'expiration pour chaque étape, les clés d'idempotence, la clôture d'une session, la mise à disposition et le déploiement d'un agent.

Exécutez (après la configuration ci-dessous), depuis le dossier client/ de l'exemple : uv run --frozen --env-file .env python run_agent.py (configuration).

Le client crée une session auprès de l'agent et envoie la tâche de nettoyage. L'agent clone le dépôt, crée une branche, effectue des modifications cohérentes pour nettoyer le code, exécute des vérifications ciblées et ouvre une pull request.

client/ → agent/ → MistralSandbox → Sandbox Proxy → private GitHub repository
                                                        ↓
                                                  pull request

client/ → agent/ → MistralSandbox → Sandbox Proxy → private GitHub repository
                                                        ↓
                                                  pull request

L'exemple sépare les responsabilités dans des dossiers distincts :

Le dossier de l'exemple a la structure suivante :

private_repository_cleanup_agent/
├── README.md
├── agent/
│   ├── Dockerfile
│   ├── instructions.md
│   ├── main.py
│   └── pyproject.toml
└── client/
    ├── cleanup_task.py
    ├── run_agent.py
    ├── scheduled_review.py
    ├── worker.py
    ├── Dockerfile
    └── pyproject.toml

private_repository_cleanup_agent/
├── README.md
├── agent/
│   ├── Dockerfile
│   ├── instructions.md
│   ├── main.py
│   └── pyproject.toml
└── client/
    ├── cleanup_task.py
    ├── run_agent.py
    ├── scheduled_review.py
    ├── worker.py
    ├── Dockerfile
    └── pyproject.toml

Vous avez besoin de Python 3.12, de uv et d'une clé MISTRAL_API_KEY. Cette clé authentifie l'agent local et le client, et crée également la sandbox gérée. Pour rendre l'agent accessible ou le déployer, l'utilisateur de la clé doit avoir un rôle dans l'espace de travail donnant accès à Workflows, comme Developer : consultez Déployer. Pour la sandbox, votre identité Mistral doit également avoir le rôle cloud_user dans l'espace de travail (Mistral Cloud User dans la console). Consultez Prérequis pour l'obtenir.

Dans Mistral Studio, activez le connecteur GitHub. Le compte GitHub connecté doit pouvoir cloner le dépôt, pousser une branche et ouvrir une pull request.

Configurez le projet de l'agent :

cd agent
cp .env.example .env

cd agent
cp .env.example .env

Ajoutez votre clé API dans agent/.env :

MISTRAL_API_KEY=...

MISTRAL_API_KEY=...

Ouvrez agent/main.py et remplacez your-org/your-private-repo par votre dépôt privé :

sandbox = agents.MistralSandbox(
    network=agents.Network.all(),
    header_bindings=(
        # Sandbox Proxy gets a token from the GitHub access of the user who starts the session.
        agents.GitHubBinding(repos=("your-org/your-private-repo",)),
    ),
)

sandbox = agents.MistralSandbox(
    network=agents.Network.all(),
    header_bindings=(
        # Sandbox Proxy gets a token from the GitHub access of the user who starts the session.
        agents.GitHubBinding(repos=("your-org/your-private-repo",)),
    ),
)

Sandbox Proxy obtient un token à courte durée de vie auprès de votre connecteur GitHub et l'ajoute au trafic GitHub de la sandbox. repos limite Sandbox Proxy à ce dépôt. Ce paramètre n'accorde pas d'autorisations GitHub. L'identité GitHub connectée dans Mistral Studio doit toujours disposer des autorisations nécessaires pour créer des branches et des pull requests.

Remplacez le dépôt de la même manière dans client/cleanup_task.py. Le client inclut le dépôt dans le prompt de la tâche et valide l'URL de la pull request renvoyée par l'agent.

La définition de l'agent reste dans agent/main.py :

agent = agents.Agent(
    model="mistral-large-4",
    name=AGENT_NAME,
    instructions=(PROJECT_DIR / "instructions.md").read_text(encoding="utf-8"),
    environment=sandbox,
    max_iterations=60,
    temperature=0.1,
)

agent = agents.Agent(
    model="mistral-large-4",
    name=AGENT_NAME,
    instructions=(PROJECT_DIR / "instructions.md").read_text(encoding="utf-8"),
    environment=sandbox,
    max_iterations=60,
    temperature=0.1,
)

Le fichier agent/pyproject.toml épingle le SDK avec l'extra koyeb, dont la sandbox gérée a besoin.

Les instructions demandent à l'agent de lire les consignes du dépôt avant de le modifier. Elles lui imposent aussi de modifier le code, d'exécuter des vérifications ciblées, de créer une branche horodatée et d'ouvrir une pull request. Elles interdisent les mises à jour des dépendances, les migrations, les fichiers générés, les modifications de l'authentification, les modifications du déploiement et les pull requests qui ne portent que sur la documentation.

Installez le projet de l'agent et examinez sa configuration :

uv sync
uv run --frozen --env-file .env agents info

uv sync
uv run --frozen --env-file .env agents info

Depuis agent/, démarrez l'agent et laissez-le en cours d'exécution :

uv run --frozen --env-file .env agents serve

uv run --frozen --env-file .env agents serve

Le projet de l'agent ne contient aucun code client ni de planification. Son Dockerfile, utilisé par agents deploy pour construire l'image, ne démarre lui aussi que l'agent.

Ouvrez un autre terminal, accédez au dossier client/ de l'exemple et configurez le projet client :

cp .env.example .env
uv sync

cp .env.example .env
uv sync

Définissez MISTRAL_API_KEY et conservez pour AGENT_NAME le même nom que dans agent/main.py :

MISTRAL_API_KEY=...
AGENT_NAME=private-repository-cleanup-agent

MISTRAL_API_KEY=...
AGENT_NAME=private-repository-cleanup-agent

Le client trouve l'agent par son nom et exécute le traitement de nettoyage dans client/cleanup_task.py. Le workflow planifié ci-dessous exécute la même fonction :

async def run_cleanup(agent: agents.Agent, job: str) -> str:
    """Run the cleanup in one session until the agent returns a pull request URL.

    `job` names this run. The session start and each message carry an idempotency key made from
    it, so running the same job again after a crash continues the same session and turns, and
    sends nothing twice.
    """
    session = await agent.start_session(idempotency_key=job)
    try:
        prompt = cleanup_prompt()
        replies = []
        for step in range(MAX_TURNS):
            # timeout= stops the wait, not the turn: a turn still running then comes back as Running.
            match await session.run(prompt, idempotency_key=f"{job}:{step}", timeout=TURN_TIMEOUT_SECONDS):
                case agents.Completed(text=reply):
                    replies.append(reply or "")
                    pull_request_url = extract_pull_request_url(reply or "")
                    if pull_request_url is not None:
                        return pull_request_url
                    prompt = CONTINUE_PROMPT
                case agents.Running():
                    raise RuntimeError(f"A turn ran for more than {TURN_TIMEOUT_SECONDS // 60} minutes.")
                case agents.Blocked(callbacks=callbacks):
                    # The agent asks for no approval, so a callback here needs a person.
                    raise RuntimeError(f"The agent waits for a person: {[callback.kind for callback in callbacks]}")
                case agents.Failed(error=error):
                    raise RuntimeError(f"The agent turn failed: {error.message}")
                case state:
                    raise RuntimeError(f"The agent turn did not complete: {state!r}")

        raise RuntimeError(
            f"The agent used {MAX_TURNS} turns without returning a pull request URL. "
            f"Its replies were:\n{'\n\n'.join(replies)}"
        )
    finally:
        # The job owns its session: end it, which also stops a turn that is still running.
        await session.cancel()

async def run_cleanup(agent: agents.Agent, job: str) -> str:
    """Run the cleanup in one session until the agent returns a pull request URL.

    `job` names this run. The session start and each message carry an idempotency key made from
    it, so running the same job again after a crash continues the same session and turns, and
    sends nothing twice.
    """
    session = await agent.start_session(idempotency_key=job)
    try:
        prompt = cleanup_prompt()
        replies = []
        for step in range(MAX_TURNS):
            # timeout= stops the wait, not the turn: a turn still running then comes back as Running.
            match await session.run(prompt, idempotency_key=f"{job}:{step}", timeout=TURN_TIMEOUT_SECONDS):
                case agents.Completed(text=reply):
                    replies.append(reply or "")
                    pull_request_url = extract_pull_request_url(reply or "")
                    if pull_request_url is not None:
                        return pull_request_url
                    prompt = CONTINUE_PROMPT
                case agents.Running():
                    raise RuntimeError(f"A turn ran for more than {TURN_TIMEOUT_SECONDS // 60} minutes.")
                case agents.Blocked(callbacks=callbacks):
                    # The agent asks for no approval, so a callback here needs a person.
                    raise RuntimeError(f"The agent waits for a person: {[callback.kind for callback in callbacks]}")
                case agents.Failed(error=error):
                    raise RuntimeError(f"The agent turn failed: {error.message}")
                case state:
                    raise RuntimeError(f"The agent turn did not complete: {state!r}")

        raise RuntimeError(
            f"The agent used {MAX_TURNS} turns without returning a pull request URL. "
            f"Its replies were:\n{'\n\n'.join(replies)}"
        )
    finally:
        # The job owns its session: end it, which also stops a turn that is still running.
        await session.cancel()

session.run() attend la fin du tour, pendant 45 minutes au maximum : timeout= arrête l'attente, pas le tour. Un tour encore en cours renvoie donc le statut Running. Si l'agent a besoin d'un autre tour pour terminer la branche ou la pull request, le traitement continue dans la même session et la même sandbox. Le traitement gère sa session et la clôture lorsqu'il a terminé, ce qui arrête également tout tour encore en cours.

Le démarrage de la session et chaque message comportent une clé d'idempotence construite à partir de l'identifiant du traitement qu'affiche run_agent.py. Une erreur ou Ctrl-C clôture la session. Si le processus s'arrête avant de pouvoir le faire, par exemple s'il est interrompu de force ou si la machine redémarre, python run_agent.py --job <id> reprend la même session sans envoyer quoi que ce soit deux fois.

Exécutez le client :

uv run --frozen --env-file .env python run_agent.py

uv run --frozen --env-file .env python run_agent.py

La commande se termine en affichant l'URL d'une pull request :

https://github.com/your-org/your-private-repo/pull/123

https://github.com/your-org/your-private-repo/pull/123

Chaque exécution crée une nouvelle branche agent-cleanup/<UTC timestamp>-<short-name>. Une pull request de nettoyage précédente peut rester ouverte sans bloquer l'exécution suivante.

Créez d'abord un commit contenant agent/ et poussez-le : la plateforme construit l'image à partir du commit poussé. Déployez ensuite depuis le répertoire agent/ :

cd ../agent
uv run --frozen --env-file .env agents deploy . \
  --name private-repository-cleanup-agent \
  --hardened

cd ../agent
uv run --frozen --env-file .env agents deploy . \
  --name private-repository-cleanup-agent \
  --hardened

L'agent déployé crée la sandbox avec son propre compte de service. Le déploiement attribue à ce compte le rôle cloud_user. Vous devez donc disposer vous-même de ce rôle.

Conservez --hardened. Sandbox Proxy utilise l'identité de l'utilisateur qui démarre la session de l'agent. Cet utilisateur doit avoir configuré le connecteur GitHub dans Mistral Studio.

Le client ne change pas lorsque l'agent est déployé. Définissez AGENT_NAME dans client/.env avec le nom de l'agent déployé, puis exécutez à nouveau client/run_agent.py.

Vous pouvez exécuter le même traitement de nettoyage selon une planification depuis un déploiement Workflows distinct.

# Up to four turns of up to 45 minutes each. A retry runs the same job: after a worker crash, noticed
# once start_to_close_timeout has passed, it continues the same session and turns instead of
# starting a second cleanup; after an error, which ended the session, it fails at once.
@workflows.activity(
    start_to_close_timeout=timedelta(hours=3, minutes=10),
    retry_policy_max_attempts=3,
)
async def run_scheduled_cleanup(job: str) -> str:
    import os  # noqa: PLC0415

    from cleanup_task import AGENT_NAME, run_cleanup  # noqa: PLC0415

    from mistralai import agents  # noqa: PLC0415

    api_key = os.environ.get(SCHEDULED_AGENT_API_KEY_ENV_VAR, "").strip()
    if not api_key:
        raise RuntimeError(
            f"Bind a user API key to {SCHEDULED_AGENT_API_KEY_ENV_VAR}. "
            "The user must have access to the configured GitHub connector."
        )

    deployed_agent = agents.Agent.load_from(
        agent_name=AGENT_NAME,
        api_key=api_key,
    )
    return await run_cleanup(deployed_agent, job)

# Up to four turns of up to 45 minutes each. A retry runs the same job: after a worker crash, noticed
# once start_to_close_timeout has passed, it continues the same session and turns instead of
# starting a second cleanup; after an error, which ended the session, it fails at once.
@workflows.activity(
    start_to_close_timeout=timedelta(hours=3, minutes=10),
    retry_policy_max_attempts=3,
)
async def run_scheduled_cleanup(job: str) -> str:
    import os  # noqa: PLC0415

    from cleanup_task import AGENT_NAME, run_cleanup  # noqa: PLC0415

    from mistralai import agents  # noqa: PLC0415

    api_key = os.environ.get(SCHEDULED_AGENT_API_KEY_ENV_VAR, "").strip()
    if not api_key:
        raise RuntimeError(
            f"Bind a user API key to {SCHEDULED_AGENT_API_KEY_ENV_VAR}. "
            "The user must have access to the configured GitHub connector."
        )

    deployed_agent = agents.Agent.load_from(
        agent_name=AGENT_NAME,
        api_key=api_key,
    )
    return await run_cleanup(deployed_agent, job)

Exécutez uv sync depuis agent/ et conservez l'extra koyeb pour mistralai-agents dans son fichier pyproject.toml : la sandbox gérée en a besoin.

Votre identité Mistral ne dispose pas du rôle cloud_user dans l'espace de travail (Mistral Cloud User dans la console). Obtenez-le comme indiqué dans Prérequis, puis redémarrez la session. Pour un agent déployé, effectuez un nouveau déploiement après avoir obtenu le rôle.

La sandbox n'a pas démarré. L'erreur de la session en indique la cause. Si vous définissez une image privée, consultez Image et configuration pour connaître le secret de registre nécessaire.

Vérifiez que le connecteur GitHub est activé pour l'identité Mistral associée à la session. Confirmez que le compte GitHub connecté peut accéder au dépôt indiqué dans repos.

Le compte GitHub connecté dispose d'un accès en lecture, mais pas en écriture, ou une règle de l'organisation bloque la branche ou la pull request. Attribuez au compte le rôle requis sur le dépôt ou utilisez un dépôt où il peut créer des branches et des pull requests.

Vérifiez que AGENT_NAME dans client/.env correspond au nom indiqué dans agent/main.py pour une exécution locale, ou au nom transmis à agents deploy pour un déploiement géré.

• Sandbox explique les sandbox gérées et Sandbox Proxy.
• Ce que l'agent peut utiliser montre comment associer des outils et des sandbox à un agent.
• Modes d'exécution compare l'exécution de l'agent dans votre processus à son déploiement avec Studio (auto-hébergé ou géré par Mistral).