From edeeb9f8e150fdbedf21080ea37a81ad15d412af Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Thu, 15 Oct 2026 07:24:59 +0200 Subject: [PATCH] MCP takes pictures of real size, by id too, and waits for a cold model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CHANGELOG.md | 4 ++ docs/clients.md | 9 +++ internal/router/files.go | 22 ++++++- internal/router/mcp_media.go | 47 ++++++++++++- internal/router/mcp_media_test.go | 105 ++++++++++++++++++++++++++++++ internal/router/memory.go | 16 ++++- 6 files changed, 195 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index df11630..2c9fefe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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-`, on the first member's diff --git a/docs/clients.md b/docs/clients.md index 9936ef3..f22f743 100644 --- a/docs/clients.md +++ b/docs/clients.md @@ -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 diff --git a/internal/router/files.go b/internal/router/files.go index f429229..76056cb 100644 --- a/internal/router/files.go +++ b/internal/router/files.go @@ -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 diff --git a/internal/router/mcp_media.go b/internal/router/mcp_media.go index 70b3ad7..e6056f1 100644 --- a/internal/router/mcp_media.go +++ b/internal/router/mcp_media.go @@ -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 { diff --git a/internal/router/mcp_media_test.go b/internal/router/mcp_media_test.go index 235dd64..6368367 100644 --- a/internal/router/mcp_media_test.go +++ b/internal/router/mcp_media_test.go @@ -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) + } +} diff --git a/internal/router/memory.go b/internal/router/memory.go index d5c98f0..7d9c4c5 100644 --- a/internal/router/memory.go +++ b/internal/router/memory.go @@ -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{