DemonGatanjieuandClaude Opus 5.5
c45c3b672c
Three exams with Gemini 3.8 Flash on an isolated ComfyUI all went the right way. They showed that a base kept by Anomalous was reported as a file changed outside it, and that the one-pass route still told the AI to show bases first. begin_task's words are shorter: with a real node list it went past its 4000 characters. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
488 lines
31 KiB
Python
488 lines
31 KiB
Python
"""MCP tools that shape the AI workbench's record (ai_tasks.py): begin_task names a piece of
|
||
work, report leaves a note for the user (with pictures made outside Anomalous's tools),
|
||
show_panel / update_panel draw a panel from Anomalous's blocks (ai_panels.py), get_answers
|
||
reads what the user picked or typed on it, log_outside keeps what the AI did with other tools
|
||
(ai_outside.py)."""
|
||
|
||
from . import ai_answer_wait, ai_decisions, ai_marks, ai_message_refs, ai_outside, ai_panels, ai_repaint, ai_run_control, ai_run_tracking, ai_task_links, ai_task_runs, ai_tasks, ai_toolbox, ai_privacy, ai_skills, mcp_routes, reference_blocks
|
||
from .mcp_tools import MAX_LIMIT, ToolError, _schema
|
||
|
||
PACK_TYPES = 150 # node types listed per picked pack
|
||
ASK_WAIT_MIN = 30 # ask_user waits this long even when told not to: the user, reading the work, often answers at once
|
||
ASK_NEXT = ("Before you end your turn, ask_user what to do next or whether to go ahead: a note or a panel alone leaves "
|
||
"the user waiting. Only a run, a download or a scan going on lets you end it without asking.")
|
||
|
||
|
||
def begin_task(ctx, title, new_task=False):
|
||
title = str(title or "").strip()
|
||
if not title:
|
||
raise ToolError("title is empty: a few words on what the user asked for.")
|
||
task = ai_tasks.begin_task(
|
||
ctx.get("client") or "AI", title, new_task=bool(new_task),
|
||
session=ctx.get("session") or "", task_id=ctx.get("task"),
|
||
)
|
||
answer = {"task": task["id"], "title": task["title"], "note": (
|
||
"Continue this task; do not begin another for its steps. Every AMB call is a step the user sees, opens and "
|
||
"undoes in Anomalous's AI workbench. Ask them with ask_user (they answer there, not in your chat); report "
|
||
"explains a choice or hands in pictures, done=true at the end. Run with run_workflow. In the task's own tab, "
|
||
"do not switch tabs yourself. "
|
||
"Everything is in AMB's tools (models, pictures, node types, the canvas, runs and their previews): do not "
|
||
"read ComfyUI's folders or source, run preprocessors or call its routes yourself. Change pictures only by "
|
||
"running workflows, never by editing files with code. If you use another tool anyway, log_outside right "
|
||
"after. The directory below lists the tools by what you want to do.")}
|
||
answer["next"] = ("get_situation shows where things stand; a daily run is one call: run_workflow(workflow: a saved "
|
||
"one, with the prompt, seed or size to change).")
|
||
answer["blocks"] = _blocks(ctx)
|
||
if ai_privacy.is_private(ai_tasks.get_task(task["id"])):
|
||
answer["privacy"] = ai_privacy.NOTE # the user writes the prompts and looks at the pictures: no skill for those
|
||
elif ai_skills.directory():
|
||
answer["skills"] = ai_skills.directory()
|
||
box = ai_toolbox.effective(ai_tasks.get_task(task["id"]))
|
||
if box:
|
||
answer["toolbox"] = ai_toolbox.words(box)
|
||
if _routes(ctx, task["id"]):
|
||
answer["next"] = mcp_routes.TOOLBOX_NOTE
|
||
goes_on = ai_task_links.carried(task)
|
||
if goes_on:
|
||
answer["continues"] = goes_on # the task this one goes on from, and what was decided there
|
||
decided = ai_decisions.lines(ai_tasks.get_task(task["id"]))
|
||
if decided:
|
||
answer["decisions"] = decided # what was settled so far, the user's own choices in Anomalous too
|
||
if task.get("about"):
|
||
# Handed over from a model page, an image or a combo in Anomalous: the ids the tools take.
|
||
answer["about"] = task["about"]
|
||
answer["about_note"] = ("The user started this task from these in Anomalous; work with them: model_id goes to "
|
||
"set_model / add_lora / get_model, image to get_image_info, a combo's name to place_combo, "
|
||
"a workflow's name to open_workflow.")
|
||
return answer
|
||
|
||
|
||
def _blocks(ctx):
|
||
"""The ready-made blocks (reference_blocks.py), each a line, for begin_task's answer: what to reach for first."""
|
||
from .mcp_canvas_edit import object_info # the node list, to say which blocks need a pack first
|
||
try:
|
||
info = object_info(ctx) if ctx.get("base_url") else None
|
||
except ToolError:
|
||
info = None
|
||
return {"note": "add_nodes(use=…, want=…) builds the best one this ComfyUI can run, wired in; prefer it.",
|
||
**reference_blocks.directory(info)}
|
||
|
||
|
||
def report(ctx, text="", images=None, done=False, reply_to=None, task=None):
|
||
try:
|
||
step, missing = ai_tasks.report(
|
||
ctx.get("client") or "AI", text, images or [], bool(done),
|
||
session=ctx.get("session") or "", task_id=task or ctx.get("task"), reply_to=reply_to,
|
||
)
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
answer = {"noted": True, "pictures": len(step["results"])}
|
||
if not done:
|
||
answer["next"] = ASK_NEXT
|
||
if any(mark in str(text or "") for mark in ("?", "？")):
|
||
answer["next"] = "Your note asks the user something: ask it with ask_user (options to press) instead. " + ASK_NEXT
|
||
if missing:
|
||
answer["not_found"] = missing
|
||
answer["note"] = "Pictures are paths inside ComfyUI's output folder (subfolder/name.png), as search_images gives them."
|
||
return answer
|
||
|
||
|
||
def log_outside(ctx, what, why="", files=None, task=None):
|
||
try:
|
||
step = ai_outside.log(ctx.get("client") or "AI", what, why, files, session=ctx.get("session") or "",
|
||
task_id=task or ctx.get("task"))
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
return {"logged": True, "step": step["id"]}
|
||
|
||
|
||
def log_decision(ctx, title, decision, reason="", rejected=None, by="user", related=None, task=None):
|
||
try:
|
||
step = ai_decisions.log(ctx.get("client") or "AI", title, decision, reason, rejected, by, related,
|
||
session=ctx.get("session") or "", task_id=task or ctx.get("task"))
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
return {"recorded": True, "step": step["id"], "related": step["decision"]["related"]}
|
||
|
||
|
||
def show_panel(ctx, title, blocks, task=None):
|
||
# A prompt box tells the user what suits the canvas's model (the prompt_pose skill's words for them).
|
||
boxed = isinstance(blocks, list) and any(isinstance(block, dict) and block.get("type") == "prompts" for block in blocks)
|
||
advice = ai_skills.for_user(_canvas_model()) if boxed else None
|
||
try:
|
||
panel = ai_panels.show(
|
||
ctx.get("client") or "AI", title, blocks,
|
||
session=ctx.get("session") or "", task_id=task or ctx.get("task"), extra={"prompt_advice": advice} if advice else None,
|
||
)
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
asks = [block["id"] for block in panel["blocks"] if block["type"] in ai_panels.INTERACTIVE]
|
||
answer = {"panel": panel["id"], "shown": True,
|
||
"blocks": [{"id": block["id"], "type": block["type"]} for block in panel["blocks"]]}
|
||
if asks:
|
||
answer["questions"] = asks
|
||
answer["note"] = ("The user answers in Anomalous when they get to it; call get_answers with this panel id "
|
||
"(wait_seconds up to 55 waits for it) before going on with what depends on it.")
|
||
else:
|
||
answer["next"] = ASK_NEXT
|
||
return answer
|
||
|
||
|
||
def _routes(ctx, task):
|
||
"""What the user can do with the two models they picked (mcp_routes.py), or None."""
|
||
from .mcp_canvas_edit import object_info
|
||
|
||
task_id = task or ctx.get("task") or ai_tasks.active_task_id(ctx.get("client") or "AI", ctx.get("session") or "")
|
||
box = ai_toolbox.effective(ai_tasks.get_task(task_id)) if task_id else None
|
||
return mcp_routes.offer(box, object_info(ctx)) if len(mcp_routes._main(box)) >= 2 else None
|
||
|
||
|
||
def ask_user(ctx, question="", options=None, multiple=False, other=True, wait_seconds=ai_panels.MAX_WAIT, task=None, routes=False):
|
||
"""A question in the AI workbench, where the user watches the work: options to press (one may be
|
||
recommended), a line of their own; waits for the answer. `routes`: Anomalous's own question and options."""
|
||
extra, words = None, {}
|
||
if routes:
|
||
offered = _routes(ctx, task)
|
||
if not offered:
|
||
raise ToolError("No routes to ask about: they need two main models of one family picked for the task. Go on with what the user asked.")
|
||
question, options, multiple, other = mcp_routes.QUESTION, offered["options"], False, True
|
||
extra, words = {"routes": offered["calls"]}, {"title_key": "aiRouteTitle"}
|
||
if not str(question or "").strip():
|
||
raise ToolError("question is empty.")
|
||
block = ({"type": "choices", "id": "answer", "options": options, "multiple": bool(multiple), "other": bool(other)}
|
||
if options else {"type": "input", "id": "answer", "question": str(question or "")[:200]})
|
||
try:
|
||
panel = ai_panels.show(ctx.get("client") or "AI", question, [block], session=ctx.get("session") or "",
|
||
task_id=task or ctx.get("task"), tool="ask_user", extra=extra, **words)
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
return _answer_of(ctx, panel["id"], wait_seconds)
|
||
|
||
|
||
def propose_repaint_area(ctx, title, marks, areas=None, grow=12, feather=8, note="", wait_seconds=ai_panels.MAX_WAIT):
|
||
"""The area to repaint for the user's marks, theirs beside the AI's for them to pick (ai_repaint.py); waits like ask_user."""
|
||
try:
|
||
panel = ai_repaint.propose(ctx.get("client") or "AI", title, str(marks or ""), areas, grow, feather,
|
||
str(note or "")[:400], session=ctx.get("session") or "")
|
||
except ai_repaint.RepaintError as error:
|
||
raise ToolError(str(error))
|
||
return _answer_of(ctx, panel["id"], wait_seconds)
|
||
|
||
|
||
def _answer_of(ctx, panel_id, wait_seconds):
|
||
"""The answer to the question of panel `panel_id` once the user gives it, waiting at least ASK_WAIT_MIN."""
|
||
client = ctx.get("client") or "AI"
|
||
try:
|
||
answer = ai_answer_wait.collect(client, panel_id, max(ASK_WAIT_MIN, float(wait_seconds or 0)))
|
||
given = next((item for item in answer["answers"] if item["question"] == "answer"), None)
|
||
if not given:
|
||
if answer.get("messages") and answer.get("still_waiting"):
|
||
# Read the user's new instruction before considering an automatic choice.
|
||
answer.update(ai_answer_wait.wait_on(panel_id, allow_recommended=False))
|
||
return {"answered": False, "panel": panel_id, **answer}
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
out = {"answered": True, "panel": panel_id, "next": "Go on with the user's answer; log_decision it when it settles a choice."}
|
||
if answer.get("messages"):
|
||
out["messages"] = answer["messages"]
|
||
if given.get("inpaint_image"):
|
||
out.update(inpaint_image=given["inpaint_image"], inpaint_note=given["note"])
|
||
elif given.get("note"):
|
||
out["next"] = given["note"]
|
||
if given.get("skipped"):
|
||
out.update(skipped=True, note="The user left this to you: decide it yourself, log_decision what you chose and "
|
||
"go on. Do not ask the same again.")
|
||
elif given.get("text"):
|
||
out["text"] = given["text"]
|
||
if given.get("choice") == ["other"]:
|
||
out["note"] = "None of your options: the user wrote their own answer."
|
||
else:
|
||
out.update(choice=given.get("choice"), labels=given.get("labels"))
|
||
if given.get("recommended_on_timeout"):
|
||
out.update(recommended_on_timeout=True, next=ai_answer_wait.TOOK)
|
||
return out
|
||
|
||
|
||
def update_panel(ctx, panel, title=None, blocks=None, closed=None, task=None):
|
||
del ctx, task
|
||
try:
|
||
updated = ai_panels.update(panel, title, blocks, closed)
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
return {"updated": True, "blocks": [{"id": block.get("id"), "type": block["type"]}
|
||
for block in updated["blocks"]]}
|
||
|
||
|
||
def run_status(ctx, wait_seconds=0, task=None):
|
||
try:
|
||
return ai_task_runs.status(ctx.get("client") or "AI", ctx.get("session") or "", task or ctx.get("task"), wait_seconds)
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
|
||
|
||
def show_run_map(ctx, title, run=None, task=None):
|
||
client, session = ctx.get("client") or "AI", ctx.get("session") or ""
|
||
task_id = ai_tasks.active_task_id(client, session, task or ctx.get("task"))
|
||
runs = [step for step in (ai_tasks.get_task(task_id) or {}).get("steps", []) if step.get("tool") == "run_workflow"] if task_id else []
|
||
found = next((step for step in reversed(runs) if step.get("map") and (not run or step["id"] == run)), None)
|
||
if not found:
|
||
raise ToolError("No run of this task has its shape yet: it is read once ComfyUI has queued the run (run_workflow).")
|
||
try:
|
||
panel = ai_panels.show(client, title, [{"type": "run_map", "step": found["id"]}], session=session, task_id=task_id,
|
||
tool="show_run_map")
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
boxes = found["map"]["boxes"]
|
||
return {"panel": panel["id"], "shown": True, "run": found["id"], "changes": found.get("changes") or [],
|
||
"boxes": [{"node_id": box["id"], "title": box["title"], "role": box["role"]} for box in boxes]}
|
||
|
||
|
||
def stop_run(ctx, task=None):
|
||
client, session = ctx.get("client") or "AI", ctx.get("session") or ""
|
||
task_id = ai_tasks.active_task_id(client, session, task or ctx.get("task"))
|
||
run = next((step for step in reversed((ai_tasks.get_task(task_id) or {}).get("steps", []))
|
||
if step.get("tool") == "run_workflow"), None) if task_id else None
|
||
if not run:
|
||
raise ToolError("No run in this task yet.")
|
||
try:
|
||
done = ai_run_control.stop(run["id"], by="ai", task_id=task_id)
|
||
except ai_run_control.RunControlError as error:
|
||
raise ToolError(str(error))
|
||
return {**done, "stopped": True, "note": "Stopping: run_status(wait_seconds=10) confirms it ended."}
|
||
|
||
|
||
def get_toolbox(ctx, task=None, skill=None):
|
||
from .mcp_canvas_edit import find_nodes
|
||
|
||
task_id = task or ctx.get("task") or ai_tasks.active_task_id(ctx.get("client") or "AI", ctx.get("session") or "")
|
||
if skill:
|
||
private = ai_privacy.applies(task_id)
|
||
try:
|
||
answer = ai_skills.get(str(skill), _canvas_model() if ai_skills.by_model(str(skill)) else None, private)
|
||
except ai_skills.SkillError as error:
|
||
raise ToolError(str(error))
|
||
ai_skills.note_use(ctx.get("client") or "AI", answer, session=ctx.get("session") or "", task_id=task or ctx.get("task"))
|
||
return answer
|
||
box = ai_toolbox.effective(ai_tasks.get_task(task_id)) if task_id else None
|
||
if not box:
|
||
return {"picked": False, "note": "The user picked nothing for this task; use what fits."}
|
||
answer = {"picked": True, "mode": box["mode"], "note": ai_toolbox.words(box)}
|
||
if box.get("nodes"):
|
||
found = find_nodes(ctx, types=box["nodes"])
|
||
answer["nodes"] = found["nodes"]
|
||
if found.get("not_found"):
|
||
answer["page_only_nodes"] = found["not_found"] # helpers that live only in the page: add_node takes them
|
||
if box.get("packs"):
|
||
from .mcp_canvas_edit import object_info
|
||
|
||
info = object_info(ctx)
|
||
answer["packs"] = []
|
||
for pack in box["packs"]:
|
||
types = sorted(name for name, item in info.items() if ai_toolbox.source_of(item) == pack)
|
||
answer["packs"].append({"pack": pack, "count": len(types), "node_types": [
|
||
{"type": name, "display_name": info[name].get("display_name") or name, "category": info[name].get("category") or ""}
|
||
for name in types[:PACK_TYPES]]})
|
||
answer["packs_note"] = "Every node type of a pack is the user's pick; find_nodes(types=[...]) gives one in full."
|
||
for kind in ("combos", "workflows", "prompts"):
|
||
if box.get(kind):
|
||
answer[kind] = [found for name in box[kind] if (found := ai_message_refs.lookup(kind, name)) is not None]
|
||
if box.get("models"):
|
||
answer["models"] = [ai_message_refs.lookup("models", model_id) or {"model_id": model_id, "missing": True} for model_id in box["models"]]
|
||
for model in answer["models"]:
|
||
if not model.get("missing") and model["model_id"].split(":", 1)[0] != "loras":
|
||
model["in_cards"] = mcp_routes.cards_of(model["model_id"]) # its examples: how the user's own routes use it
|
||
if _routes(ctx, task_id):
|
||
answer["routes"] = mcp_routes.TOOLBOX_NOTE
|
||
return answer
|
||
|
||
|
||
def _canvas_model():
|
||
"""The main model the canvas's sampler uses {name, params}, for advice that depends on its size; None when
|
||
the page cannot tell (no page, several samplers, no loader)."""
|
||
from .mcp_actions import _page
|
||
from .model_advice import main_params
|
||
|
||
try:
|
||
name = (_page("block_model", {"near_node_id": None}) or {}).get("model") or ""
|
||
except ToolError:
|
||
return None
|
||
return {"name": name, "params": main_params(name)} if name else None
|
||
|
||
|
||
def get_answers(ctx, panel=None, wait_seconds=0, task=None):
|
||
try:
|
||
client = ctx.get("client") or "AI"
|
||
task_id = task or ctx.get("task") or None
|
||
if not task_id and ctx.get("session"):
|
||
task_id = ai_tasks.active_task_id(client, ctx["session"]) or ""
|
||
answer = ai_answer_wait.collect(client, panel, wait_seconds, task_id=task_id)
|
||
return answer, ai_marks.image_items(answer.get("messages") or []) # marked pictures, as pictures
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
|
||
|
||
BLOCKS_HELP = (
|
||
"blocks, top to bottom, each {type, ...}: heading {text}; text {text, tone: normal|muted|tip|warning}; "
|
||
"images {items: [{image: 'subfolder/name.png' from search_images, caption, marks: [{x, y, r | w, h, label, "
|
||
"color: red|green|blue|yellow|purple}]}], layout: row|grid|compare} (marks: circles or boxes with a label, x/y/w/h/r "
|
||
"as fractions 0-1 of the picture, x/y the centre; a click on the label zooms there); "
|
||
"slider {before: {image}, after: {image}, before_label, after_label} (one picture over the other, a line to drag); "
|
||
"picks {id, question, items: [{id, image, label, reason, recommended}] (2-12), multiple} (pictures to choose from; "
|
||
"the user can also put one's workflow on the canvas); slider and picks: zoom {x, y, w, h} shows only that part, enlarged; "
|
||
"diagram {nodes: [{id, label, detail, node_id, color}], edges: [{from, to, label}], direction: right|down} (boxes and "
|
||
"arrows Anomalous lays out; node_id: a canvas node it finds); "
|
||
"models {items: [{model_id from search_models, note}]} (cards with their covers, a click opens them); "
|
||
"table {columns: [...], rows: [[...]]}; list {items: [...], ordered}; params {items: [{label, value}]}; "
|
||
"choices {id, question, options: [{id, label, detail}] (2-8), multiple} (buttons the user presses); "
|
||
"input {id, question, placeholder} (a line the user types); progress {label, value, max} (your own "
|
||
"numbers, kept current with update_panel; none for a run: its progress shows already); divider {}. "
|
||
"Each block may have a stable id; the tool returns block ids in display order. "
|
||
"Plain text only (no markup, no links); up to 30 blocks, 12 pictures or models a block."
|
||
)
|
||
|
||
_TASK_PROP = {"type": "string", "description": "begin_task's id (optional)."}
|
||
|
||
|
||
def _task_tool(name, title, description, run, properties, required=()):
|
||
return name, {"run": run, "context": True, "spec": {
|
||
"name": name, "title": title, "description": description,
|
||
"inputSchema": _schema(properties, required),
|
||
"annotations": {"title": title, "readOnlyHint": name in ("run_status", "get_toolbox"), "destructiveHint": False,
|
||
"idempotentHint": False, "openWorldHint": False},
|
||
}}
|
||
|
||
|
||
def _question_with_images(run):
|
||
"""The question receipt also carries pictures the user sent while it was waiting."""
|
||
def call(ctx, **arguments):
|
||
result = run(ctx, **arguments)
|
||
return result, ai_marks.image_items(result.get("messages") or [])
|
||
return call
|
||
|
||
|
||
TASK_TOOLS = dict([
|
||
_task_tool("track_run", "Track an externally submitted run",
|
||
"Immediately after YOU submit to this ComfyUI server via /prompt or another tool, attach the exact "
|
||
"prompt_ids returned by your submissions to your task. This does NOT submit or rerun anything. "
|
||
"The workbench then shows real queue/node progress and results. Never use another user's IDs or "
|
||
"guess from queue differences. Repeating the same IDs is safe. Waits up to wait_seconds and returns "
|
||
"state, summary, pictures and errors; if still running, keep waiting with run_status(wait_seconds=55).",
|
||
ai_run_tracking.track_run, {"prompt_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 8},
|
||
"task": _TASK_PROP, "wait_seconds": {"type": "number", "minimum": 0, "maximum": 55, "default": 15}}, ["prompt_ids"]),
|
||
_task_tool("begin_task", "Begin a task",
|
||
"Call first: names what the user asked for in Anomalous's AI workbench, where they see, open and undo "
|
||
"each of your steps. Continues the waiting or active task; new_task=true only for a separate request.",
|
||
begin_task, {"title": {"type": "string", "description": "A few words, in the user's language."},
|
||
"new_task": {"type": "boolean", "default": False}},
|
||
["title"]),
|
||
_task_tool("report", "Tell the user",
|
||
"A note in the AI workbench, where the user reads your work: your choices, what to look at, what is "
|
||
"left, your answers. Markdown headings, lists, bold and code are laid out. reply_to: the id of the "
|
||
"user's message you answer. images: output pictures (subfolder/name.png) to keep with the task. "
|
||
"done=true ends it.",
|
||
report, {"text": {"type": "string"},
|
||
"images": {"type": "array", "items": {"type": "string"}, "maxItems": MAX_LIMIT},
|
||
"done": {"type": "boolean", "default": False},
|
||
"reply_to": {"type": "string", "description": "The message id you answer."},
|
||
"task": _TASK_PROP}),
|
||
_task_tool("log_outside", "Say what you did outside Anomalous",
|
||
"Right after using a tool that is not Anomalous's (a command, a script, files, another server): what "
|
||
"you ran, why, and the files it touched, for the task's timeline. Prefer Anomalous's tools.",
|
||
log_outside, {"what": {"type": "string"}, "why": {"type": "string"},
|
||
"files": {"type": "array", "items": {"type": "string"}, "maxItems": ai_outside.MAX_FILES},
|
||
"task": _TASK_PROP},
|
||
["what"]),
|
||
_task_tool("log_decision", "Record a decision",
|
||
"Records a settled decision of the task (a style, a model, a fix, what to drop) where the user looks "
|
||
"back on it: title, decision, reason, rejected options, by (user, ai or together), related step ids. "
|
||
"The user's own choices in Anomalous are kept already.",
|
||
log_decision, {"title": {"type": "string"}, "decision": {"type": "string"}, "reason": {"type": "string"},
|
||
"rejected": {"type": "array", "items": {"type": "string"}, "maxItems": ai_decisions.MAX_REJECTED},
|
||
"by": {"type": "string", "enum": list(ai_decisions.BY), "default": "user"},
|
||
"related": {"type": "array", "items": {"type": "string"}, "maxItems": ai_decisions.MAX_RELATED},
|
||
"task": _TASK_PROP},
|
||
["title", "decision"]),
|
||
_task_tool("show_panel", "Show the user a panel",
|
||
"Draws a panel in the AI workbench, in Anomalous's own style, from blocks: to compare pictures side by "
|
||
"side, present models with their covers, lay out settings, a table, and to ask (buttons or a line of "
|
||
"text). Use a new panel for each new phase (plan, workflow, then results), in the order they happen. "
|
||
"Use it when a choice is the user's, or to show results better than words do. " + BLOCKS_HELP,
|
||
show_panel, {"title": {"type": "string"},
|
||
"blocks": {"type": "array", "items": {"type": "object"}, "maxItems": ai_panels.MAX_BLOCKS},
|
||
"task": _TASK_PROP},
|
||
["title", "blocks"]),
|
||
_task_tool("ask_user", "Ask the user",
|
||
"Asks the user in the AI workbench and waits: a question with 2-8 options ({label, detail, "
|
||
"recommended}; mark your suggestion recommended=true), and a line for their own answer unless "
|
||
"other=false; no options: a free answer. routes=true: Anomalous's own question on the two models picked. "
|
||
"Use it for every choice that is the user's, not your chat's "
|
||
f"question form. Waits at least {ASK_WAIT_MIN} s, up to wait_seconds; unanswered and keep_waiting: wait on "
|
||
"with get_answers. End every round of work with it.",
|
||
_question_with_images(ask_user), {"question": {"type": "string"},
|
||
"options": {"type": "array", "maxItems": 8, "items": {"type": "object", "properties": {
|
||
"label": {"type": "string"}, "detail": {"type": "string"}, "recommended": {"type": "boolean"}}}},
|
||
"multiple": {"type": "boolean", "default": False}, "other": {"type": "boolean", "default": True},
|
||
"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_panels.MAX_WAIT, "default": ai_panels.MAX_WAIT},
|
||
"routes": {"type": "boolean"}, "task": _TASK_PROP}),
|
||
_task_tool("propose_repaint_area", "Agree on the area to repaint",
|
||
"Before repainting only the marked areas: the user's marks are rough, so show them the area you would "
|
||
"repaint. marks: the marks' message id. areas: boxes or ellipses {shape, x, y, w, h} as fractions of the "
|
||
"picture (x, y its top left), or none to keep the marks; grow: pixels added around; feather: pixels of "
|
||
"soft edge. The user sees theirs and yours side by side and picks one, or marks again; with an area "
|
||
"picked the answer has inpaint_image (the picture with that area as its alpha, in ComfyUI's input "
|
||
"folder) and inpaint_note: add_nodes(use=\"repaint\", image=inpaint_image) builds the repaint. "
|
||
"Waits like ask_user.",
|
||
_question_with_images(propose_repaint_area), {"title": {"type": "string", "description": "A few words, in the user's language."},
|
||
"marks": {"type": "string"},
|
||
"areas": {"type": "array", "maxItems": ai_repaint.MAX_AREAS, "items": {"type": "object", "properties": {
|
||
"shape": {"type": "string", "enum": list(ai_repaint.SHAPES)}, "x": {"type": "number"},
|
||
"y": {"type": "number"}, "w": {"type": "number"}, "h": {"type": "number"}}}},
|
||
"grow": {"type": "integer", "minimum": 0, "maximum": ai_repaint.MAX_GROW, "default": 12},
|
||
"feather": {"type": "integer", "minimum": 0, "maximum": ai_repaint.MAX_FEATHER, "default": 8},
|
||
"note": {"type": "string", "description": "Why you moved it, in a line."},
|
||
"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_panels.MAX_WAIT, "default": ai_panels.MAX_WAIT}},
|
||
["title", "marks"]),
|
||
_task_tool("update_panel", "Change a panel",
|
||
"Adds new output at the BOTTOM of an existing panel, preserving earlier content and its order. "
|
||
"Send only the new blocks; repeated unchanged blocks are kept once. To deliberately correct an "
|
||
"existing block or update a progress bar/question in place, include its returned block id and "
|
||
"the same type. Omitted blocks are retained, not deleted. Never reuse an old block id for a new "
|
||
"result: append it or use show_panel for the next phase. Up to 30 total blocks per panel. "
|
||
"Can also change title, or set closed=true to stop answers; existing answers stay.",
|
||
update_panel, {"panel": {"type": "string"}, "title": {"type": "string"},
|
||
"blocks": {"type": "array", "items": {"type": "object"}, "maxItems": ai_panels.MAX_BLOCKS},
|
||
"closed": {"type": "boolean"}, "task": _TASK_PROP}, ["panel"]),
|
||
_task_tool("run_status", "How the run is going",
|
||
"Your task's latest run: none, queued (runs ahead), running (node, steps, percent), done (its "
|
||
"pictures; the first answer telling you it ended attaches up to 4) or failed; summary says it in a "
|
||
"sentence. wait_seconds (up to 55) waits for the end. Call it instead of guessing or looking in "
|
||
"folders.",
|
||
run_status, {"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_task_runs.MAX_STATUS_WAIT, "default": 0},
|
||
"task": _TASK_PROP}),
|
||
_task_tool("show_run_map", "Show what the run really was",
|
||
"Puts the shape of your task's latest run (or `run`, a run step id) on the user's stage: the nodes it "
|
||
"really ran as boxes (models, LoRAs with weights, prompts, size, sampler settings, saving), linked, "
|
||
"each one locates its node on the canvas, with what changed since the run before. Anomalous reads it "
|
||
"from the prompt ComfyUI queued, so use this instead of drawing the workflow yourself; add your "
|
||
"explanation with report. title: a few words in the user's language.",
|
||
show_run_map, {"title": {"type": "string"}, "run": {"type": "string"}, "task": _TASK_PROP}, ["title"]),
|
||
_task_tool("stop_run", "Stop the run",
|
||
"Stops your task's run: its queued prompts leave the queue, the running one is interrupted (never "
|
||
"anyone else's). When run_status warns it is far too big and the user agrees, or when the user asks.",
|
||
stop_run, {"task": _TASK_PROP}),
|
||
_task_tool("get_toolbox", "What the user picked for you",
|
||
"The node types, node packs, combos, workflows, prompts and models the user picked for this task, in "
|
||
"full, and how strictly: prefer (use them first) or only (add_node and new_workflow accept only those "
|
||
"node types, the packs' and ComfyUI's built-in ones). begin_task says when there are any. "
|
||
"skill: a skill begin_task lists, instead.",
|
||
get_toolbox, {"task": _TASK_PROP, "skill": {"type": "string"}}),
|
||
_task_tool("get_answers", "Read the user's answers and messages",
|
||
"What the user picked or typed on your panels and wrote to you in the AI workbench since you last "
|
||
"asked, each once (kind=change: something they changed themselves; keep it). panel: one panel. "
|
||
"wait_seconds (up to 55) waits when nothing is there. still_waiting: questions not answered, and "
|
||
"keep_waiting=true: call it again; skipped=true: they left it to you, decide yourself.",
|
||
get_answers, {"panel": {"type": "string"},
|
||
"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_panels.MAX_WAIT, "default": 0},
|
||
"task": _TASK_PROP}),
|
||
])
|