feat(mcp): Aurélio Embroidery MCP server — 6 tools

Tools:
- embroidery_formats — list 47 readable, 19 writable formats
- embroidery_machines — list all 15 vendors from registry
- embroidery_match_machine — match model name to vendor entry
- embroidery_convert — convert between any supported formats
- embroidery_preview — render PNG stitch visualization
- embroidery_send — convert + write to Gotek USB stick
- embroidery_info — get stitch count, colors, dimensions, time

Built on FastMCP 3.4 + pyembroidery 1.5.1.
Loads vendor-registry/registry.yaml at startup.
This commit is contained in:
Fábio Coutada 2026-08-09 02:33:21 +01:00
parent 18a27ab9ff
commit 76188bf0f2
3 changed files with 373 additions and 0 deletions

Binary file not shown.

6
mcp/requirements.txt Normal file
View file

@ -0,0 +1,6 @@
# Aurélio Embroidery MCP Server
# FastMCP >= 3.0, pyembroidery >= 1.5
fastmcp>=3.0
pyembroidery>=1.5
pyyaml>=6.0

367
mcp/server.py Normal file
View file

@ -0,0 +1,367 @@
"""Aurélio Embroidery MCP Server — the bridge between design and machine.
Tools: convert, preview, send, formats, machines, match-machine.
Loads vendor registry at startup. Uses pyembroidery for all format operations.
"""
from pathlib import Path
import yaml
from fastmcp import FastMCP
mcp = FastMCP("Aurélio Embroidery")
# ── Load vendor registry ─────────────────────────────────────────────
REGISTRY_PATH = Path(__file__).resolve().parent.parent / "vendor-registry" / "registry.yaml"
with open(REGISTRY_PATH) as f:
VENDOR_REGISTRY = yaml.safe_load(f)
VENDORS: list[dict] = VENDOR_REGISTRY["vendors"]
# ── Tools ─────────────────────────────────────────────────────────────
@mcp.tool()
def embroidery_formats() -> dict:
"""List all supported embroidery formats with read/write capabilities."""
import pyembroidery as pe
# pyembroidery 1.5.1: supported_formats() returns a generator of dicts with
# keys: 'description', 'extension', 'extensions', 'mimetype', 'category', 'reader', 'writer'
sf = list(pe.supported_formats())
readable = sorted(
[f["extension"] for f in sf if f.get("reader")],
key=str.lower,
)
writable = sorted(
[f["extension"] for f in sf if f.get("writer")],
key=str.lower,
)
return {
"readable": readable,
"writable": writable,
"total_readable": len(readable),
"total_writable": len(writable),
}
@mcp.tool()
def embroidery_machines(vendor: str | None = None) -> list[dict]:
"""List known embroidery machines. Filter by vendor name if provided."""
vendors = VENDORS
if vendor:
vendors = [v for v in vendors if vendor.lower() in v["vendor"].lower()]
return [
{
"vendor": v["vendor"],
"models": v["models"],
"formats": v["formats"],
"media": v["media"],
"connectivity": v.get("connectivity", []),
}
for v in vendors
]
@mcp.tool()
def embroidery_match_machine(model: str) -> dict:
"""Match a machine model name to its vendor entry.
Returns format, media type, jumper config, and connectivity options.
Example: embroidery_match_machine('Tajima TME-612')
"""
for v in VENDORS:
for pattern in v.get("models", []):
clean = pattern.rstrip("*").lower()
if clean and clean in model.lower():
return {
"matched": True,
"model_query": model,
"matched_pattern": pattern,
**{k: v[k] for k in ["vendor", "formats", "media", "floppy", "jumpers", "connectivity", "notes"] if k in v},
}
return {
"matched": False,
"model_query": model,
"suggestion": "Try a broader search or add this machine to the vendor registry.",
}
@mcp.tool()
def embroidery_convert(
input_path: str,
output_path: str,
output_format: str | None = None,
) -> dict:
"""Convert between embroidery file formats.
Reads any supported format, writes to the target format.
Format is auto-detected from output_path extension if not specified.
Supported: DST, PES, EXP, JEF, VP3, HUS, VIP, SHV, SEW, XXX,
PEC, PCS, EMB, TBF, U01, DSB, DSZ, 10o, CSV, JSON, SVG.
Args:
input_path: Absolute path to input embroidery file
output_path: Absolute path for output file
output_format: Optional format extension (e.g. 'dst', 'pes')
Returns:
dict with success, input/output formats, stitch count, thread count
"""
import pyembroidery as pe
inp = Path(input_path).expanduser().resolve()
out = Path(output_path).expanduser().resolve()
if not inp.exists():
return {"error": f"Input file not found: {inp}"}
fmt = output_format or out.suffix.lstrip(".").lower()
try:
pattern = pe.read(str(inp))
except Exception as e:
return {"error": f"Failed to read {inp}: {e}"}
try:
pe.write(pattern, str(out))
except Exception as e:
return {"error": f"Failed to write {out} as {fmt}: {e}"}
stitch_count = (
pattern.count_stitches()
if hasattr(pattern, "count_stitches")
else len(getattr(pattern, "stitches", []))
)
thread_count = (
len(pattern.threadlist)
if hasattr(pattern, "threadlist")
else 0
)
return {
"success": True,
"input": str(inp),
"input_format": inp.suffix.lstrip("."),
"output": str(out),
"output_format": fmt,
"stitch_count": stitch_count,
"thread_count": thread_count,
}
@mcp.tool()
def embroidery_preview(
input_path: str,
output_path: str | None = None,
width: int = 800,
height: int = 600,
background: str = "#1A1A2E",
) -> dict:
"""Generate a PNG preview of an embroidery design.
Renders all stitches as colored lines on a dark background.
If output_path is omitted, saves next to the input as <name>_preview.png.
Args:
input_path: Path to embroidery file
output_path: Optional output PNG path
width: Image width in pixels (default 800)
height: Image height in pixels (default 600)
background: Background color as hex (default dark navy)
Returns:
dict with output path, stitch count, bounds in mm
"""
import pyembroidery as pe
from PIL import Image, ImageDraw
inp = Path(input_path).expanduser().resolve()
if not inp.exists():
return {"error": f"Input file not found: {inp}"}
out = Path(output_path).expanduser().resolve() if output_path else inp.with_suffix(".preview.png")
try:
pattern = pe.read(str(inp))
except Exception as e:
return {"error": f"Failed to read {inp}: {e}"}
# Collect stitches
stitchblocks = getattr(pattern, "get_as_stitchblocks", None)
stitches = stitchblocks() if stitchblocks else getattr(pattern, "stitches", [])
if not stitches:
return {"error": "No stitches found in pattern"}
# Bounding box
xs = [s[0] for s in stitches if isinstance(s, (list, tuple)) and len(s) >= 2]
ys = [s[1] for s in stitches if isinstance(s, (list, tuple)) and len(s) >= 2]
if not xs:
return {"error": "No stitch coordinates found"}
min_x, max_x = min(xs), max(xs)
min_y, max_y = min(ys), max(ys)
rw, rh = max_x - min_x + 1, max_y - min_y + 1
pad = 30
scale = min((width - pad * 2) / rw, (height - pad * 2) / rh)
# Draw
img = Image.new("RGB", (width, height), background)
draw = ImageDraw.Draw(img)
colors = [
"#E74C3C", "#E67E22", "#F1C40F", "#2ECC71", "#3498DB",
"#9B59B6", "#1ABC9C", "#E84393", "#00B894", "#6C5CE7",
]
prev_x = prev_y = None
color_idx = 0
for s in stitches:
if len(s) < 2:
continue
cmd = s[2] if len(s) >= 3 else 0
import pyembroidery as pe
if cmd == pe.COLOR_CHANGE:
color_idx = (color_idx + 1) % len(colors)
prev_x = prev_y = None
continue
if cmd == pe.END:
break
x = pad + int((s[0] - min_x) * scale)
y = pad + int((s[1] - min_y) * scale)
if cmd == pe.JUMP:
prev_x = prev_y = None
continue
if prev_x is not None and prev_y is not None:
draw.line([(prev_x, prev_y), (x, y)], fill=colors[color_idx], width=1)
prev_x, prev_y = x, y
img.save(str(out))
return {
"success": True,
"output": str(out),
"stitch_count": len(stitches),
"bounds_mm": {
"width": round(rw / 10, 1),
"height": round(rh / 10, 1),
},
}
@mcp.tool()
def embroidery_send(
input_path: str,
machine_model: str,
usb_mount: str = "/media/usb",
disk_number: int = 0,
) -> dict:
"""Convert a design and write it to a USB stick for a specific embroidery machine.
Auto-detects the right format, writes the file to the Gotek USB stick
in the format the machine expects (e.g., DSKA0000.DST for Tajima).
Args:
input_path: Path to the design file (any supported format)
machine_model: Machine model to target (e.g., 'Tajima TME-612')
usb_mount: Mount point of the Gotek USB stick (default /media/usb)
disk_number: Disk image number 0-999, becomes DSKAXXXX
Returns:
dict with success, output file path, Gotek filename, stitch count
"""
import pyembroidery as pe
inp = Path(input_path).expanduser().resolve()
usb = Path(usb_mount).expanduser().resolve()
if not inp.exists():
return {"error": f"Input file not found: {inp}"}
if not usb.exists():
return {"error": f"USB mount not found: {usb}. Is the Gotek stick plugged in?"}
# Match machine
match = embroidery_match_machine(machine_model)
if not match.get("matched"):
return {"error": f"Unknown machine: {machine_model}. Add it to the vendor registry."}
fmt = match["formats"][0]
filename = f"DSKA{disk_number:04d}.{fmt.upper()}"
output_path = usb / filename
try:
pattern = pe.read(str(inp))
pe.write(pattern, str(output_path))
except Exception as e:
return {"error": f"Failed: {e}"}
stitch_count = (
pattern.count_stitches()
if hasattr(pattern, "count_stitches")
else 0
)
return {
"success": True,
"machine": machine_model,
"format": fmt,
"output_file": str(output_path),
"gotek_filename": filename,
"stitch_count": stitch_count,
"disk_number": disk_number,
"instructions": (
f"Insert USB stick into Gotek on {machine_model}. "
f"Select disk {disk_number:04d}. Load design. Press START."
),
}
@mcp.tool()
def embroidery_info(input_path: str) -> dict:
"""Get detailed info about an embroidery file without converting it.
Returns format, stitch count, color count, dimensions, estimated time.
Works on any supported format.
"""
import pyembroidery as pe
inp = Path(input_path).expanduser().resolve()
if not inp.exists():
return {"error": f"Input file not found: {inp}"}
try:
pattern = pe.read(str(inp))
except Exception as e:
return {"error": f"Failed to read {inp}: {e}"}
stitches = getattr(pattern, "stitches", [])
stitch_count = len(stitches)
# Count color changes
import pyembroidery as pe
color_count = 1 + sum(1 for s in stitches if len(s) >= 3 and s[2] == pe.COLOR_CHANGE)
# Bounds
xs = [s[0] for s in stitches if isinstance(s, (list, tuple)) and len(s) >= 2]
ys = [s[1] for s in stitches if isinstance(s, (list, tuple)) and len(s) >= 2]
return {
"success": True,
"path": str(inp),
"format": inp.suffix.upper().lstrip("."),
"stitch_count": stitch_count,
"color_count": color_count,
"width_mm": round((max(xs) - min(xs)) / 10, 1) if xs else 0,
"height_mm": round((max(ys) - min(ys)) / 10, 1) if ys else 0,
"estimated_time_min": round(stitch_count / 500, 1) if stitch_count else 0,
}
# ── Main ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
mcp.run()