MCP takes pictures of real size, by id too, and waits for a cold model
The game's bridge found three things on the PoC. /mcp held its body to 64 KiB — the edit meant for it had raised the memory notes' limit instead — so a 1024 px concept sent as base64 failed as "expected one JSON-RPC request object". The notes are back to 64 KiB, /mcp takes 64 MB, and a call past it is a 413 saying so. /v1/files took batch files only, so a picture could not be sent once and named after. purpose vision and user_data are kept too, as long as job results, and generate_mesh hands a workflow model the id as it is. generate_image and review_images failed at once while their model was cold. They wait out an answer that says when to come back — loading, another model holding the GPU, a full queue — for as long as the call may, saying who holds the GPU and about how long; a refusal that says no Retry-After, a spent budget, is said at once. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
c22528c26e
commit
edeeb9f8e1
6 files changed
+195
-8
No files matched your search
@@ -28,6 +28,10 @@ APIs change between minors without a deprecation period.
|
||||
|
||||
### Added
|
||||
|
||||
- **Pictures uploaded to work from**: `/v1/files` takes `purpose=vision` or
|
||||
`user_data` besides `batch`, kept as long as job results, so a picture is
|
||||
sent once and named by its `file-…` id in the MCP tools and a workflow's
|
||||
inputs.
|
||||
- **One ComfyUI for an environment's models**: ComfyUI models in one
|
||||
`comfyui.environment` (two or more) share one replica — the catalog
|
||||
makes it a hidden entry, `comfyui-env-<name>`, on the first member's
|
||||
|
||||
@@ -556,6 +556,15 @@ over HTTP cannot write to your disk, so an agent that needs the file
|
||||
locally — to import it in Blender — fetches it, or goes through a small
|
||||
local bridge that does.
|
||||
|
||||
A picture can be uploaded once and named by its id after: `POST /v1/files`
|
||||
with `purpose=vision` (or `user_data`) keeps it, for as long as job results
|
||||
are kept, as a file of yours that `generate_mesh`, `review_images` and a
|
||||
workflow's file inputs take as `file-…`. A call is at most 64 MB.
|
||||
|
||||
A model that is not loaded is waited for: a picture or a look at one whose
|
||||
model is starting, or waiting for another to give up the GPU, waits as a
|
||||
mesh does, saying so, rather than failing with "model loading".
|
||||
|
||||
A mesh takes from half a minute to a few minutes. A client that takes the
|
||||
answer as a stream (`Accept: text/event-stream`, as Claude Code does) is
|
||||
answered that way: `notifications/progress` for the call's
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/auth"
|
||||
@@ -20,8 +21,16 @@ import (
|
||||
const (
|
||||
purposeBatch = "batch"
|
||||
purposeBatchOutput = "batch_output"
|
||||
// purposeVision and purposeUserData are files a request works from — a
|
||||
// picture to make a mesh of or to look at, a workflow's input — named
|
||||
// by id rather than sent as base64 in every call. OpenAI's names.
|
||||
purposeVision = "vision"
|
||||
purposeUserData = "user_data"
|
||||
)
|
||||
|
||||
// uploadPurposes are what a client may upload a file for.
|
||||
var uploadPurposes = []string{purposeBatch, purposeVision, purposeUserData}
|
||||
|
||||
func newID(prefix string) string {
|
||||
var b [12]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
@@ -99,9 +108,11 @@ func (r *Router) handleUploadFile(w http.ResponseWriter, req *http.Request) {
|
||||
fmt.Sprintf("a multipart form with purpose and file, at most %d MB: %v", maxBytes>>20, err))
|
||||
return
|
||||
}
|
||||
if purpose := req.FormValue("purpose"); purpose != purposeBatch {
|
||||
purpose := req.FormValue("purpose")
|
||||
if !slices.Contains(uploadPurposes, purpose) {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request_error",
|
||||
fmt.Sprintf("purpose %q: this gateway keeps files for batches only (purpose batch)", purpose))
|
||||
fmt.Sprintf("purpose %q: this gateway keeps files for batches (batch) and for requests to work from "+
|
||||
"(vision, user_data)", purpose))
|
||||
return
|
||||
}
|
||||
file, header, err := req.FormFile("file")
|
||||
@@ -120,8 +131,13 @@ func (r *Router) handleUploadFile(w http.ResponseWriter, req *http.Request) {
|
||||
fmt.Sprintf("the file is larger than %d MB", maxBytes>>20))
|
||||
return
|
||||
}
|
||||
f := store.File{ID: newID("file-"), Owner: ownerOf(p), Purpose: purposeBatch,
|
||||
f := store.File{ID: newID("file-"), Owner: ownerOf(p), Purpose: purpose,
|
||||
Filename: header.Filename, Bytes: int64(len(content)), CreatedAt: time.Now()}
|
||||
// A file to work from is kept as long as what is made from it.
|
||||
if purpose != purposeBatch {
|
||||
_, _, _, days := r.batchCfg()
|
||||
f.ExpiresAt = f.CreatedAt.AddDate(0, 0, days)
|
||||
}
|
||||
if err := r.store.PutFile(req.Context(), f, content); err != nil {
|
||||
writeError(w, http.StatusInternalServerError, "server_error", err.Error())
|
||||
return
|
||||
|
||||
@@ -348,6 +348,45 @@ func (r *Router) selfJSON(c mediaCall, method, path string, body, into any) (htt
|
||||
return h, json.Unmarshal(b, into)
|
||||
}
|
||||
|
||||
// selfJSONWaiting is selfJSON for a model that may be cold: an answer that
|
||||
// says to come back — a model loading, another holding the GPU, a full
|
||||
// queue, all with Retry-After — is waited out for as long as the call may
|
||||
// wait, saying so meanwhile. One that does not say when, a spent budget, is
|
||||
// not: waiting would not help.
|
||||
func (r *Router) selfJSONWaiting(c mediaCall, model, path string, body, into any) (http.Header, error) {
|
||||
deadline := time.Now().Add(c.wait(nil))
|
||||
for {
|
||||
status, h, b, err := r.self(c, http.MethodPost, path, body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
retry, _ := strconv.Atoi(h.Get("Retry-After"))
|
||||
if status < 300 {
|
||||
return h, json.Unmarshal(b, into)
|
||||
}
|
||||
if retry <= 0 || (status != http.StatusTooManyRequests && status < 500) || !time.Now().Before(deadline) {
|
||||
msg := fmt.Sprintf("POST %s: %d %s", path, status, errorMessage(b))
|
||||
if retry > 0 {
|
||||
msg += "; still not ready after waiting — call again, or ask gateway_status which model holds the GPU"
|
||||
}
|
||||
return h, errors.New(msg)
|
||||
}
|
||||
say := model + " is loading"
|
||||
if held := h.Get(waitingForHeader); held != "" {
|
||||
say = model + " is waiting for " + held + " to give up the GPU"
|
||||
if eta := h.Get(etaHeader); eta != "" {
|
||||
say += ", about " + eta + " s"
|
||||
}
|
||||
}
|
||||
c.progress(5, say)
|
||||
select {
|
||||
case <-c.req.Context().Done():
|
||||
return nil, c.req.Context().Err()
|
||||
case <-time.After(min(time.Duration(retry)*time.Second, 10*time.Second, time.Until(deadline)+time.Second)):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fileURL is where a file is fetched from, with the caller's key, as the
|
||||
// caller reached the gateway.
|
||||
func fileURL(req *http.Request, id string) string {
|
||||
@@ -518,7 +557,7 @@ func (r *Router) makeImage(c mediaCall, prompt, style, negative, model string, s
|
||||
B64 string `json:"b64_json"`
|
||||
} `json:"data"`
|
||||
}
|
||||
h, err := r.selfJSON(c, http.MethodPost, imagesPath, body, &out)
|
||||
h, err := r.selfJSONWaiting(c, model, imagesPath, body, &out)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
@@ -614,6 +653,10 @@ func (r *Router) toolGenerateMesh(c mediaCall, raw json.RawMessage) (map[string]
|
||||
return nil, fmt.Errorf("%s's workflow takes no picture", a.Model)
|
||||
}
|
||||
inputs := map[string]any{image: dataURL(pic)}
|
||||
// A file of the caller's goes by its id, which the API takes.
|
||||
if id := strings.TrimSpace(a.Image); strings.HasPrefix(id, "file-") {
|
||||
inputs[image] = id
|
||||
}
|
||||
for k, v := range a.Inputs {
|
||||
inputs[k] = v
|
||||
}
|
||||
@@ -827,7 +870,7 @@ func (r *Router) toolReviewImages(c mediaCall, raw json.RawMessage) (map[string]
|
||||
} `json:"message"`
|
||||
} `json:"choices"`
|
||||
}
|
||||
if _, err := r.selfJSON(c, http.MethodPost, "/v1/chat/completions", body, &out); err != nil {
|
||||
if _, err := r.selfJSONWaiting(c, a.Model, "/v1/chat/completions", body, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(out.Choices) == 0 {
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"image"
|
||||
"image/png"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
@@ -264,3 +265,107 @@ func TestPicturesAreLookedAtByAVisionModelShrunkAndToASchema(t *testing.T) {
|
||||
t.Errorf("the picture sent is %v (%v), not shrunk to %d", im.Bounds(), err, lookEdge)
|
||||
}
|
||||
}
|
||||
|
||||
// noisePNG is a picture that does not compress: n by n, so its base64 is
|
||||
// past any small limit.
|
||||
func noisePNG(t *testing.T, n int) []byte {
|
||||
t.Helper()
|
||||
im := image.NewRGBA(image.Rect(0, 0, n, n))
|
||||
seed := uint32(1)
|
||||
for i := range im.Pix {
|
||||
seed = seed*1664525 + 1013904223
|
||||
im.Pix[i] = byte(seed >> 24)
|
||||
}
|
||||
var b bytes.Buffer
|
||||
_ = png.Encode(&b, im)
|
||||
return b.Bytes()
|
||||
}
|
||||
|
||||
func TestAPictureOfRealSizeGoesThroughMCPAndByFileID(t *testing.T) {
|
||||
c := newMeshComfy(t)
|
||||
rt, secret := meshRouter(t, c)
|
||||
h := rt.Handler()
|
||||
pic := noisePNG(t, 400) // ~640 KB, as a 1024 px concept is
|
||||
res, _ := mcp(t, h, secret, "tools/call", map[string]any{"name": "generate_mesh", "arguments": map[string]any{
|
||||
"model": "trellis2", "image": "data:image/png;base64," + base64.StdEncoding.EncodeToString(pic)}}, false)
|
||||
if info := resultInfo(t, res); info["status"] != "completed" {
|
||||
t.Fatalf("a mesh from a picture of real size: %v", info)
|
||||
}
|
||||
|
||||
// Uploaded once as a picture to work from, then named by id.
|
||||
var form bytes.Buffer
|
||||
mw := multipart.NewWriter(&form)
|
||||
_ = mw.WriteField("purpose", "vision")
|
||||
fw, _ := mw.CreateFormFile("file", "rock.png")
|
||||
_, _ = fw.Write(pic)
|
||||
_ = mw.Close()
|
||||
req := httptest.NewRequest(http.MethodPost, "/v1/files", &form)
|
||||
req.Header.Set("Content-Type", mw.FormDataContentType())
|
||||
req.Header.Set("Authorization", "Bearer "+secret)
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, req)
|
||||
var f map[string]any
|
||||
if rec.Code != http.StatusOK || json.Unmarshal(rec.Body.Bytes(), &f) != nil || f["purpose"] != "vision" ||
|
||||
f["expires_at"] == nil {
|
||||
t.Fatalf("a picture uploaded to work from: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
res, _ = mcp(t, h, secret, "tools/call", map[string]any{"name": "generate_mesh", "arguments": map[string]any{
|
||||
"model": "trellis2", "image": f["id"]}}, false)
|
||||
if info := resultInfo(t, res); info["status"] != "completed" {
|
||||
t.Fatalf("a mesh from an uploaded picture: %v", info)
|
||||
}
|
||||
}
|
||||
|
||||
func TestACallTooLargeIsSaidSo(t *testing.T) {
|
||||
h := newTestRouter(t, baseConfig()).Handler()
|
||||
big := `{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"x","arguments":{"image":"` +
|
||||
strings.Repeat("A", mcpBody) + `"}}}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/mcp", strings.NewReader(big))
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusRequestEntityTooLarge || !strings.Contains(rec.Body.String(), "larger than") {
|
||||
t.Errorf("%d %s", rec.Code, rec.Body.String()[:min(200, rec.Body.Len())])
|
||||
}
|
||||
}
|
||||
|
||||
func TestACallToAColdModelWaitsItOutSayingSo(t *testing.T) {
|
||||
rt := newTestRouter(t, baseConfig())
|
||||
_ = rt.Handler()
|
||||
calls := 0
|
||||
var fake http.Handler = http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
calls++
|
||||
if calls < 3 {
|
||||
w.Header().Set("Retry-After", "1")
|
||||
w.Header().Set(waitingForHeader, "sf3d")
|
||||
w.Header().Set(etaHeader, "40")
|
||||
writeError(w, http.StatusServiceUnavailable, "server_error", "model loading: waiting")
|
||||
return
|
||||
}
|
||||
w.Header().Set("x-llmgw-seed", "4")
|
||||
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
|
||||
})
|
||||
rt.selfHandler.Store(&fake)
|
||||
var said []string
|
||||
c := mediaCall{req: httptest.NewRequest(http.MethodPost, "/mcp", nil), streamed: true,
|
||||
progress: func(_ float64, m string) { said = append(said, m) }}
|
||||
var out map[string]any
|
||||
h, err := rt.selfJSONWaiting(c, "flux-schnell", imagesPath, map[string]any{}, &out)
|
||||
if err != nil || out["ok"] != true || h.Get("x-llmgw-seed") != "4" || calls != 3 {
|
||||
t.Fatalf("%v %v after %d calls", err, out, calls)
|
||||
}
|
||||
if len(said) != 2 || !strings.Contains(said[0], "waiting for sf3d to give up the GPU, about 40 s") {
|
||||
t.Errorf("said %q", said)
|
||||
}
|
||||
|
||||
// Refused for good — no Retry-After — is said at once.
|
||||
calls = 0
|
||||
fake = http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
calls++
|
||||
writeError(w, http.StatusPaymentRequired, "insufficient_quota", "the budget is spent")
|
||||
})
|
||||
rt.selfHandler.Store(&fake)
|
||||
if _, err := rt.selfJSONWaiting(c, "flux-schnell", imagesPath, map[string]any{}, &out); err == nil ||
|
||||
!strings.Contains(err.Error(), "budget") || calls != 1 {
|
||||
t.Errorf("%v after %d calls", err, calls)
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ package router
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"slices"
|
||||
@@ -121,8 +122,7 @@ func (r *Router) handlePutNote(w http.ResponseWriter, req *http.Request) {
|
||||
Text string `json:"text"`
|
||||
Tags []string `json:"tags"`
|
||||
}
|
||||
// Pictures to look at come in the call, as base64.
|
||||
raw, err := io.ReadAll(http.MaxBytesReader(w, req.Body, mcpBody))
|
||||
raw, err := io.ReadAll(http.MaxBytesReader(w, req.Body, maxMemoryBody))
|
||||
if err != nil || json.Unmarshal(raw, &body) != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request_error",
|
||||
`body must be JSON like {"text": "...", "tags": ["..."]}`)
|
||||
@@ -257,7 +257,17 @@ var mcpTools = []map[string]any{
|
||||
|
||||
// handleMCP serves POST /mcp.
|
||||
func (r *Router) handleMCP(w http.ResponseWriter, req *http.Request) {
|
||||
raw, err := io.ReadAll(http.MaxBytesReader(w, req.Body, maxMemoryBody))
|
||||
// Pictures to make a mesh from or to look at come in the call, as
|
||||
// base64.
|
||||
raw, err := io.ReadAll(http.MaxBytesReader(w, req.Body, mcpBody))
|
||||
if mbe := (*http.MaxBytesError)(nil); errors.As(err, &mbe) {
|
||||
writeJSON(w, http.StatusRequestEntityTooLarge, rpcResponse{
|
||||
JSONRPC: "2.0", ID: json.RawMessage("null"),
|
||||
Error: &rpcError{Code: -32600, Message: fmt.Sprintf("the call is larger than %d MB: send pictures smaller, "+
|
||||
"or fewer at once", mcpBody>>20)},
|
||||
})
|
||||
return
|
||||
}
|
||||
var rpc rpcRequest
|
||||
if err != nil || json.Unmarshal(raw, &rpc) != nil || rpc.Method == "" {
|
||||
writeJSON(w, http.StatusBadRequest, rpcResponse{
|
||||
|
||||
Reference in new issue
Block a user