Drive Py Shift from your own code
Everything the web page does is available over HTTP: paste a Python project's packaging and tooling
configuration in, get the same structured migration plan onto uv, ruff, ty, pytest, prek and a real
security lane back. The natural use is a CI job that re-plans whenever pyproject.toml
changes and fails the build when the posture regresses, or a script that runs the same twelve-check
pass across every repository in an organisation and reports which ones are still on Poetry.
Base URL and the envelope
Every endpoint lives under https://api.skillsafe.ai/v1/app-api and every response uses
the same envelope, so one helper covers the whole API:
{ "ok": true, "data": { ... } }
{ "ok": false, "error": { "code": "...", "message": "...", "status": 402, "details": { ... } } }
Send your app slug as X-App-Slug: py-shift and your token as
Authorization: Bearer … on every call.
Error codes
| code | status | what to do |
|---|---|---|
unauthorized | 401 | The token is missing, malformed or expired. Get a new one from the token page. |
payment_required | 402 | The balance is below min_credits. Call /estimate first and top up. |
forbidden | 403 | The token is valid but not for this app, or a guest token tried a metered run. Check the X-App-Slug header. |
not_found | 404 | Unknown job id, unknown collection, or the app slug does not exist. |
conflict | 409 | The same Idempotency-Key was replayed with a different body. Change the key or send the original input. |
validation_error | 422 | The input object is missing a required field — files is the usual one — or a field is the wrong type. A body that is not valid JSON at all comes back as a 400. |
rate_limited | 429 | Too many requests. Back off and retry; do not tight-loop. |
internal | 5xx | A server-side failure, reported as server_error on a plain 500. Retry with the SAME Idempotency-Key so you are not billed twice. |
1. Get a token
The easiest route is the token page: it shows the token this browser already holds, with Copy token and Copy shell export buttons, and a sign-in button for a personal token. Nothing on that page needs a developer tool — it reads the same storage the app itself uses and prints the token for you.
A guest token can call /me and /estimate. Planning a
migration is metered, so it needs a personal token from signing in.
# The token page is the shortest path. It shows the token this browser holds and
# hands you a ready-made shell export:
#
# https://py-shift.skillsafe.ai/tokens.html
# export SKILLSAFE_TOKEN="..."
#
# To mint a guest token from the command line instead. A guest token is enough
# for /me and /estimate; planning a migration needs a personal token from signing in.
curl -sS -X POST "https://api.skillsafe.ai/v1/app-api/guest" -H "X-App-Slug: py-shift"
# {"ok":true,"data":{"token":"sk_guest_...","subject_type":"guest"}}
# Open https://py-shift.skillsafe.ai/tokens.html and press "Copy token",
# or mint a guest token here. A guest token can call /me and /estimate but
# cannot run a metered plan.
import json, urllib.request
req = urllib.request.Request(
"https://api.skillsafe.ai/v1/app-api/guest", data=b"{}", method="POST")
req.add_header("X-App-Slug", "py-shift")
req.add_header("Content-Type", "application/json")
with urllib.request.urlopen(req) as r:
TOKEN = json.load(r)["data"]["token"]
// Open https://py-shift.skillsafe.ai/tokens.html and press "Copy token",
// or mint a guest token here. A guest token can call /me and /estimate but
// cannot run a metered plan.
const res = await fetch("https://api.skillsafe.ai/v1/app-api/guest", {
method: "POST",
headers: { "X-App-Slug": "py-shift", "Content-Type": "application/json" },
body: "{}",
});
const TOKEN = (await res.json()).data.token;
// Open https://py-shift.skillsafe.ai/tokens.html and press "Copy token",
// or mint a guest token here. A guest token can call /me and /estimate but
// cannot run a metered plan.
guestReq, _ := http.NewRequest(http.MethodPost,
"https://api.skillsafe.ai/v1/app-api/guest", bytes.NewReader([]byte("{}")))
guestReq.Header.Set("X-App-Slug", "py-shift")
guestReq.Header.Set("Content-Type", "application/json")
guestRes, err := http.DefaultClient.Do(guestReq)
if err != nil {
panic(err)
}
defer guestRes.Body.Close()
var guest struct {
Data struct {
Token string `json:"token"`
} `json:"data"`
}
_ = json.NewDecoder(guestRes.Body).Decode(&guest)
fmt.Println(guest.Data.Token)
// Open https://py-shift.skillsafe.ai/tokens.html and press "Copy token",
// or mint a guest token here. A guest token can call /me and /estimate but
// cannot run a metered plan.
var http = HttpClient.newHttpClient();
var guestReq = HttpRequest.newBuilder(URI.create("https://api.skillsafe.ai/v1/app-api/guest"))
.header("X-App-Slug", "py-shift")
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString("{}"))
.build();
HttpResponse<String> guest = http.send(guestReq, HttpResponse.BodyHandlers.ofString());
System.out.println(guest.body()); // {"ok":true,"data":{"token":"sk_guest_..."}}
# Open https://py-shift.skillsafe.ai/tokens.html and press "Copy token",
# or mint a guest token here. A guest token can call /me and /estimate but
# cannot run a metered plan.
require "json"
require "net/http"
require "uri"
uri = URI("https://api.skillsafe.ai/v1/app-api/guest")
req = Net::HTTP::Post.new(uri)
req["X-App-Slug"] = "py-shift"
req["Content-Type"] = "application/json"
req.body = "{}"
res = Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) { |h| h.request(req) }
TOKEN = JSON.parse(res.body)["data"]["token"]
<?php
// Open https://py-shift.skillsafe.ai/tokens.html and press "Copy token",
// or mint a guest token here. A guest token can call /me and /estimate but
// cannot run a metered plan.
$ch = curl_init("https://api.skillsafe.ai/v1/app-api/guest");
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_POSTFIELDS, "{}");
curl_setopt($ch, CURLOPT_HTTPHEADER, [
"X-App-Slug: py-shift",
"Content-Type: application/json",
]);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$guest = json_decode(curl_exec($ch), true);
curl_close($ch);
echo $guest["data"]["token"];
// Open https://py-shift.skillsafe.ai/tokens.html and press "Copy token",
// or mint a guest token here. A guest token can call /me and /estimate but
// cannot run a metered plan.
using var http = new HttpClient();
var guestReq = new HttpRequestMessage(HttpMethod.Post, "https://api.skillsafe.ai/v1/app-api/guest");
guestReq.Headers.Add("X-App-Slug", "py-shift");
guestReq.Content = new StringContent("{}", Encoding.UTF8, "application/json");
var guestRes = await http.SendAsync(guestReq);
var guest = await guestRes.Content.ReadFromJsonAsync<JsonElement>();
Console.WriteLine(guest.GetProperty("data").GetProperty("token").GetString());
2. A tiny client
One helper that adds the headers, unwraps data and raises on error.
# Every call is the same three things: the base URL, your bearer token,
# and a JSON body. Keep the token in a shell variable.
BASE="https://api.skillsafe.ai/v1/app-api"
SLUG="py-shift"
TOKEN="$SKILLSAFE_TOKEN" # from https://py-shift.skillsafe.ai/tokens.html
call() { # call <path> [json-body]
if [ -n "$2" ]; then
curl -sS -X POST "$BASE/$1" \
-H "Authorization: Bearer $TOKEN" \
-H "X-App-Slug: $SLUG" \
-H "Content-Type: application/json" \
-d "$2"
else
curl -sS "$BASE/$1" -H "Authorization: Bearer $TOKEN" -H "X-App-Slug: $SLUG"
fi
}
import json, os, urllib.error, urllib.request
BASE = "https://api.skillsafe.ai/v1/app-api"
SLUG = "py-shift"
TOKEN = os.environ.get("SKILLSAFE_TOKEN", "YOUR_TOKEN") # from https://py-shift.skillsafe.ai/tokens.html
def call(path, body=None):
"""Returns the unwrapped `data`, or raises with the API error code."""
data = json.dumps(body).encode() if body is not None else None
req = urllib.request.Request(f"{BASE}/{path}", data=data, method="POST" if body is not None else "GET")
req.add_header("Authorization", f"Bearer {TOKEN}")
req.add_header("X-App-Slug", SLUG)
if body is not None:
req.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(req) as r:
payload = json.load(r)
except urllib.error.HTTPError as e:
payload = json.load(e)
if not payload.get("ok"):
err = payload.get("error", {})
raise RuntimeError(f"{err.get('code')}: {err.get('message')}")
return payload["data"]
const BASE = "https://api.skillsafe.ai/v1/app-api";
const SLUG = "py-shift";
const TOKEN = "YOUR_TOKEN"; // from https://py-shift.skillsafe.ai/tokens.html
async function call(path, body) {
const res = await fetch(`${BASE}/${path}`, {
method: body ? "POST" : "GET",
headers: {
Authorization: `Bearer ${TOKEN}`,
"X-App-Slug": SLUG,
...(body ? { "Content-Type": "application/json" } : {}),
},
body: body ? JSON.stringify(body) : undefined,
});
const payload = await res.json();
if (!payload.ok) throw new Error(`${payload.error.code}: ${payload.error.message}`);
return payload.data;
}
package main
import (
"bufio"
"bytes"
"crypto/sha256"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"strings"
"time"
)
const (
base = "https://api.skillsafe.ai/v1/app-api"
slug = "py-shift"
)
var token = os.Getenv("SKILLSAFE_TOKEN") // from https://py-shift.skillsafe.ai/tokens.html
type envelope struct {
OK bool `json:"ok"`
Data json.RawMessage `json:"data"`
Error struct {
Code string `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
func call(path string, body any) (json.RawMessage, error) {
method := http.MethodGet
var rdr io.Reader
if body != nil {
method = http.MethodPost
b, _ := json.Marshal(body)
rdr = bytes.NewReader(b)
}
req, _ := http.NewRequest(method, base+"/"+path, rdr)
req.Header.Set("Authorization", "Bearer "+token)
req.Header.Set("X-App-Slug", slug)
if body != nil {
req.Header.Set("Content-Type", "application/json")
}
res, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
defer res.Body.Close()
var env envelope
if err := json.NewDecoder(res.Body).Decode(&env); err != nil {
return nil, err
}
if !env.OK {
return nil, fmt.Errorf("%s: %s", env.Error.Code, env.Error.Message)
}
return env.Data, nil
}
import java.net.URI;
import java.net.http.*;
public class PyShift {
static final String BASE = "https://api.skillsafe.ai/v1/app-api";
static final String SLUG = "py-shift";
static final String TOKEN = System.getenv().getOrDefault("SKILLSAFE_TOKEN", "YOUR_TOKEN");
static final HttpClient HTTP = HttpClient.newHttpClient();
static String call(String path, String jsonBody) throws Exception {
HttpRequest.Builder b = HttpRequest.newBuilder(URI.create(BASE + "/" + path))
.header("Authorization", "Bearer " + TOKEN)
.header("X-App-Slug", SLUG);
if (jsonBody != null) {
b.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(jsonBody));
} else {
b.GET();
}
HttpResponse<String> res = HTTP.send(b.build(), HttpResponse.BodyHandlers.ofString());
// The envelope is always {"ok":true,"data":...} or {"ok":false,"error":...}.
return res.body();
}
}
require "json"
require "net/http"
require "uri"
BASE = "https://api.skillsafe.ai/v1/app-api"
SLUG = "py-shift"
TOKEN = ENV.fetch("SKILLSAFE_TOKEN", "YOUR_TOKEN") # from https://py-shift.skillsafe.ai/tokens.html
def call(path, body = nil)
uri = URI("#{BASE}/#{path}")
req = body ? Net::HTTP::Post.new(uri) : Net::HTTP::Get.new(uri)
req["Authorization"] = "Bearer #{TOKEN}"
req["X-App-Slug"] = SLUG
if body
req["Content-Type"] = "application/json"
req.body = JSON.generate(body)
end
res = Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) { |h| h.request(req) }
payload = JSON.parse(res.body)
raise "#{payload['error']['code']}: #{payload['error']['message']}" unless payload["ok"]
payload["data"]
end
<?php
const BASE = "https://api.skillsafe.ai/v1/app-api";
const SLUG = "py-shift";
define("TOKEN", getenv("SKILLSAFE_TOKEN") ?: "YOUR_TOKEN"); // from /tokens.html
function call(string $path, ?array $body = null) {
$ch = curl_init(BASE . "/" . $path);
$headers = ["Authorization: Bearer " . TOKEN, "X-App-Slug: " . SLUG];
if ($body !== null) {
$headers[] = "Content-Type: application/json";
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($body));
}
curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$payload = json_decode(curl_exec($ch), true);
curl_close($ch);
if (empty($payload["ok"])) {
throw new RuntimeException($payload["error"]["code"] . ": " . $payload["error"]["message"]);
}
return $payload["data"];
}
using System.Net.Http.Json;
using System.Text.Json;
static class PyShift
{
const string Base = "https://api.skillsafe.ai/v1/app-api";
const string Slug = "py-shift";
static readonly string Token =
Environment.GetEnvironmentVariable("SKILLSAFE_TOKEN") ?? "YOUR_TOKEN";
static readonly HttpClient Http = new();
public static async Task<JsonElement> Call(string path, object? body = null)
{
var req = new HttpRequestMessage(body is null ? HttpMethod.Get : HttpMethod.Post, $"{Base}/{path}");
req.Headers.Add("Authorization", $"Bearer {Token}");
req.Headers.Add("X-App-Slug", Slug);
if (body is not null) req.Content = JsonContent.Create(body);
var res = await Http.SendAsync(req);
var payload = await res.Content.ReadFromJsonAsync<JsonElement>();
if (!payload.GetProperty("ok").GetBoolean())
{
var e = payload.GetProperty("error");
throw new Exception($"{e.GetProperty("code")}: {e.GetProperty("message")}");
}
return payload.GetProperty("data");
}
}
3. Check the session and the balance
GET /me tells you whether the token is a guest or a person, and what the balance is.
subject_type is guest or user — a guest can
price a run but cannot start one — and credits is the wallet balance in credits.
Compare it against min_credits from the next step before you run, so a shortfall
surfaces as your own clear message rather than a 402.
call me
# {"ok":true,"data":{"subject_type":"user","username":"you","credits":51234}}
me = call("me")
print(me["subject_type"], me.get("credits"))
const me = await call("me");
console.log(me.subject_type, me.credits);
raw, err := call("me", nil)
if err != nil {
panic(err)
}
var me struct {
SubjectType string `json:"subject_type"`
Credits int `json:"credits"`
}
_ = json.Unmarshal(raw, &me)
fmt.Println(me.SubjectType, me.Credits)
System.out.println(call("me", null));
// {"ok":true,"data":{"subject_type":"user","username":"you","credits":51234}}
me = call("me")
puts "#{me['subject_type']} #{me['credits']}"
<?php
$me = call("me");
echo $me["subject_type"], " ", $me["credits"], PHP_EOL;
var me = await PyShift.Call("me");
Console.WriteLine(me.GetProperty("subject_type").GetString());
4. Price the run — free
The input object is exactly what the app's own form submits:
| field | type | meaning |
|---|---|---|
files | string, required | The pasted configuration — pyproject.toml, setup.py, setup.cfg, requirements*.txt, Pipfile, .pre-commit-config.yaml, mypy.ini, .flake8, tox.ini, a Makefile, a CI workflow, a Python source file or two. Any subset. Put a # file: pyproject.toml marker line above each one so they can be told apart. This is the plan's only evidence. A file whose middle has been removed should say so with a # [... clipped ...] comment. |
target | string | What the project is: package (distributable, published), application (deployed, not published), library (internal, imported by other repositories) or script (one or a few standalone files). It changes the src/ layout recommendation, whether the lockfile argument applies, and whether extras-versus-groups has a user-facing consequence. |
floor | string | The Python floor to plan toward: 3.11, 3.12, 3.13, or keep to plan around the current floor without raising it. With keep and a floor below 3.11, the plan says in assumptions which parts of the template that rules out. |
focus | string | general, dependencies, lint, types, testing, security or scripts. Emphasis, not exclusivity: a critical step from another phase is never suppressed. |
context | string, optional | Free-form notes: team size, CI provider, deadline, what has already been tried, what the team refuses to change. An explicit refusal — "we are keeping Poetry", "we cannot move off 3.9" — is respected and recorded in assumptions rather than argued with. |
prescan_facts | object | {resources: [{id,label}], flags: [{id,label}]} — see below. |
retry_note | string, optional | Send only on a retry, when a previous reply failed to parse or came back truncated. The instruction is obeyed exactly. |
prescan_facts, honestly
In the browser, prescan_facts is computed for free before the run by a local TOML and
INI reader: resources is what it read — the sections, the build backend, the declared
dependencies, the floor, the layout, the tools each config file configures — and flags
is the set of deterministic checks that fired. An API caller does not have to reproduce any of
that. Sending {"resources": [], "flags": []} is legitimate and the plan still works;
the model reads files either way.
What makes it worth sending is the reconciliation contract: every id you send in
flags must come back exactly once in coverage_check. That turns
a fact your own tooling already established into something the plan is held to. An entry with
addressed: false is a correct answer — the model deliberately setting a flag aside,
with the reason in note — and is a different thing from silence. A flag that never
appears at all is a failed run, not a passing one. Ids you did not send should not appear either.
These are the flag ids an API caller is most likely to want to raise by hand:
| id | fires when |
|---|---|
MGR-POETRY | Poetry manages the project — [tool.poetry], a poetry.lock, or a poetry-core build backend. |
SETUP-PY | A setup.py still carries metadata or a setup() call. |
REQ-TXT | Dependencies are declared in a requirements*.txt rather than in pyproject.toml. |
EXTRAS-DEV | Dev tooling is declared under [project.optional-dependencies], so it ships to users. |
NO-GROUPS | No [dependency-groups] table exists at all. |
FLOOR-LOW | requires-python is below the planned floor, or is missing. |
TOOL-BLACK | black is configured or declared as a dependency. |
TOOL-ISORT | isort is configured or declared as a dependency. |
TOOL-FLAKE8 | flake8 is configured — .flake8, setup.cfg or tox.ini. |
TOOL-MYPY | mypy is configured — [tool.mypy] or mypy.ini. |
NO-RUFF | ruff is nowhere in the paste. |
RUFF-NARROW | ruff is present but select is a short list such as ["E", "F"] — the flake8 defaults wearing a new name. |
NO-TY | No [tool.ty] configuration. |
TY-ENV-KEY | python-version sits at the top of [tool.ty] instead of under [tool.ty.environment], where it is silently ignored. |
LAYOUT-FLAT | The package sits at the repository root rather than under src/. |
NO-LOCK | No uv.lock, or only a foreign lockfile. |
NO-COV-FLOOR | Coverage runs without --cov-fail-under. |
PRECOMMIT-LEGACY | Hooks run through pre-commit rather than prek. |
NO-SECURITY | None of pip-audit, detect-secrets, actionlint or zizmor appears. |
NO-DEPENDABOT | No Dependabot configuration or equivalent update automation. |
UV-PIP-INSTALL | uv pip install is used as the workflow rather than as a compatibility shim. |
VENV-ACTIVATE | A source .venv/bin/activate line appears in a Makefile, a README or a CI job. |
SCRIPT-NO-PEP723 | A standalone script imports third-party packages without PEP 723 inline metadata. |
/estimate creates no job and charges nothing. It returns the model
binding — model, model_alias, markup_bps — and the
reservation: hold_credits is what gets held, min_credits is the balance
you must clear to start, and sponsor_enabled says whether the app is covering the run.
The hold is a reservation, not the price. It prices the full output cap, so the
charged_credits you see after settlement is usually far lower — often a small fraction
of the hold. Budget against hold_credits, report against charged_credits.
INPUT='{"files": "# file: pyproject.toml\n[tool.poetry]\nname = \"acme-cli\"\nversion = \"1.4.0\"\n\n[tool.poetry.dependencies]\npython = \"^3.9\"\nrequests = \"^2.31\"\n\n[tool.poetry.group.dev.dependencies]\nblack = \"^24.3\"\nflake8 = \"^7.0\"\nmypy = \"^1.9\"\n\n[build-system]\nrequires = [\"poetry-core\"]\nbuild-backend = \"poetry.core.masonry.api\"\n\n# file: Makefile\nlint:\n\tsource .venv/bin/activate; black .; flake8 acme_cli", "target": "package", "floor": "3.12", "focus": "dependencies", "context": "Four engineers, GitHub Actions, published to an internal index. We are willing to move off Poetry this quarter.", "prescan_facts": {"resources": [{"id": "RES-PYPROJECT", "label": "pyproject.toml, 4 tables, no [project]"}, {"id": "RES-MAKEFILE", "label": "Makefile with a lint target"}], "flags": [{"id": "MGR-POETRY", "label": "[tool.poetry] with a poetry-core build backend"}, {"id": "FLOOR-LOW", "label": "python = ^3.9 is below the 3.12 floor"}, {"id": "TOOL-BLACK", "label": "black declared in the dev group"}, {"id": "VENV-ACTIVATE", "label": "source .venv/bin/activate in Makefile:lint"}]}}'
call estimate "$INPUT"
# {"ok":true,"data":{"model":"gpt-5.6-terra","model_alias":"gpt-terra",
# "markup_bps":1000,"hold_credits":2140,"min_credits":310,"sponsor_enabled":false}}
#
# estimate is FREE. It creates no job and charges nothing. hold_credits is what
# gets RESERVED; charged_credits after settlement is normally much lower.
INPUT = {
"files": "# file: pyproject.toml\n[tool.poetry]\nname = \"acme-cli\"\nversion = \"1.4.0\"\n\n[tool.poetry.dependencies]\npython = \"^3.9\"\nrequests = \"^2.31\"\n\n[tool.poetry.group.dev.dependencies]\nblack = \"^24.3\"\nflake8 = \"^7.0\"\nmypy = \"^1.9\"\n\n[build-system]\nrequires = [\"poetry-core\"]\nbuild-backend = \"poetry.core.masonry.api\"\n\n# file: Makefile\nlint:\n\tsource .venv/bin/activate; black .; flake8 acme_cli",
"target": "package",
"floor": "3.12",
"focus": "dependencies",
"context": "Four engineers, GitHub Actions, published to an internal index.",
"prescan_facts": {
"resources": [
{"id": "RES-PYPROJECT", "label": "pyproject.toml, 4 tables, no [project]"},
{"id": "RES-MAKEFILE", "label": "Makefile with a lint target"}
],
"flags": [
{"id": "MGR-POETRY", "label": "[tool.poetry] with a poetry-core build backend"},
{"id": "FLOOR-LOW", "label": "python = ^3.9 is below the 3.12 floor"},
{"id": "TOOL-BLACK", "label": "black declared in the dev group"},
{"id": "VENV-ACTIVATE", "label": "source .venv/bin/activate in Makefile:lint"}
]
}
}
est = call("estimate", INPUT)
print(est["model"], est["model_alias"], est["markup_bps"])
print(est["hold_credits"], est["min_credits"], est["sponsor_enabled"])
# estimate is free: no job is created and nothing is charged. The hold is a
# reservation against the full output cap, not the price of the run.
const INPUT = {
files:
'# file: pyproject.toml\n[tool.poetry]\nname = "acme-cli"\nversion = "1.4.0"\n\n[tool.poetry.dependencies]\npython = "^3.9"\nrequests = "^2.31"\n\n[tool.poetry.group.dev.dependencies]\nblack = "^24.3"\nflake8 = "^7.0"\nmypy = "^1.9"\n\n[build-system]\nrequires = ["poetry-core"]\nbuild-backend = "poetry.core.masonry.api"\n\n# file: Makefile\nlint:\n\tsource .venv/bin/activate; black .; flake8 acme_cli',
target: "package",
floor: "3.12",
focus: "dependencies",
context: "Four engineers, GitHub Actions, published to an internal index.",
prescan_facts: {
resources: [{ id: "RES-PYPROJECT", label: "pyproject.toml, 4 tables, no [project]" }],
flags: [
{ id: "MGR-POETRY", label: "[tool.poetry] with a poetry-core build backend" },
{ id: "FLOOR-LOW", label: "python = ^3.9 is below the 3.12 floor" },
{ id: "TOOL-BLACK", label: "black declared in the dev group" },
{ id: "VENV-ACTIVATE", label: "source .venv/bin/activate in Makefile:lint" },
],
},
};
const est = await call("estimate", INPUT);
console.log(est.model, est.model_alias, est.markup_bps);
console.log(est.hold_credits, est.min_credits, est.sponsor_enabled);
// estimate is free: no job is created and nothing is charged. hold_credits is a
// reservation against the output cap; charged_credits is normally far lower.
input := map[string]any{
"files": "# file: pyproject.toml\n[tool.poetry]\nname = \"acme-cli\"\nversion = \"1.4.0\"\n\n[tool.poetry.dependencies]\npython = \"^3.9\"\nrequests = \"^2.31\"\n\n[tool.poetry.group.dev.dependencies]\nblack = \"^24.3\"\nflake8 = \"^7.0\"\n\n[build-system]\nrequires = [\"poetry-core\"]\nbuild-backend = \"poetry.core.masonry.api\"",
"target": "package",
"floor": "3.12",
"focus": "dependencies",
"context": "Four engineers, GitHub Actions, published to an internal index.",
"prescan_facts": map[string]any{
"resources": []any{map[string]string{"id": "RES-PYPROJECT", "label": "pyproject.toml, 4 tables, no [project]"}},
"flags": []any{
map[string]string{"id": "MGR-POETRY", "label": "[tool.poetry] with a poetry-core build backend"},
map[string]string{"id": "FLOOR-LOW", "label": "python = ^3.9 is below the 3.12 floor"},
map[string]string{"id": "TOOL-BLACK", "label": "black declared in the dev group"},
map[string]string{"id": "VENV-ACTIVATE", "label": "source .venv/bin/activate in Makefile:lint"},
},
},
}
raw, err := call("estimate", input)
if err != nil {
panic(err)
}
fmt.Println(string(raw)) // estimate is free - no job, no charge; the hold is a reservation
String input = """
{
"files": "# file: pyproject.toml\\n[tool.poetry]\\nname = \\"acme-cli\\"\\nversion = \\"1.4.0\\"\\n\\n[tool.poetry.dependencies]\\npython = \\"^3.9\\"\\nrequests = \\"^2.31\\"\\n\\n[tool.poetry.group.dev.dependencies]\\nblack = \\"^24.3\\"\\nflake8 = \\"^7.0\\"\\n\\n[build-system]\\nrequires = [\\"poetry-core\\"]\\nbuild-backend = \\"poetry.core.masonry.api\\"",
"target": "package",
"floor": "3.12",
"focus": "dependencies",
"context": "Four engineers, GitHub Actions, published to an internal index.",
"prescan_facts": {
"resources": [
{ "id": "RES-PYPROJECT", "label": "pyproject.toml, 4 tables, no [project]" }
],
"flags": [
{ "id": "MGR-POETRY", "label": "[tool.poetry] with a poetry-core build backend" },
{ "id": "FLOOR-LOW", "label": "python = ^3.9 is below the 3.12 floor" },
{ "id": "TOOL-BLACK", "label": "black declared in the dev group" },
{ "id": "VENV-ACTIVATE", "label": "source .venv/bin/activate in Makefile:lint" }
]
}
}
""";
System.out.println(call("estimate", input));
// estimate is free: no job is created and nothing is charged.
// The data object carries model, model_alias, markup_bps, hold_credits,
// min_credits and sponsor_enabled. hold_credits is a reservation against the
// full output cap, so the settled charge is normally far lower.
input = {
"files" => "# file: pyproject.toml\n[tool.poetry]\nname = \"acme-cli\"\nversion = \"1.4.0\"\n\n[tool.poetry.dependencies]\npython = \"^3.9\"\nrequests = \"^2.31\"\n\n[tool.poetry.group.dev.dependencies]\nblack = \"^24.3\"\nflake8 = \"^7.0\"\n\n[build-system]\nrequires = [\"poetry-core\"]\nbuild-backend = \"poetry.core.masonry.api\"",
"target" => "package",
"floor" => "3.12",
"focus" => "dependencies",
"context" => "Four engineers, GitHub Actions, published to an internal index.",
"prescan_facts" => {
"resources" => [{ "id" => "RES-PYPROJECT", "label" => "pyproject.toml, 4 tables, no [project]" }],
"flags" => [
{ "id" => "MGR-POETRY", "label" => "[tool.poetry] with a poetry-core build backend" },
{ "id" => "FLOOR-LOW", "label" => "python = ^3.9 is below the 3.12 floor" },
{ "id" => "TOOL-BLACK", "label" => "black declared in the dev group" },
{ "id" => "VENV-ACTIVATE", "label" => "source .venv/bin/activate in Makefile:lint" }
]
}
}
est = call("estimate", input)
puts "#{est['model']} hold=#{est['hold_credits']} min=#{est['min_credits']}"
# estimate is free: no job is created and nothing is charged.
<?php
$input = [
"files" => "# file: pyproject.toml\n[tool.poetry]\nname = \"acme-cli\"\nversion = \"1.4.0\"\n\n[tool.poetry.dependencies]\npython = \"^3.9\"\nrequests = \"^2.31\"\n\n[tool.poetry.group.dev.dependencies]\nblack = \"^24.3\"\nflake8 = \"^7.0\"\n\n[build-system]\nrequires = [\"poetry-core\"]\nbuild-backend = \"poetry.core.masonry.api\"",
"target" => "package",
"floor" => "3.12",
"focus" => "dependencies",
"context" => "Four engineers, GitHub Actions, published to an internal index.",
"prescan_facts" => [
"resources" => [["id" => "RES-PYPROJECT", "label" => "pyproject.toml, 4 tables, no [project]"]],
"flags" => [
["id" => "MGR-POETRY", "label" => "[tool.poetry] with a poetry-core build backend"],
["id" => "FLOOR-LOW", "label" => "python = ^3.9 is below the 3.12 floor"],
["id" => "TOOL-BLACK", "label" => "black declared in the dev group"],
["id" => "VENV-ACTIVATE", "label" => "source .venv/bin/activate in Makefile:lint"],
],
],
];
$est = call("estimate", $input);
echo $est["model"], " ", $est["hold_credits"], " ", $est["min_credits"], PHP_EOL;
// estimate is free: no job is created and nothing is charged.
var input = new
{
files = "# file: pyproject.toml\n[tool.poetry]\nname = \"acme-cli\"\nversion = \"1.4.0\"\n\n[tool.poetry.dependencies]\npython = \"^3.9\"\nrequests = \"^2.31\"\n\n[tool.poetry.group.dev.dependencies]\nblack = \"^24.3\"\nflake8 = \"^7.0\"\n\n[build-system]\nrequires = [\"poetry-core\"]\nbuild-backend = \"poetry.core.masonry.api\"",
target = "package",
floor = "3.12",
focus = "dependencies",
context = "Four engineers, GitHub Actions, published to an internal index.",
prescan_facts = new
{
resources = new[] { new { id = "RES-PYPROJECT", label = "pyproject.toml, 4 tables, no [project]" } },
flags = new[]
{
new { id = "MGR-POETRY", label = "[tool.poetry] with a poetry-core build backend" },
new { id = "FLOOR-LOW", label = "python = ^3.9 is below the 3.12 floor" },
new { id = "TOOL-BLACK", label = "black declared in the dev group" },
new { id = "VENV-ACTIVATE", label = "source .venv/bin/activate in Makefile:lint" }
}
}
};
var est = await PyShift.Call("estimate", input);
Console.WriteLine(est.GetProperty("hold_credits").GetInt32());
Console.WriteLine(est.GetProperty("sponsor_enabled").GetBoolean());
// estimate is free: no job is created and nothing is charged. The hold is a
// reservation against the output cap, not the price of the run.
5. Run it, then poll
POST /run returns a job_id; poll GET jobs/{job_id} until
status is succeeded or failed. The plan JSON is the string
at data.output.output. The terminal job also carries charged_credits —
the real price — and the truncated flag.
Always send an Idempotency-Key. It is not formally required by the
endpoint, and it is required in practice: derive it from the input as the web app does, a content
hash plus an attempt counter (py-shift:<hash>:a<attempt>). A retried
request carrying the same key returns the same job instead of billing a second run, which is what
makes a CI retry safe after a network blip. Replaying a key with a different body is a 409
conflict, so bump the attempt suffix whenever the input actually changed.
# Always send an Idempotency-Key derived from the input. A retried request with
# the same key returns the SAME job instead of billing a second run.
KEY="py-shift:$(printf '%s' "$INPUT" | shasum -a 256 | cut -c1-16):a1"
JOB=$(curl -sS -X POST "$BASE/run" \
-H "Authorization: Bearer $TOKEN" \
-H "X-App-Slug: $SLUG" \
-H "Content-Type: application/json" \
-H "Idempotency-Key: $KEY" \
-d "$INPUT" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["job_id"])')
# Poll until the job reaches a terminal status.
while :; do
OUT=$(call "jobs/$JOB")
STATUS=$(printf '%s' "$OUT" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["status"])')
[ "$STATUS" = "succeeded" ] && break
[ "$STATUS" = "failed" ] && echo "$OUT" && exit 1
sleep 2
done
# The terminal job looks like this:
# {"ok":true,"data":{"job_id":"job_...","status":"succeeded",
# "output":{"output":"{\"plan_name\":\"acme-cli — Poetry to uv migration\", ...}"},
# "charged_credits":486,"truncated":false}}
printf '%s' "$OUT" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["output"]["output"])'
import hashlib, time
# Always send an Idempotency-Key derived from the input: a retried request with
# the same key returns the SAME job instead of billing a second run.
digest = hashlib.sha256(json.dumps(INPUT, sort_keys=True).encode()).hexdigest()[:16]
key = f"py-shift:{digest}:a1"
req = urllib.request.Request(f"{BASE}/run", data=json.dumps(INPUT).encode(), method="POST")
req.add_header("Authorization", f"Bearer {TOKEN}")
req.add_header("X-App-Slug", SLUG)
req.add_header("Content-Type", "application/json")
req.add_header("Idempotency-Key", key)
with urllib.request.urlopen(req) as r:
job_id = json.load(r)["data"]["job_id"]
while True:
job = call(f"jobs/{job_id}")
if job["status"] == "succeeded":
break
if job["status"] == "failed":
raise RuntimeError(job.get("error"))
time.sleep(2)
plan = json.loads(job["output"]["output"])
print(plan["posture"], plan["target_python"], len(plan["steps"]), "steps")
print("charged", job.get("charged_credits"), "truncated", job.get("truncated"))
import { createHash } from "node:crypto";
// Always send an Idempotency-Key derived from the input: a retried request with
// the same key returns the SAME job instead of billing a second run.
const digest = createHash("sha256").update(JSON.stringify(INPUT)).digest("hex").slice(0, 16);
const key = `py-shift:${digest}:a1`;
const started = await fetch(`${BASE}/run`, {
method: "POST",
headers: {
Authorization: `Bearer ${TOKEN}`,
"X-App-Slug": SLUG,
"Content-Type": "application/json",
"Idempotency-Key": key,
},
body: JSON.stringify(INPUT),
}).then((r) => r.json());
let job = started.data;
while (job.status !== "succeeded" && job.status !== "failed") {
await new Promise((r) => setTimeout(r, 2000));
job = await call(`jobs/${job.job_id}`);
}
if (job.status === "failed") throw new Error(JSON.stringify(job.error));
const plan = JSON.parse(job.output.output);
console.log(plan.posture, plan.toolchain, plan.steps.length, "steps", job.charged_credits);
// Always send an Idempotency-Key derived from the input: a retried request with
// the same key returns the SAME job instead of billing a second run.
body, _ := json.Marshal(input)
sum := sha256.Sum256(body)
key := fmt.Sprintf("py-shift:%x:a1", sum[:8])
req, _ := http.NewRequest(http.MethodPost, base+"/run", bytes.NewReader(body))
req.Header.Set("Authorization", "Bearer "+token)
req.Header.Set("X-App-Slug", slug)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Idempotency-Key", key)
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
var started struct {
Data struct {
JobID string `json:"job_id"`
} `json:"data"`
}
_ = json.NewDecoder(res.Body).Decode(&started)
for {
raw, err := call("jobs/"+started.Data.JobID, nil)
if err != nil {
panic(err)
}
var job struct {
Status string `json:"status"`
Output struct {
Output string `json:"output"`
} `json:"output"`
ChargedCredits int `json:"charged_credits"`
Truncated bool `json:"truncated"`
}
_ = json.Unmarshal(raw, &job)
if job.Status == "succeeded" {
fmt.Println(job.Output.Output) // the plan JSON, as a string
fmt.Println(job.ChargedCredits, job.Truncated)
break
}
if job.Status == "failed" {
panic("run failed")
}
time.Sleep(2 * time.Second)
}
// Always send an Idempotency-Key derived from the input: a retried request with
// the same key returns the SAME job instead of billing a second run.
var digest = java.security.MessageDigest.getInstance("SHA-256")
.digest(input.getBytes(java.nio.charset.StandardCharsets.UTF_8));
var key = "py-shift:" + java.util.HexFormat.of().formatHex(digest).substring(0, 16) + ":a1";
var start = HttpRequest.newBuilder(URI.create(BASE + "/run"))
.header("Authorization", "Bearer " + TOKEN)
.header("X-App-Slug", SLUG)
.header("Content-Type", "application/json")
.header("Idempotency-Key", key)
.POST(HttpRequest.BodyPublishers.ofString(input))
.build();
String started = HTTP.send(start, HttpResponse.BodyHandlers.ofString()).body();
// Parse job_id out of `started`, then poll GET jobs/{job_id} every two seconds
// until status is "succeeded" or "failed". The plan JSON is data.output.output,
// and the terminal job also carries charged_credits and truncated.
System.out.println(started);
require "digest"
# Always send an Idempotency-Key derived from the input: a retried request with
# the same key returns the SAME job instead of billing a second run.
digest = Digest::SHA256.hexdigest(JSON.generate(input))[0, 16]
key = "py-shift:#{digest}:a1"
uri = URI("#{BASE}/run")
req = Net::HTTP::Post.new(uri)
req["Authorization"] = "Bearer #{TOKEN}"
req["X-App-Slug"] = SLUG
req["Content-Type"] = "application/json"
req["Idempotency-Key"] = key
req.body = JSON.generate(input)
res = Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) { |h| h.request(req) }
job_id = JSON.parse(res.body)["data"]["job_id"]
loop do
job = call("jobs/#{job_id}")
if job["status"] == "succeeded"
puts job["output"]["output"]
puts "charged=#{job['charged_credits']} truncated=#{job['truncated']}"
break
end
raise "run failed" if job["status"] == "failed"
sleep 2
end
<?php
// Always send an Idempotency-Key derived from the input: a retried request with
// the same key returns the SAME job instead of billing a second run.
$digest = substr(hash("sha256", json_encode($input)), 0, 16);
$key = "py-shift:{$digest}:a1";
$ch = curl_init(BASE . "/run");
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($input));
curl_setopt($ch, CURLOPT_HTTPHEADER, [
"Authorization: Bearer " . TOKEN,
"X-App-Slug: " . SLUG,
"Content-Type: application/json",
"Idempotency-Key: " . $key,
]);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$jobId = json_decode(curl_exec($ch), true)["data"]["job_id"];
curl_close($ch);
while (true) {
$job = call("jobs/" . $jobId);
if ($job["status"] === "succeeded") {
echo $job["output"]["output"];
echo PHP_EOL, "charged=", $job["charged_credits"], " truncated=", var_export($job["truncated"], true), PHP_EOL;
break;
}
if ($job["status"] === "failed") { throw new RuntimeException("run failed"); }
sleep(2);
}
using System.Security.Cryptography;
using System.Text;
// Always send an Idempotency-Key derived from the input: a retried request with
// the same key returns the SAME job instead of billing a second run.
var json = JsonSerializer.Serialize(input);
var digest = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(json)))[..16].ToLowerInvariant();
var key = $"py-shift:{digest}:a1";
var run = new HttpRequestMessage(HttpMethod.Post, "https://api.skillsafe.ai/v1/app-api/run");
run.Headers.Add("Authorization", "Bearer YOUR_TOKEN");
run.Headers.Add("X-App-Slug", "py-shift");
run.Headers.Add("Idempotency-Key", key);
run.Content = JsonContent.Create(input);
// POST it, read data.job_id, then poll GET jobs/{job_id} every two seconds until
// status is "succeeded" or "failed". The plan JSON is data.output.output, and the
// terminal job also carries charged_credits and truncated.
6. Or stream it
POST /run-stream is the same call over server-sent events. Each delta
event carries {"text": "..."}, a chunk of the plan JSON, and the final
done event carries status, charged_credits — the real price,
normally a fraction of the hold — and the truncated flag.
The practical tip: the web app does not parse the partial JSON to drive its progress display, it
watches for key names arriving in the accumulating text. The appearance of "steps" or
"new_pyproject" is what advances the stage from working the twelve checks to ordering
the migration steps and then to rewriting pyproject.toml. Substring matching on the
quoted key name is enough, and it costs nothing.
# Server-sent events. Each `delta` carries a chunk of the JSON plan; the final
# `done` event carries the status, charged_credits and the truncated flag.
curl -N -X POST "$BASE/run-stream" \
-H "Authorization: Bearer $TOKEN" \
-H "X-App-Slug: $SLUG" \
-H "Content-Type: application/json" \
-H "Idempotency-Key: $KEY" \
-H "Accept: text/event-stream" \
-d "$INPUT"
# event: job {"job_id":"job_..."}
# event: delta {"text":"{\"plan_name\":\"acme-cli"}
# event: delta {"text":" — Poetry to uv migration\","}
# event: done {"status":"succeeded","charged_credits":486,"truncated":false}
# Server-sent events: the plan arrives in chunks, so a UI can show progress.
req = urllib.request.Request(f"{BASE}/run-stream", data=json.dumps(INPUT).encode(), method="POST")
req.add_header("Authorization", f"Bearer {TOKEN}")
req.add_header("X-App-Slug", SLUG)
req.add_header("Content-Type", "application/json")
req.add_header("Idempotency-Key", key)
req.add_header("Accept", "text/event-stream")
raw = ""
done = {}
event = None
stage = "reading the configuration"
with urllib.request.urlopen(req) as stream:
for line in stream:
line = line.decode().rstrip("\n")
if line.startswith("event: "):
event = line[7:]
elif line.startswith("data: ") and event == "delta":
raw += json.loads(line[6:]).get("text", "")
# The arrival of a key name is the progress signal the web app uses.
if '"new_pyproject"' in raw:
stage = "rewriting pyproject.toml"
elif '"steps"' in raw:
stage = "ordering the migration steps"
elif '"checks"' in raw:
stage = "working the twelve checks"
elif line.startswith("data: ") and event == "done":
done = json.loads(line[6:])
plan = json.loads(raw[raw.index("{"):raw.rindex("}") + 1])
print(stage, plan["posture"], len(plan["steps"]), "steps", done.get("charged_credits"))
// Server-sent events: the plan arrives in chunks, so a UI can show progress.
const res = await fetch(`${BASE}/run-stream`, {
method: "POST",
headers: {
Authorization: `Bearer ${TOKEN}`,
"X-App-Slug": SLUG,
"Content-Type": "application/json",
"Idempotency-Key": key,
Accept: "text/event-stream",
},
body: JSON.stringify(INPUT),
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let raw = "";
let done = {};
let event = null;
let stage = "reading the configuration";
while (true) {
const chunk = await reader.read();
if (chunk.done) break;
buffer += decoder.decode(chunk.value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop();
for (const line of lines) {
if (line.startsWith("event: ")) event = line.slice(7);
else if (line.startsWith("data: ") && event === "delta") {
raw += JSON.parse(line.slice(6)).text ?? "";
// The arrival of a key name is the progress signal the web app uses.
if (raw.includes('"new_pyproject"')) stage = "rewriting pyproject.toml";
else if (raw.includes('"steps"')) stage = "ordering the migration steps";
} else if (line.startsWith("data: ") && event === "done") {
done = JSON.parse(line.slice(6));
}
}
}
const plan = JSON.parse(raw.slice(raw.indexOf("{"), raw.lastIndexOf("}") + 1));
console.log(plan.posture, plan.steps.length, "steps", done.charged_credits);
// Server-sent events: the plan arrives in chunks, so a UI can show progress.
req, _ = http.NewRequest(http.MethodPost, base+"/run-stream", bytes.NewReader(body))
req.Header.Set("Authorization", "Bearer "+token)
req.Header.Set("X-App-Slug", slug)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Idempotency-Key", key)
req.Header.Set("Accept", "text/event-stream")
res, _ = http.DefaultClient.Do(req)
defer res.Body.Close()
var raw strings.Builder
var event string
sc := bufio.NewScanner(res.Body)
sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
for sc.Scan() {
line := sc.Text()
switch {
case strings.HasPrefix(line, "event: "):
event = strings.TrimPrefix(line, "event: ")
case strings.HasPrefix(line, "data: ") && event == "delta":
var d struct {
Text string `json:"text"`
}
_ = json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &d)
raw.WriteString(d.Text)
// The arrival of "steps" or "new_pyproject" advances the progress stage.
case strings.HasPrefix(line, "data: ") && event == "done":
fmt.Println(strings.TrimPrefix(line, "data: ")) // status, charged_credits, truncated
}
}
fmt.Println(raw.String())
// Server-sent events: the plan arrives in chunks, so a UI can show progress.
var stream = HttpRequest.newBuilder(URI.create(BASE + "/run-stream"))
.header("Authorization", "Bearer " + TOKEN)
.header("X-App-Slug", SLUG)
.header("Content-Type", "application/json")
.header("Idempotency-Key", key)
.header("Accept", "text/event-stream")
.POST(HttpRequest.BodyPublishers.ofString(input))
.build();
StringBuilder raw = new StringBuilder();
String[] event = { null };
HTTP.send(stream, HttpResponse.BodyHandlers.ofLines()).body().forEach(line -> {
if (line.startsWith("event: ")) event[0] = line.substring(7);
else if (line.startsWith("data: ") && "delta".equals(event[0])) {
raw.append(line.substring(6)); // each data line is {"text":"..."} - decode and append .text
}
});
System.out.println(raw);
// Watch the accumulating text for "steps" and "new_pyproject" to advance a
// progress display. The final `done` event carries status, charged_credits
// and truncated.
# Server-sent events: the plan arrives in chunks, so a UI can show progress.
uri = URI("#{BASE}/run-stream")
req = Net::HTTP::Post.new(uri)
req["Authorization"] = "Bearer #{TOKEN}"
req["X-App-Slug"] = SLUG
req["Content-Type"] = "application/json"
req["Idempotency-Key"] = key
req["Accept"] = "text/event-stream"
req.body = JSON.generate(input)
raw = +""
event = nil
Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) do |http|
http.request(req) do |res|
res.read_body do |chunk|
chunk.each_line do |line|
line = line.chomp
if line.start_with?("event: ") then event = line[7..]
elsif line.start_with?("data: ") && event == "delta"
raw << (JSON.parse(line[6..])["text"] || "")
# The arrival of "steps" or "new_pyproject" advances the progress stage.
end
end
end
end
end
plan = JSON.parse(raw[raw.index("{")..raw.rindex("}")])
puts "#{plan['posture']} #{plan['steps'].length} steps"
<?php
// Server-sent events: the plan arrives in chunks, so a UI can show progress.
$raw = "";
$event = null;
$ch = curl_init(BASE . "/run-stream");
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($input));
curl_setopt($ch, CURLOPT_HTTPHEADER, [
"Authorization: Bearer " . TOKEN,
"X-App-Slug: " . SLUG,
"Content-Type: application/json",
"Idempotency-Key: " . $key,
"Accept: text/event-stream",
]);
curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($ch, $chunk) use (&$raw, &$event) {
foreach (explode("\n", $chunk) as $line) {
if (str_starts_with($line, "event: ")) {
$event = substr($line, 7);
} elseif (str_starts_with($line, "data: ") && $event === "delta") {
$raw .= json_decode(substr($line, 6), true)["text"] ?? "";
}
}
return strlen($chunk);
});
curl_exec($ch);
curl_close($ch);
$plan = json_decode(substr($raw, strpos($raw, "{")), true);
echo $plan["posture"], PHP_EOL;
// Server-sent events: the plan arrives in chunks, so a UI can show progress.
var stream = new HttpRequestMessage(HttpMethod.Post, "https://api.skillsafe.ai/v1/app-api/run-stream");
stream.Headers.Add("Authorization", "Bearer YOUR_TOKEN");
stream.Headers.Add("X-App-Slug", "py-shift");
stream.Headers.Add("Idempotency-Key", key);
stream.Headers.Add("Accept", "text/event-stream");
stream.Content = JsonContent.Create(input);
using var res = await Http.SendAsync(stream, HttpCompletionOption.ResponseHeadersRead);
using var reader = new StreamReader(await res.Content.ReadAsStreamAsync());
var raw = new StringBuilder();
string? evt = null;
while (await reader.ReadLineAsync() is { } line)
{
if (line.StartsWith("event: ")) evt = line[7..];
else if (line.StartsWith("data: ") && evt == "delta")
{
var d = JsonSerializer.Deserialize<JsonElement>(line[6..]);
if (d.TryGetProperty("text", out var t)) raw.Append(t.GetString());
// Watch raw for "steps" and "new_pyproject" to advance a progress display.
}
}
Console.WriteLine(raw.ToString());
7. Parse the plan
data.output.output is a string holding one JSON object. The web app strips an optional
code fence, takes everything from the first { to the last }, parses that,
and then normalizes it. Doing the same two things — the slice and the normalization — is what makes
a caller robust against the small variations a model produces.
Here is an abbreviated plan for the Poetry project above, structurally complete:
{
"plan_name": "acme-cli — Poetry to uv migration",
"posture": "legacy",
"verdict": "Everything else waits on the manager: the floor, the dev tooling and the lockfile are all declared in tables uv does not read.",
"python_floor": "^3.9 (from [tool.poetry.dependencies])",
"toolchain": "poetry + black + flake8 + mypy",
"project_kind": "distributable-package",
"target_python": "3.12",
"exec_summary": "acme-cli is a Poetry project with a caret floor of 3.9 and the black/flake8/mypy shelf declared as a Poetry dev group...",
"assumptions": ["CI is GitHub Actions; the paste shows a Makefile but no workflow."],
"open_questions": ["Does anything outside this repository install acme-cli from the index?"],
"inventory": [
{ "kind": "section", "name": "[tool.poetry]", "value": "name=acme-cli, version=1.4.0",
"role": "package metadata, moves to [project]" },
{ "kind": "backend", "name": "poetry.core.masonry.api", "value": "requires = [\"poetry-core\"]",
"role": "build backend, replaced by uv_build" },
{ "kind": "tool", "name": "black", "value": "^24.3, dev group", "role": "formatter, replaced by ruff format" }
],
"checks": [
{ "check": "dependency manager", "status": "fail",
"evidence": "[tool.poetry] plus a poetry-core build backend",
"requirement": "uv init --bare, then uv add each runtime dependency." },
{ "check": "dependency groups", "status": "partial",
"evidence": "[tool.poetry.group.dev.dependencies] exists, but it is Poetry's dialect, not PEP 735",
"requirement": "Declare dev tooling under [dependency-groups]." }
],
"steps": [
{
"id": "MP-001",
"phase": "dependencies",
"risk": "medium",
"effort": "medium",
"priority": "critical",
"target": "pyproject.toml [tool.poetry.dependencies]",
"problem": "requests is declared under [tool.poetry.dependencies], which uv does not read.",
"impact": "uv sync installs nothing, so every later step is untestable.",
"action": "Initialise uv and re-add the one runtime dependency, then delete the Poetry tables.",
"commands": ["uv init --bare", "uv add requests"],
"snippet": "[project]\nname = \"acme-cli\"\nrequires-python = \">=3.12\"\ndependencies = [\"requests>=2.31\"]"
},
{
"id": "MP-002",
"phase": "lint",
"risk": "low",
"effort": "low",
"priority": "high",
"target": "pyproject.toml [tool.poetry.group.dev.dependencies]",
"problem": "black and flake8 are two tools where ruff is one.",
"impact": "Two config surfaces, two CI steps, and pyupgrade-class fixes nobody is running.",
"action": "Replace black and flake8 with ruff, selecting ALL with a short explicit ignore list.",
"commands": ["uv remove --group dev black flake8", "uv add --group dev ruff"],
"snippet": "[tool.ruff.lint]\nselect = [\"ALL\"]\nignore = [\"D203\", \"D213\", \"COM812\"]"
}
],
"coverage_check": [
{ "id": "MGR-POETRY", "addressed": true, "note": "Confirmed; it is MP-001." },
{ "id": "FLOOR-LOW", "addressed": true, "note": "MP-003 raises requires-python to >=3.12." },
{ "id": "TOOL-BLACK", "addressed": true, "note": "MP-002 replaces it with ruff format." },
{ "id": "VENV-ACTIVATE", "addressed": false, "note": "The Makefile goes away with MP-002; no separate step." }
],
"new_pyproject": "[project]\nname = \"acme-cli\"\nversion = \"1.4.0\"\nrequires-python = \">=3.12\"\ndependencies = [\"requests>=2.31\"]\n\n[build-system]\nrequires = [\"uv_build>=0.9\"]\nbuild-backend = \"uv_build\"\n\n[dependency-groups]\ndev = [\"ruff>=0.14\", \"ty>=0.0.1a\", \"pytest>=8.1\", \"pytest-cov>=5.0\"]\n\n[tool.ruff.lint]\nselect = [\"ALL\"]\nignore = [\"D203\", \"D213\", \"COM812\"]\n\n[tool.ty.environment]\npython-version = \"3.12\"\n\n[tool.pytest.ini_options]\naddopts = \"--cov=acme_cli --cov-fail-under=72\"",
"deletions": ["poetry.lock — superseded by uv.lock once uv sync has run."],
"commands": ["uv sync --all-groups", "uv run ruff check", "uv run ty check src/", "uv run pytest"],
"quick_wins": ["Swap pre-commit for prek: the same .pre-commit-config.yaml, no Python runtime of its own."],
"focus_areas": [
{ "area": "Getting off Poetry", "why": "Nothing else can be verified until uv owns the environment.",
"step_ids": ["MP-001", "MP-002"] }
],
"summary": "Do MP-001 and MP-002 in one afternoon; the security lane can follow next sprint."
}
What the normalizer does to it
The web app does not trust the reply verbatim, and neither should a caller. These are the behaviours you will actually hit:
- An unrecognised
posturebecomespartial. Onlymodern,partialandlegacysurvive. - An unrecognised
phaseon a step becomescleanup; an unrecognisedprioritybecomesmedium; an unrecognisedriskoreffortbecomesmedium. - A
checks[].statusoutside the four allowed values becomesunknown. - A step with neither
problemnoractionis dropped — it carries no information a reader could act on. - If zero steps survive that filter, parsing throws. That is the awkward case:
/runsucceeded, you were charged, and the client still rejects the reply. Handle it as a retry with aretry_note, not as a transport error. - A missing
step idis backfilled positionally asMP-001,MP-002, and so on. focus_areas[].step_idsentries that do not match a real step id are silently dropped, so an area can come back with an emptystep_idsrather than an error. Check for that if you render areas as links.- A
focus_areasentry with noarea, and achecksentry with nocheck, are dropped. - Empty strings in
assumptions,open_questions,deletions,commands,quick_winsand each step'scommandsare filtered out.
Invariants worth asserting in CI
postureis nevermodernwhen any step iscritical, or when three or more checks arefail.postureis neverlegacywithout at least onecriticalorhighstep.- Every id you sent in
prescan_facts.flagsappears exactly once incoverage_check, and no ids you did not send appear there. new_pyprojectcontains none of[tool.poetry],[tool.black],[tool.isort],[tool.flake8],[tool.mypy],[tool.pyright].- Step ids are unique and sequential from
MP-001.
# The plan JSON is a string inside the envelope, so unwrap it twice.
PLAN=$(printf '%s' "$OUT" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["output"]["output"])')
printf '%s' "$PLAN" | python3 -c '
import sys, json
p = json.load(sys.stdin)
print(p["posture"], "|", p["verdict"])
print(p["toolchain"], "->", p["target_python"])
for c in p["checks"]:
print(" %-8s %s" % (c["status"], c["check"]))
for s in p["steps"]:
print(" %s %-9s %-12s %s" % (s["id"], s["priority"], s["phase"], s["target"]))
'
# Every prescan flag id must come back exactly once in coverage_check.
printf '%s' "$PLAN" | python3 -c '
import sys, json
seen = [c["id"] for c in json.load(sys.stdin)["coverage_check"]]
want = ["MGR-POETRY", "FLOOR-LOW", "TOOL-BLACK", "VENV-ACTIVATE"]
missing = [i for i in want if seen.count(i) != 1]
if missing:
raise SystemExit("unreconciled prescan flags: " + ", ".join(missing))
print("coverage_check reconciles")
'
# The rewritten pyproject.toml must not carry a legacy table forward.
printf '%s' "$PLAN" | python3 -c '
import sys, json
tables = ["[tool.poetry]", "[tool.black]", "[tool.isort]", "[tool.flake8]", "[tool.mypy]", "[tool.pyright]"]
body = json.load(sys.stdin)["new_pyproject"]
left = [t for t in tables if t in body]
if left:
raise SystemExit("new_pyproject still contains: " + ", ".join(left))
print("new_pyproject is clean")
'
plan = json.loads(job["output"]["output"])
# 1. Every prescan flag id appears exactly once in coverage_check, and nothing else does.
sent = [f["id"] for f in INPUT["prescan_facts"]["flags"]]
seen = [c["id"] for c in plan["coverage_check"]]
missing = [i for i in sent if seen.count(i) != 1]
extra = [i for i in seen if i not in sent]
if missing or extra:
raise RuntimeError(f"coverage_check drift: missing={missing} extra={extra}")
# 2. The posture cannot contradict the steps or the checks.
critical = [s for s in plan["steps"] if s["priority"] == "critical"]
fails = [c for c in plan["checks"] if c["status"] == "fail"]
if plan["posture"] == "modern" and (critical or len(fails) >= 3):
raise RuntimeError("posture says modern but the steps and checks say otherwise")
if plan["posture"] == "legacy" and not any(s["priority"] in ("critical", "high") for s in plan["steps"]):
raise RuntimeError("posture says legacy but no step is critical or high")
# 3. The rewritten pyproject must not carry a legacy table forward.
LEGACY = ["[tool.poetry]", "[tool.black]", "[tool.isort]", "[tool.flake8]", "[tool.mypy]", "[tool.pyright]"]
left = [t for t in LEGACY if t in plan["new_pyproject"]]
if left:
raise RuntimeError("new_pyproject still contains: " + ", ".join(left))
# 4. focus_areas step_ids are silently dropped when they do not match a step,
# so an empty list here means the model referenced ids that do not exist.
for area in plan["focus_areas"]:
if not area["step_ids"]:
print("note: focus area", area["area"], "resolved to no steps")
# 5. A truncated reply is a prefix, not a plan. Retry, do not repair.
if job.get("truncated"):
INPUT["retry_note"] = (
"The previous reply was truncated. Return the same twelve checks but at most "
"eight steps, each with a shorter snippet."
)
# ... resubmit with an incremented attempt suffix in the Idempotency-Key.
for c in plan["checks"]:
print(f"{c['status']:8} {c['check']:24} {c['evidence']}")
for s in plan["steps"]:
print(s["id"], s["priority"], s["phase"], s["target"])
print(plan["new_pyproject"])
print("\n".join(plan["commands"]))
const plan = JSON.parse(job.output.output);
// 1. Every prescan flag id appears exactly once in coverage_check.
const sent = INPUT.prescan_facts.flags.map((f) => f.id);
const seen = plan.coverage_check.map((c) => c.id);
const missing = sent.filter((id) => seen.filter((s) => s === id).length !== 1);
const extra = seen.filter((id) => !sent.includes(id));
if (missing.length || extra.length) {
throw new Error(`coverage_check drift: missing=${missing} extra=${extra}`);
}
// 2. The posture cannot contradict the steps or the checks.
const critical = plan.steps.filter((s) => s.priority === "critical");
const fails = plan.checks.filter((c) => c.status === "fail");
if (plan.posture === "modern" && (critical.length || fails.length >= 3)) {
throw new Error("posture says modern but the steps and checks say otherwise");
}
if (plan.posture === "legacy" &&
!plan.steps.some((s) => s.priority === "critical" || s.priority === "high")) {
throw new Error("posture says legacy but no step is critical or high");
}
// 3. The rewritten pyproject must not carry a legacy table forward.
const LEGACY = ["[tool.poetry]", "[tool.black]", "[tool.isort]", "[tool.flake8]",
"[tool.mypy]", "[tool.pyright]"];
const left = LEGACY.filter((t) => plan.new_pyproject.includes(t));
if (left.length) throw new Error(`new_pyproject still contains: ${left.join(", ")}`);
// 4. A truncated reply is a prefix, not a plan. Retry, do not repair.
if (job.truncated) {
INPUT.retry_note =
"The previous reply was truncated. Return the same twelve checks but at most eight steps.";
}
for (const c of plan.checks) console.log(c.status.padEnd(8), c.check, "—", c.evidence);
for (const s of plan.steps) console.log(s.id, s.priority, s.phase, s.target);
console.log(plan.new_pyproject);
console.log(plan.commands.join("\n"));
type check struct {
Check string `json:"check"`
Status string `json:"status"`
Evidence string `json:"evidence"`
Requirement string `json:"requirement"`
}
type step struct {
ID string `json:"id"`
Phase string `json:"phase"`
Risk string `json:"risk"`
Effort string `json:"effort"`
Priority string `json:"priority"`
Target string `json:"target"`
Problem string `json:"problem"`
Impact string `json:"impact"`
Action string `json:"action"`
Commands []string `json:"commands"`
Snippet string `json:"snippet"`
}
type plan struct {
PlanName string `json:"plan_name"`
Posture string `json:"posture"`
Verdict string `json:"verdict"`
PythonFloor string `json:"python_floor"`
Toolchain string `json:"toolchain"`
ProjectKind string `json:"project_kind"`
TargetPython string `json:"target_python"`
Checks []check `json:"checks"`
Steps []step `json:"steps"`
NewPyproject string `json:"new_pyproject"`
Deletions []string `json:"deletions"`
Commands []string `json:"commands"`
CoverageCheck []struct {
ID string `json:"id"`
Addressed bool `json:"addressed"`
Note string `json:"note"`
} `json:"coverage_check"`
}
var p plan
if err := json.Unmarshal([]byte(job.Output.Output), &p); err != nil {
panic(err)
}
// Every prescan flag id must come back exactly once in coverage_check.
count := map[string]int{}
for _, c := range p.CoverageCheck {
count[c.ID]++
}
for _, id := range []string{"MGR-POETRY", "FLOOR-LOW", "TOOL-BLACK", "VENV-ACTIVATE"} {
if count[id] != 1 {
panic("unreconciled prescan flag: " + id)
}
}
// The rewritten pyproject must not carry a legacy table forward.
for _, t := range []string{"[tool.poetry]", "[tool.black]", "[tool.isort]",
"[tool.flake8]", "[tool.mypy]", "[tool.pyright]"} {
if strings.Contains(p.NewPyproject, t) {
panic("new_pyproject still contains " + t)
}
}
fmt.Println(p.Posture, p.Toolchain, "->", p.TargetPython, len(p.Steps), "steps")
// The plan JSON is a string inside data.output.output - parse it, then check the
// invariants before you trust it:
//
// 1. every prescan_facts.flags id appears exactly once in coverage_check,
// and no id you did not send appears there;
// 2. posture is not "modern" when a step is critical or three checks fail,
// and not "legacy" without a critical or high step;
// 3. new_pyproject contains none of [tool.poetry], [tool.black], [tool.isort],
// [tool.flake8], [tool.mypy], [tool.pyright];
// 4. focus_areas[].step_ids entries that name no real step are dropped by the
// normalizer, so an empty list means the model invented ids.
//
// A `truncated` job is a prefix, not a plan: resubmit with a retry_note such as
// "The previous reply was truncated. Return the same twelve checks but at most
// eight steps" and an incremented attempt suffix on the Idempotency-Key.
String planJson = /* data.output.output */ call("jobs/" + jobId, null);
System.out.println(planJson);
// checks[] is always the same twelve entries in the same order, so a table can
// be rendered by index without searching for a check by name.
plan = JSON.parse(job["output"]["output"])
# 1. Every prescan flag id appears exactly once in coverage_check.
sent = input["prescan_facts"]["flags"].map { |f| f["id"] }
seen = plan["coverage_check"].map { |c| c["id"] }
missing = sent.reject { |id| seen.count(id) == 1 }
extra = seen - sent
raise "coverage_check drift: #{missing} / #{extra}" unless missing.empty? && extra.empty?
# 2. The posture cannot contradict the steps or the checks.
critical = plan["steps"].count { |s| s["priority"] == "critical" }
fails = plan["checks"].count { |c| c["status"] == "fail" }
raise "modern contradicted" if plan["posture"] == "modern" && (critical > 0 || fails >= 3)
# 3. The rewritten pyproject must not carry a legacy table forward.
legacy = ["[tool.poetry]", "[tool.black]", "[tool.isort]", "[tool.flake8]",
"[tool.mypy]", "[tool.pyright]"].select { |t| plan["new_pyproject"].include?(t) }
raise "new_pyproject still contains #{legacy.join(', ')}" unless legacy.empty?
plan["checks"].each { |c| puts format("%-8s %s", c["status"], c["check"]) }
plan["steps"].each { |s| puts "#{s['id']} #{s['priority']} #{s['phase']} #{s['target']}" }
puts plan["new_pyproject"]
puts plan["commands"].join("\n")
<?php
$plan = json_decode($job["output"]["output"], true);
// 1. Every prescan flag id appears exactly once in coverage_check.
$sent = array_column($input["prescan_facts"]["flags"], "id");
$seen = array_column($plan["coverage_check"], "id");
$counts = array_count_values($seen);
foreach ($sent as $id) {
if (($counts[$id] ?? 0) !== 1) {
throw new RuntimeException("unreconciled prescan flag: " . $id);
}
}
// 2. The posture cannot contradict the steps or the checks.
$critical = count(array_filter($plan["steps"], fn($s) => $s["priority"] === "critical"));
$fails = count(array_filter($plan["checks"], fn($c) => $c["status"] === "fail"));
if ($plan["posture"] === "modern" && ($critical > 0 || $fails >= 3)) {
throw new RuntimeException("posture says modern but the steps and checks say otherwise");
}
// 3. The rewritten pyproject must not carry a legacy table forward.
foreach (["[tool.poetry]", "[tool.black]", "[tool.isort]", "[tool.flake8]",
"[tool.mypy]", "[tool.pyright]"] as $t) {
if (str_contains($plan["new_pyproject"], $t)) {
throw new RuntimeException("new_pyproject still contains " . $t);
}
}
foreach ($plan["checks"] as $c) {
printf("%-8s %s\n", $c["status"], $c["check"]);
}
echo $plan["new_pyproject"], PHP_EOL;
var plan = JsonSerializer.Deserialize<JsonElement>(planJson);
// 1. Every prescan flag id appears exactly once in coverage_check.
var seen = plan.GetProperty("coverage_check")
.EnumerateArray()
.Select(c => c.GetProperty("id").GetString())
.ToList();
foreach (var id in new[] { "MGR-POETRY", "FLOOR-LOW", "TOOL-BLACK", "VENV-ACTIVATE" })
{
if (seen.Count(s => s == id) != 1)
throw new Exception($"unreconciled prescan flag: {id}");
}
// 2. The posture cannot contradict the steps or the checks.
var critical = plan.GetProperty("steps").EnumerateArray()
.Count(s => s.GetProperty("priority").GetString() == "critical");
var fails = plan.GetProperty("checks").EnumerateArray()
.Count(c => c.GetProperty("status").GetString() == "fail");
if (plan.GetProperty("posture").GetString() == "modern" && (critical > 0 || fails >= 3))
throw new Exception("posture says modern but the steps and checks say otherwise");
// 3. The rewritten pyproject must not carry a legacy table forward.
var body = plan.GetProperty("new_pyproject").GetString() ?? "";
foreach (var t in new[] { "[tool.poetry]", "[tool.black]", "[tool.isort]",
"[tool.flake8]", "[tool.mypy]", "[tool.pyright]" })
if (body.Contains(t)) throw new Exception($"new_pyproject still contains {t}");
foreach (var c in plan.GetProperty("checks").EnumerateArray())
Console.WriteLine($"{c.GetProperty("status")} {c.GetProperty("check")}");
The output contract
Every key in the object, as the web app reads it:
| key | type | meaning |
|---|---|---|
plan_name | string | Short title naming the project and the migration, e.g. "acme-cli — Poetry to uv migration". Empty becomes Untitled migration plan. |
posture | enum | modern, partial or legacy. Anything else normalizes to partial. The single value a CI gate should branch on. |
verdict | string | One sentence naming the single thing that decides the posture. |
python_floor | string | The floor the paste declares, quoted with where it came from — "^3.9 (from [tool.poetry.dependencies])". Empty becomes unknown. |
toolchain | string | The tooling the paste actually declares, as a short chain: "poetry + black + isort + flake8 + mypy". Empty becomes unknown. |
project_kind | string | What the project turned out to be, in the plan's own words — distributable-package, deployed-application, internal-library, standalone-script. Empty becomes unknown. |
target_python | string | The floor the plan is written against, echoing floor unless it was keep. Empty becomes unknown. |
exec_summary | string | Two to five sentences a tech lead can act on without reading the rest. |
assumptions | string[] | What had to be assumed because the paste is silent — the CI provider, whether a tests/ directory exists off-screen, whether the package is published. An explicit refusal in context is recorded here too. |
open_questions | string[] | Questions whose answers would change the plan or its ordering. |
inventory | object[] | {kind, name, value, role} — the configuration sections, the build backend, the declared runtime and dev dependencies, the tools each config file configures, the layout and the floor. An entry with neither kind nor name is dropped. |
checks | object[] | {check, status, evidence, requirement}. Always the same twelve entries in the same fixed order — render by index, do not search by name. An entry with no check string is dropped. |
steps | object[] | {id, phase, risk, effort, priority, target, problem, impact, action, commands, snippet}. Ids are sequential MP-001, MP-002, … in execution order. commands is the exact shell for that step; snippet is the corrected configuration fragment, or "". |
coverage_check | object[] | {id, addressed, note}. One entry per prescan_facts.flags id, exactly once, and no ids the prescan did not send. addressed: false means deliberately set aside, with the reason in note. |
new_pyproject | string | The whole rewritten pyproject.toml, as pasteable TOML in one JSON string: [project], [build-system] on uv_build where appropriate, [dependency-groups], [tool.ruff], [tool.pytest.ini_options], [tool.ty.environment]. Never a legacy table. "" when the paste gives nothing to base it on. |
deletions | string[] | Files to delete once the migration is verified, one per line, each with its reason. |
commands | string[] | Top-level verification commands, in the order they should be run against the repository after the steps are done — uv sync --all-groups, uv run ruff check, uv run ty check src/, uv run pytest, uv build. |
quick_wins | string[] | Changes worth doing today regardless of the rest, one line each. |
focus_areas | object[] | {area, why, step_ids}. Ids that match no step are dropped silently; an entry with no area is dropped entirely. |
summary | string | One paragraph closing the plan. |
The enums
| field | values | notes |
|---|---|---|
posture | modern, partial, legacy | modern: already on uv, ruff and ty, and the plan spends itself on what is left — a wider select, zizmor, a committed lockfile, a coverage floor. partial: some of the stack has landed and the rest has not. legacy: the manager or the lint-and-type shelf is still the old one. Anything the model returns outside these three normalizes to partial. |
steps[].phase | dependencies, layout, build, lint, types, testing, security, ci, scripts, cleanup | Roughly the execution order: the manager and dependencies first, because everything else is installed by them; cleanup and deletions last, because deleting setup.py before uv init --bare has written a pyproject.toml loses metadata. An unrecognised phase normalizes to cleanup. |
steps[].risksteps[].effort | low, medium, high | Risk is what the change could break; effort is how long it takes. Both default to medium when unrecognised. |
steps[].priority | critical, high, medium, low | critical is reserved for something that blocks the rest of the migration or is actively wrong today — dependencies declared where uv will not read them, a [tool.ty] python-version at the top level so type checking silently runs against the wrong Python, a secret committed in a workflow. Unrecognised values normalize to medium. |
checks[].status | pass, fail, partial, unknown | unknown is a legitimate answer and is preferred over a guess: a paste with no CI workflow cannot prove dependency automation. partial means the practice is present but incomplete — ruff configured with a narrow select, a dev group that is Poetry's dialect rather than PEP 735. Unrecognised values normalize to unknown. |
The twelve checks
checks always carries these twelve check strings, in this order, on every
run — so a table can be rendered by index and two plans for the same repository are diffable row by
row:
1. dependency manager 7. lint and format
2. dependency groups 8. type checking
3. python floor 9. testing and coverage
4. project layout 10. pre-commit hooks
5. build backend 11. security scanning
6. lockfile 12. dependency automation
The plan never echoes a secret value. If the paste contains a PyPI upload token, an API key in a
workflow or a password in a URL, the step names the setting and says to rotate it — the value
itself does not appear in problem, snippet or new_pyproject.
8. Use it in CI
The worked example: a job reads pyproject.toml out of the repository, runs a plan, and
exits non-zero when posture is anything but modern or when any step comes
back critical. Derive the Idempotency-Key from the file contents so a re-run of the
same commit replays the same job instead of re-billing, and only bump the attempt suffix when the
configuration actually changed. Point the gate at a scheduled job as well as at pull requests: the
posture can regress without pyproject.toml changing at all, because a new advisory or
a new ruff release moves what pass means.
#!/bin/sh
# py-shift-gate.sh - fail the build when the packaging posture regresses.
set -eu
BASE="https://api.skillsafe.ai/v1/app-api"
SLUG="py-shift"
TOKEN="$SKILLSAFE_TOKEN" # from https://py-shift.skillsafe.ai/tokens.html
# 1. Collect whatever the repository has, each behind a marker line.
FILES=$(
for f in pyproject.toml requirements-dev.txt .pre-commit-config.yaml Makefile; do
[ -f "$f" ] || continue
printf '# file: %s\n' "$f"
cat "$f"
printf '\n'
done
)
# 2. Build the input. An API caller may send empty prescan facts; the plan still works.
INPUT=$(FILES="$FILES" python3 -c '
import json, os
print(json.dumps({
"files": os.environ["FILES"],
"target": "package",
"floor": "3.12",
"focus": "general",
"context": "CI gate on every pull request.",
"prescan_facts": {"resources": [], "flags": []},
}))')
KEY="py-shift:$(printf '%s' "$INPUT" | shasum -a 256 | cut -c1-16):a1"
JOB=$(curl -sS -X POST "$BASE/run" \
-H "Authorization: Bearer $TOKEN" -H "X-App-Slug: $SLUG" \
-H "Content-Type: application/json" -H "Idempotency-Key: $KEY" \
-d "$INPUT" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["job_id"])')
while :; do
OUT=$(curl -sS "$BASE/jobs/$JOB" -H "Authorization: Bearer $TOKEN" -H "X-App-Slug: $SLUG")
STATUS=$(printf '%s' "$OUT" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["status"])')
[ "$STATUS" = "succeeded" ] && break
[ "$STATUS" = "failed" ] && echo "$OUT" && exit 1
sleep 2
done
# 3. Gate on the posture and on any critical step.
printf '%s' "$OUT" | python3 -c '
import sys, json
job = json.load(sys.stdin)["data"]
if job.get("truncated"):
raise SystemExit("::error::py-shift: reply was truncated, plan is incomplete")
plan = json.loads(job["output"]["output"])
crit = [s["id"] for s in plan["steps"] if s["priority"] == "critical"]
print(plan["posture"], "-", plan["verdict"])
if plan["posture"] != "modern" or crit:
raise SystemExit("::error::py-shift: posture=%s, critical=%s" % (plan["posture"], ",".join(crit) or "none"))
print("py-shift: modern, nothing critical")
'
#!/usr/bin/env python3
"""py_shift_gate.py - fail CI when the packaging posture regresses.
Reuses the `call` helper from section 2. Exits 1 on a non-modern posture or on
any critical step, and 1 on a truncated reply, which is a prefix and not a plan.
"""
import hashlib, json, pathlib, sys, time, urllib.request
CANDIDATES = [
"pyproject.toml", "setup.py", "setup.cfg", "requirements.txt", "requirements-dev.txt",
".pre-commit-config.yaml", "mypy.ini", ".flake8", "tox.ini", "Makefile",
".github/workflows/ci.yml",
]
parts = []
for name in CANDIDATES:
p = pathlib.Path(name)
if p.is_file():
parts.append(f"# file: {name}\n{p.read_text(encoding='utf-8')}")
if not parts:
sys.exit("py-shift: no packaging configuration found in this repository")
INPUT = {
"files": "\n\n".join(parts),
"target": "package",
"floor": "3.12",
"focus": "general",
"context": "CI gate. Fails the build on a non-modern posture or any critical step.",
# An API caller may send empty prescan facts; the plan still works. Anything
# you DO send here must come back exactly once in coverage_check.
"prescan_facts": {"resources": [], "flags": []},
}
digest = hashlib.sha256(json.dumps(INPUT, sort_keys=True).encode()).hexdigest()[:16]
key = f"py-shift:{digest}:a1"
req = urllib.request.Request(f"{BASE}/run", data=json.dumps(INPUT).encode(), method="POST")
req.add_header("Authorization", f"Bearer {TOKEN}")
req.add_header("X-App-Slug", SLUG)
req.add_header("Content-Type", "application/json")
req.add_header("Idempotency-Key", key)
with urllib.request.urlopen(req) as r:
job_id = json.load(r)["data"]["job_id"]
while True:
job = call(f"jobs/{job_id}")
if job["status"] in ("succeeded", "failed"):
break
time.sleep(2)
if job["status"] == "failed":
sys.exit(f"py-shift: run failed: {job.get('error')}")
if job.get("truncated"):
sys.exit("py-shift: reply was truncated - the plan is a prefix, not a plan")
plan = json.loads(job["output"]["output"])
critical = [s["id"] for s in plan["steps"] if s["priority"] == "critical"]
fails = [c["check"] for c in plan["checks"] if c["status"] == "fail"]
print(f"posture={plan['posture']} floor={plan['python_floor']} -> {plan['target_python']}")
print(plan["verdict"])
for c in plan["checks"]:
print(f" {c['status']:8} {c['check']}")
for s in plan["steps"]:
print(f" {s['id']} {s['priority']:8} {s['phase']:12} {s['target']}")
if plan["posture"] != "modern" or critical:
sys.exit(
f"py-shift: posture={plan['posture']}, "
f"critical={','.join(critical) or 'none'}, failing checks={len(fails)}"
)
print("py-shift: modern, nothing critical. Charged", job.get("charged_credits"), "credits.")
// py-shift-gate.mjs - fail CI when the packaging posture regresses.
// Reuses the `call` helper from section 2.
import { createHash } from "node:crypto";
import { existsSync, readFileSync } from "node:fs";
const CANDIDATES = [
"pyproject.toml", "setup.py", "requirements.txt", "requirements-dev.txt",
".pre-commit-config.yaml", "Makefile", ".github/workflows/ci.yml",
];
const parts = CANDIDATES.filter((f) => existsSync(f)).map(
(f) => `# file: ${f}\n${readFileSync(f, "utf8")}`
);
if (!parts.length) throw new Error("py-shift: no packaging configuration found");
const INPUT = {
files: parts.join("\n\n"),
target: "package",
floor: "3.12",
focus: "general",
context: "CI gate. Fails the build on a non-modern posture or any critical step.",
prescan_facts: { resources: [], flags: [] }, // empty is legitimate
};
const digest = createHash("sha256").update(JSON.stringify(INPUT)).digest("hex").slice(0, 16);
const started = await fetch(`${BASE}/run`, {
method: "POST",
headers: {
Authorization: `Bearer ${TOKEN}`,
"X-App-Slug": SLUG,
"Content-Type": "application/json",
"Idempotency-Key": `py-shift:${digest}:a1`,
},
body: JSON.stringify(INPUT),
}).then((r) => r.json());
let job = started.data;
while (job.status !== "succeeded" && job.status !== "failed") {
await new Promise((r) => setTimeout(r, 2000));
job = await call(`jobs/${job.job_id}`);
}
if (job.status === "failed") throw new Error("py-shift: run failed");
if (job.truncated) throw new Error("py-shift: reply was truncated");
const plan = JSON.parse(job.output.output);
const critical = plan.steps.filter((s) => s.priority === "critical").map((s) => s.id);
console.log(plan.posture, "-", plan.verdict);
if (plan.posture !== "modern" || critical.length) {
console.error(`py-shift: posture=${plan.posture}, critical=${critical.join(",") || "none"}`);
process.exitCode = 1;
}
// The gate, on top of the client and the plan struct from the earlier sections:
// read the repository's configuration, run a plan, exit non-zero on a non-modern
// posture or any critical step.
var parts []string
for _, name := range []string{"pyproject.toml", "requirements-dev.txt",
".pre-commit-config.yaml", "Makefile"} {
b, err := os.ReadFile(name)
if err != nil {
continue
}
parts = append(parts, "# file: "+name+"\n"+string(b))
}
if len(parts) == 0 {
panic("py-shift: no packaging configuration found")
}
input := map[string]any{
"files": strings.Join(parts, "\n\n"),
"target": "package",
"floor": "3.12",
"focus": "general",
"context": "CI gate.",
// Empty prescan facts are legitimate; the plan still works.
"prescan_facts": map[string]any{"resources": []any{}, "flags": []any{}},
}
// ... POST /run with the Idempotency-Key, poll jobs/{job_id}, unmarshal into `plan`.
var crit []string
for _, s := range p.Steps {
if s.Priority == "critical" {
crit = append(crit, s.ID)
}
}
if p.Posture != "modern" || len(crit) > 0 {
fmt.Fprintf(os.Stderr, "py-shift: posture=%s critical=%s\n", p.Posture, strings.Join(crit, ","))
os.Exit(1)
}
fmt.Println("py-shift: modern, nothing critical")
// The gate, on top of the PyShift client from section 2. Read the repository's
// configuration into `files`, each file behind a "# file: <name>" marker, POST
// /run with the Idempotency-Key, poll jobs/{job_id}, then:
//
// var plan = /* parse data.output.output */;
// boolean bad = !"modern".equals(plan.posture)
// || plan.steps.stream().anyMatch(s -> "critical".equals(s.priority));
// if (bad) System.exit(1);
//
// Send "prescan_facts": {"resources": [], "flags": []} when you have no local
// analyzer - it is legitimate, and it keeps coverage_check trivially satisfied.
// A truncated job is also a failure: the plan you hold is a prefix.
String files = java.nio.file.Files.readString(java.nio.file.Path.of("pyproject.toml"));
String input = "{\"files\":\"# file: pyproject.toml\\n\" + ... , \"target\":\"package\","
+ "\"floor\":\"3.12\",\"focus\":\"general\",\"context\":\"CI gate.\","
+ "\"prescan_facts\":{\"resources\":[],\"flags\":[]}}";
System.out.println(files.length() + " bytes of configuration to plan from");
# py_shift_gate.rb - fail CI when the packaging posture regresses.
# Reuses the `call` helper from section 2.
require "digest"
candidates = ["pyproject.toml", "setup.py", "requirements-dev.txt",
".pre-commit-config.yaml", "Makefile"]
parts = candidates.select { |f| File.file?(f) }
.map { |f| "# file: #{f}\n#{File.read(f)}" }
abort "py-shift: no packaging configuration found" if parts.empty?
input = {
"files" => parts.join("\n\n"),
"target" => "package",
"floor" => "3.12",
"focus" => "general",
"context" => "CI gate.",
"prescan_facts" => { "resources" => [], "flags" => [] }
}
key = "py-shift:#{Digest::SHA256.hexdigest(JSON.generate(input))[0, 16]}:a1"
# ... POST /run with that Idempotency-Key, then poll jobs/{job_id} as in section 5.
plan = JSON.parse(job["output"]["output"])
critical = plan["steps"].select { |s| s["priority"] == "critical" }.map { |s| s["id"] }
puts "#{plan['posture']} - #{plan['verdict']}"
abort "py-shift: posture=#{plan['posture']} critical=#{critical.join(',')}" if
plan["posture"] != "modern" || !critical.empty?
puts "py-shift: modern, nothing critical"
<?php
// py-shift-gate.php - fail CI when the packaging posture regresses.
// Reuses the `call` helper from section 2.
$candidates = ["pyproject.toml", "setup.py", "requirements-dev.txt",
".pre-commit-config.yaml", "Makefile"];
$parts = [];
foreach ($candidates as $f) {
if (is_file($f)) { $parts[] = "# file: {$f}\n" . file_get_contents($f); }
}
if (!$parts) { fwrite(STDERR, "py-shift: no packaging configuration found\n"); exit(1); }
$input = [
"files" => implode("\n\n", $parts),
"target" => "package",
"floor" => "3.12",
"focus" => "general",
"context" => "CI gate.",
"prescan_facts" => ["resources" => [], "flags" => []],
];
$key = "py-shift:" . substr(hash("sha256", json_encode($input)), 0, 16) . ":a1";
// ... POST /run with that Idempotency-Key, then poll jobs/{job_id} as in section 5.
$plan = json_decode($job["output"]["output"], true);
$critical = array_column(array_filter($plan["steps"], fn($s) => $s["priority"] === "critical"), "id");
echo $plan["posture"], " - ", $plan["verdict"], PHP_EOL;
if ($plan["posture"] !== "modern" || $critical) {
fwrite(STDERR, "py-shift: posture=" . $plan["posture"] .
" critical=" . (implode(",", $critical) ?: "none") . "\n");
exit(1);
}
echo "py-shift: modern, nothing critical", PHP_EOL;
// The gate, on top of the PyShift client from section 2.
var candidates = new[] { "pyproject.toml", "setup.py", "requirements-dev.txt",
".pre-commit-config.yaml", "Makefile" };
var parts = candidates.Where(File.Exists)
.Select(f => $"# file: {f}\n{File.ReadAllText(f)}")
.ToList();
if (parts.Count == 0) throw new Exception("py-shift: no packaging configuration found");
var input = new
{
files = string.Join("\n\n", parts),
target = "package",
floor = "3.12",
focus = "general",
context = "CI gate.",
prescan_facts = new { resources = Array.Empty<object>(), flags = Array.Empty<object>() },
};
// ... POST /run with the Idempotency-Key, poll jobs/{job_id}, parse data.output.output.
var posture = plan.GetProperty("posture").GetString();
var critical = plan.GetProperty("steps").EnumerateArray()
.Where(s => s.GetProperty("priority").GetString() == "critical")
.Select(s => s.GetProperty("id").GetString())
.ToList();
if (posture != "modern" || critical.Count > 0)
{
Console.Error.WriteLine($"py-shift: posture={posture}, critical={string.Join(",", critical)}");
Environment.Exit(1);
}
Console.WriteLine("py-shift: modern, nothing critical");
Truncation and partial results
When the balance sits between min_credits and hold_credits, the run is not
refused: it executes with a reduced output cap and comes back with truncated: true on
the finished job and on the streaming done event. What you hold then is a prefix of the
plan, not the plan — the twelve checks may be complete while new_pyproject,
deletions and summary are missing or cut mid-string.
Check the flag before you treat a plan as complete. The right response is a retry, not a repair:
resubmit with a retry_note asking for fewer, denser steps and a shorter
new_pyproject, and with the attempt suffix on the Idempotency-Key
incremented so the new body is not a replay of the old key. Repairing truncated JSON by appending
closing braces produces something that parses and is not what the model meant.