94 lines
5.3 KiB
TypeScript
94 lines
5.3 KiB
TypeScript
import type { AxiosInstance } from "axios";
|
|
import FormData from "form-data";
|
|
import { open, realpath, stat } from "node:fs/promises";
|
|
import { isAbsolute, relative, basename, sep } from "node:path";
|
|
|
|
const MAX_IMAGE_BYTES = 10 * 1024 * 1024;
|
|
const IMAGE_TYPES: Record<string, string> = {
|
|
jpg: "image/jpeg",
|
|
jpeg: "image/jpeg",
|
|
png: "image/png",
|
|
webp: "image/webp",
|
|
gif: "image/gif",
|
|
};
|
|
|
|
type ToolResult = { content: [{ type: "text"; text: string }]; isError?: true };
|
|
const result = (value: unknown, isError = false): ToolResult => ({
|
|
content: [{ type: "text", text: JSON.stringify(value) }],
|
|
...(isError ? { isError: true as const } : {}),
|
|
});
|
|
const failure = (code: string, message: string): ToolResult => result({ success: false, code, message }, true);
|
|
|
|
export const ASSET_TOOL_DEFINITIONS = [{
|
|
name: "upload_cms_asset",
|
|
description: "Upload one local JPEG, PNG, WebP or GIF image to the active backend Media Library, without attaching it to a product. Requires MCP_ASSET_ROOT (absolute directory). filePath must be an absolute path inside that directory; max 10 MiB. Returns url and website-relative path for ARTICLE_MEDIA or ARTICLE_GALLERY. Confirm the active target first; this creates a media record even before a page uses the image.",
|
|
inputSchema: {
|
|
type: "object",
|
|
properties: {
|
|
filePath: { type: "string", description: "Absolute path to an existing image inside MCP_ASSET_ROOT. No remote URLs, SVG, or base64." },
|
|
},
|
|
required: ["filePath"],
|
|
additionalProperties: false,
|
|
},
|
|
}] as const;
|
|
|
|
function detectedMime(buffer: Buffer): string | null {
|
|
if (buffer.length >= 3 && buffer[0] === 0xff && buffer[1] === 0xd8 && buffer[2] === 0xff) return "image/jpeg";
|
|
if (buffer.length >= 8 && buffer.subarray(0, 8).equals(Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]))) return "image/png";
|
|
if (buffer.length >= 12 && buffer.toString("ascii", 0, 4) === "RIFF" && buffer.toString("ascii", 8, 12) === "WEBP") return "image/webp";
|
|
if (buffer.length >= 6 && ["GIF87a", "GIF89a"].includes(buffer.toString("ascii", 0, 6))) return "image/gif";
|
|
return null;
|
|
}
|
|
|
|
export async function handleAssetTool(api: AxiosInstance, name: string, rawParams: unknown, assetRoot = process.env.MCP_ASSET_ROOT): Promise<ToolResult | null> {
|
|
if (name !== "upload_cms_asset") return null;
|
|
const params = rawParams && typeof rawParams === "object" && !Array.isArray(rawParams) ? rawParams as Record<string, unknown> : {};
|
|
if (typeof assetRoot !== "string" || !isAbsolute(assetRoot)) return failure("ASSET_ROOT_REQUIRED", "Set MCP_ASSET_ROOT to an absolute image staging directory and restart MCP.");
|
|
if (typeof params.filePath !== "string" || !isAbsolute(params.filePath)) return failure("INVALID_FILE_PATH", "filePath must be an absolute local path inside MCP_ASSET_ROOT.");
|
|
|
|
let buffer: Buffer;
|
|
let fileName: string;
|
|
let mime: string;
|
|
try {
|
|
const canonicalRoot = await realpath(assetRoot);
|
|
if (!(await stat(canonicalRoot)).isDirectory()) return failure("INVALID_ASSET_ROOT", "MCP_ASSET_ROOT must be a directory.");
|
|
const canonicalFile = await realpath(params.filePath);
|
|
const subpath = relative(canonicalRoot, canonicalFile);
|
|
if (!subpath || subpath === ".." || subpath.startsWith(`..${sep}`) || isAbsolute(subpath)) {
|
|
return failure("FILE_OUTSIDE_ASSET_ROOT", "Image is outside MCP_ASSET_ROOT.");
|
|
}
|
|
const fileInfo = await stat(canonicalFile);
|
|
if (!fileInfo.isFile()) return failure("INVALID_IMAGE_FILE", "filePath must refer to a regular file.");
|
|
if (fileInfo.size === 0 || fileInfo.size > MAX_IMAGE_BYTES) return failure("INVALID_IMAGE_SIZE", "Image must be nonempty and at most 10 MiB.");
|
|
fileName = basename(canonicalFile);
|
|
const extension = fileName.split(".").pop()?.toLowerCase() ?? "";
|
|
const expectedMime = IMAGE_TYPES[extension];
|
|
if (!expectedMime) return failure("INVALID_IMAGE_TYPE", "Only JPEG, PNG, WebP and GIF files are supported.");
|
|
const handle = await open(canonicalFile, "r");
|
|
try { buffer = await handle.readFile(); } finally { await handle.close(); }
|
|
if (buffer.length === 0 || buffer.length > MAX_IMAGE_BYTES) return failure("INVALID_IMAGE_SIZE", "Image must be nonempty and at most 10 MiB.");
|
|
mime = detectedMime(buffer) ?? "";
|
|
if (mime !== expectedMime) return failure("INVALID_IMAGE_TYPE", "Image content does not match its supported file extension.");
|
|
} catch {
|
|
return failure("IMAGE_NOT_READABLE", "Image or MCP_ASSET_ROOT is missing or cannot be read.");
|
|
}
|
|
|
|
try {
|
|
const form = new FormData();
|
|
form.append("file", buffer, { filename: fileName, contentType: mime, knownLength: buffer.length });
|
|
const response = await api.post("/upload", form, {
|
|
headers: form.getHeaders(),
|
|
maxBodyLength: MAX_IMAGE_BYTES + 1024 * 1024,
|
|
maxContentLength: 1024 * 1024,
|
|
});
|
|
const data = response.data;
|
|
if (data?.success !== true || typeof data.url !== "string" || typeof data.path !== "string" || !data.path.startsWith("/uploads/")) {
|
|
return failure("INVALID_UPLOAD_RESPONSE", "Backend upload response did not include a usable media URL and path.");
|
|
}
|
|
return result({ success: true, url: data.url, path: data.path, mimeType: mime, sizeBytes: buffer.length });
|
|
} catch (error: any) {
|
|
const status = error?.response?.status;
|
|
return failure("UPLOAD_FAILED", `Backend upload failed${typeof status === "number" ? ` (HTTP ${status})` : ""}.`);
|
|
}
|
|
}
|