Script Tools / Rubric Score

Source: examples/tools/script_tools/rubric_score.py

Introduction

This script focuses on the practical so-what for tool integration: a deterministic, inspectable JSON contract that keeps scripted scoring auditable in agent workflows.

Technical Implementation

  1. Read one JSON payload from stdin and compute rubric score fields.

  2. Write report + trace artifacts under artifacts/.

  3. Emit one JSON envelope on stdout with ok/result/artifacts/warnings/error.

        flowchart LR
    A["JSON input payload"] --> B["rubric_score.py"]
    B --> C["Score + artifacts"]
    C --> D["JSON envelope on stdout"]
    
 1from __future__ import annotations
 2
 3import json
 4import os
 5import sys
 6from datetime import datetime, timedelta, timezone
 7from pathlib import Path
 8
 9UTC = datetime.UTC if hasattr(datetime, "UTC") else timezone(timedelta(0))
10
11
12def _sanitize(value: str) -> str:
13    safe = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "_" for ch in value)
14    return safe or "example_script"
15
16
17def _write_trace(*, request_id: str, payload: dict[str, object]) -> str:
18    trace_dir = Path("artifacts/examples/traces")
19    trace_dir.mkdir(parents=True, exist_ok=True)
20    timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
21    trace_path = trace_dir / f"run_{timestamp}_{_sanitize(request_id)}.jsonl"
22    event = {
23        "event_type": "ScriptToolCompleted",
24        "run_id": request_id,
25        "timestamp": datetime.now(UTC).isoformat(),
26        "attributes": payload,
27    }
28    trace_path.write_text(json.dumps(event, ensure_ascii=True) + "\n", encoding="utf-8")
29    return str(trace_path)
30
31
32def main() -> int:
33    """Run rubric scoring over stdin JSON input and emit a tool envelope."""
34    raw_input_text = sys.stdin.read()
35    input_payload = json.loads(raw_input_text) if raw_input_text.strip() else {}
36    rubric_text = str(input_payload.get("text", ""))
37    max_score = int(input_payload.get("max_score", 10))
38
39    score = min(max_score, max(0, len(rubric_text.split()) // 5))
40    artifact_dir = Path("artifacts") / "rubric_score"
41    artifact_dir.mkdir(parents=True, exist_ok=True)
42    artifact_path = artifact_dir / "rubric_score_report.json"
43    report_payload = {
44        "score": score,
45        "max_score": max_score,
46        "word_count": len(rubric_text.split()),
47    }
48    artifact_path.write_text(json.dumps(report_payload, indent=2) + "\n", encoding="utf-8")
49
50    request_id = str(input_payload.get("request_id", "example-script-rubric-score-001"))
51    trace_path = _write_trace(request_id=request_id, payload=report_payload)
52
53    envelope = {
54        "ok": True,
55        "result": {
56            "score": score,
57            "max_score": max_score,
58            "cwd": os.getcwd(),
59            "trace_path": trace_path,
60        },
61        "artifacts": [
62            {
63                "path": str(artifact_path),
64                "mime": "application/json",
65            },
66            {
67                "path": trace_path,
68                "mime": "application/x-ndjson",
69            },
70        ],
71        "warnings": [],
72        "error": None,
73    }
74    print(json.dumps(envelope, ensure_ascii=True))
75    return 0
76
77
78if __name__ == "__main__":
79    raise SystemExit(main())

Expected Results

Run Command

PYTHONPATH=src python3 examples/tools/script_tools/rubric_score.py

  • Reads one JSON object from stdin.

  • Prints one JSON envelope containing ok/result/artifacts.

  • Writes artifacts under artifacts/rubric_score and artifacts/examples/traces.