1"""execute_python — run Python in the sandbox-runner, collect outputs and figures."""23from __future__ import annotations45import base646from typing import Any78from pydantic import BaseModel, Field910from app.core.ratelimit import MSG_SANDBOX, limiter11from app.llm.schemas import Artifact, ToolResult12from app.sandbox.client import SandboxClient13from app.services import files as file_service14from app.tools.registry import ToolContext, registry151617class ExecArgs(BaseModel):18 code: str = Field(..., description="Python source code")19 description: str = Field("", max_length=300)20 heavy: bool = False212223IMAGE_EXT = {".png", ".jpg", ".jpeg", ".svg"}242526async def run(args: dict[str, Any], ctx: ToolContext) -> ToolResult:27 if ctx.role == "student":28 limiter.check(f"sandbox:{ctx.user_id}", ctx.settings.RATE_SANDBOX_PER_HOUR, 3600,29 MSG_SANDBOX)30 await ctx.report("running", "Exécution du code Python…")31 client = SandboxClient(ctx.settings)32 files_in = await file_service.sandbox_inputs(ctx.user_id, ctx.file_ids)33 timeout = ctx.settings.SANDBOX_HEAVY_TIMEOUT_S if args.get("heavy") else None34 res = await client.run(args["code"], files_in, timeout)35 if res.error and res.exit_code == 127:36 return ToolResult(content=f"Le sandbox n'a pas pu exécuter le code : {res.error}. "37 "Explique le calcul à la main ou réessaie plus tard.", error=True,38 payload={"code": args["code"], "stderr": res.error})3940 artifacts: list[Artifact] = []41 file_lines: list[str] = []42 for f in res.files_out:43 name = f.get("name", "sortie")44 data = base64.b64decode(f.get("content_b64", "")) if f.get("content_b64") else b""45 if not data:46 continue47 ext = "." + name.rsplit(".", 1)[-1].lower() if "." in name else ""48 ftype = "image" if ext in IMAGE_EXT else ("xlsx" if ext == ".xlsx" else49 "csv" if ext == ".csv" else "file")50 stored = await file_service.store_artifact(51 user_id=ctx.user_id, conversation_id=ctx.conversation_id, filename=name,52 data=data, ftype=ftype)53 artifacts.append(Artifact(type=ftype, file_id=stored.id, filename=name,54 url=f"/api/v1/files/{stored.id}"))55 file_lines.append(f"- {name} ({len(data)} octets)")5657 status = "ok" if res.exit_code == 0 else f"exit {res.exit_code}"58 content = (f"Exécution terminée ({status}, {res.duration_ms} ms).\n"59 f"STDOUT:\n{res.stdout or '(vide)'}\n")60 if res.stderr:61 content += f"STDERR:\n{res.stderr}\n"62 if file_lines:63 content += "Fichiers produits (déjà affichés à l'étudiant) :\n" + "\n".join(file_lines)64 if res.truncated:65 content += "\n(sortie tronquée)"66 return ToolResult(67 content=content, artifacts=artifacts, error=res.exit_code != 0,68 payload={"code": args["code"], "description": args.get("description", ""),69 "stdout": res.stdout, "stderr": res.stderr, "exit_code": res.exit_code,70 "duration_ms": res.duration_ms, "artifacts": [a.to_dict() for a in artifacts]},71 meta={"summary": args.get("description") or "Exécution Python"},72 )737475registry.register("execute_python", run, ExecArgs, heavy=True)76