Faire confiance aveuglément à l’output d’un tool (Anti-Pattern) + Correctifs + Code

from __future__ import annotations

import json
from typing import Any, Literal


class ToolOutputInvalid(RuntimeError):
  pass


def parse_json_strict(raw: str, *, max_chars: int) -> Any:
  """
  Strict parse with a size cap. The cap is a safety boundary.
  Typical API JSON payloads are ~1–10KB. 200_000 (~200KB) is a common cap that
  covers edge cases while preventing multi‑MB responses that blow up parsing
  and model context.
  """
  if len(raw) > max_chars:
      raise ToolOutputInvalid("tool_output_too_large")
  try:
      return json.loads(raw)
  except Exception as e:
      raise ToolOutputInvalid(f"invalid_json:{type(e).__name__}")


def require_json_content_type(content_type: str | None) -> None:
  if not content_type:
      raise ToolOutputInvalid("missing_content_type")
  if "application/json" not in content_type.lower():
      raise ToolOutputInvalid(f"unexpected_content_type:{content_type}")


# Example generic schema validation using Pydantic.
# pip install pydantic
from pydantic import BaseModel, Field, ValidationError


Plan = Literal["free", "pro", "enterprise"]


class UserProfile(BaseModel):
  user_id: str = Field(min_length=1)
  plan: Plan | None = None
  tags: list[str] = []


def fetch_profile(user_id: str, *, tools, max_chars: int = 200_000) -> UserProfile:
  resp = tools.call("http.get", args={"url": f"https://api.internal/users/{user_id}", "timeout_s": 10})  # (pseudo)

  require_json_content_type(resp.get("content_type"))
  obj = parse_json_strict(resp["body"], max_chars=max_chars)

  try:
      return UserProfile.model_validate(obj)
  except ValidationError as e:
      raise ToolOutputInvalid("schema_invalid") from e


def safe_profile_flow(user_id: str, *, tools, mode: str = "degrade") -> dict[str, Any]:
  """
  mode:
    - "fail_closed": stop immediately
    - "degrade": return a safe partial and skip writes
  """
  try:
      profile = fetch_profile(user_id, tools=tools)
      return {"status": "ok", "profile": profile.model_dump(), "stop_reason": "success"}
  except ToolOutputInvalid as e:
      if mode == "fail_closed":
          return {"status": "stopped", "stop_reason": "invalid_tool_output", "error": str(e)}

      # Degrade: do not write; return a safe partial. Optionally use last-known-good cache.
      cached = tools.cache_get(f"profile:{user_id}") if hasattr(tools, "cache_get") else None  # (pseudo)
      if cached:
          return {
              "status": "degraded",
              "stop_reason": "invalid_tool_output",
              "safe_mode": "skip_writes",
              "profile": {**cached, "_degraded": True},
              "message": "Upstream returned invalid data. Using cached profile and skipping writes.",
          }
      return {
          "status": "degraded",
          "stop_reason": "invalid_tool_output",
          "safe_mode": "skip_writes",
          "profile": None,
          "message": "Upstream returned invalid data. Skipping writes.",
      }

// Example generic schema validation using Zod.
// npm i zod
import { z } from "zod";

export class ToolOutputInvalid extends Error {}

export function requireJsonContentType(contentType) {
if (!contentType) throw new ToolOutputInvalid("missing_content_type");
if (!String(contentType).toLowerCase().includes("application/json")) {
  throw new ToolOutputInvalid("unexpected_content_type:" + contentType);
}
}

export function parseJsonStrict(raw, { maxChars }) {
if (String(raw).length > maxChars) throw new ToolOutputInvalid("tool_output_too_large");
try {
  return JSON.parse(raw);
} catch (e) {
  throw new ToolOutputInvalid("invalid_json:" + (e?.name || "Error"));
}
}

const UserProfile = z.object({
user_id: z.string().min(1),
plan: z.enum(["free", "pro", "enterprise"]).optional(),
tags: z.array(z.string()).default([]),
});

export async function fetchProfile(userId, { tools, maxChars = 200000 }) {
const resp = await tools.call("http.get", { args: { url: "https://api.internal/users/" + userId, timeout_s: 10 } }); // (pseudo)
requireJsonContentType(resp.content_type);
const obj = parseJsonStrict(resp.body, { maxChars });

const parsed = UserProfile.safeParse(obj);
if (!parsed.success) throw new ToolOutputInvalid("schema_invalid");
return parsed.data;
}

export async function safeProfileFlow(userId, { tools, mode = "degrade" }) {
try {
  const profile = await fetchProfile(userId, { tools });
  return { status: "ok", profile, stop_reason: "success" };
} catch (e) {
  if (!(e instanceof ToolOutputInvalid)) throw e;
  if (mode === "fail_closed") return { status: "stopped", stop_reason: "invalid_tool_output", error: String(e.message) };

  // Degrade: do not write; return a safe partial. Optionally use last-known-good cache.
  const cached = typeof tools?.cacheGet === "function" ? await tools.cacheGet("profile:" + userId) : null; // (pseudo)
  if (cached) {
    return {
      status: "degraded",
      stop_reason: "invalid_tool_output",
      safe_mode: "skip_writes",
      profile: { ...cached, _degraded: true },
      message: "Upstream returned invalid data. Using cached profile and skipping writes.",
    };
  }
  return {
    status: "degraded",
    stop_reason: "invalid_tool_output",
    safe_mode: "skip_writes",
    profile: null,
    message: "Upstream returned invalid data. Skipping writes.",
  };
}
}