lightpanda-mcp.ts

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { spawn } from "node:child_process";
import puppeteer, { Browser, BrowserContext, Page } from "puppeteer-core";
import { fileURLToPath } from "url";
import { dirname } from "path";
import { setTimeout } from "node:timers/promises";

const LIGHTPANDA_PORT = 9222;
const DEV = false;
const c = console;

const __dirname = dirname(fileURLToPath(import.meta.url));

const lightpanda = spawn("./lightpanda", ["serve", "--port", `${LIGHTPANDA_PORT}`], { cwd: __dirname });

lightpanda.stdout.on("data", data => {
	if (DEV) {
		c.log(`[panda] stdout: ${data}`);
	}
});
lightpanda.stderr.on("data", data => {
	if (DEV) {
		c.error(`[panda] stderr: ${data}`);
	}
});
lightpanda.on("close", code => {
	if (DEV) {
		c.log(`[panda] child process exited with code ${code}`);
	}
});
const cleanupLp = () => {
	lightpanda.kill();
	process.exit();
};
process.on("exit", cleanupLp);
process.on("SIGINT", cleanupLp);
process.on("SIGTERM", cleanupLp);
process.on("SIGQUIT", cleanupLp);

let browser: Browser | undefined;
let context: BrowserContext | undefined;
let page: Page | undefined;

// Create server instance
const server = new McpServer({
	name:         "lightpanda",
	version:      "1.0.0",
	capabilities: {
		tools: {},
	},
});

server.tool(
	"navigate",
	"Navigate to a URL and return the full raw HTML",
	{
		url: z.string().url().describe("URL to navigate to"),
	},
	async ({ url }) => {
		if (!browser || !browser.connected) {
			browser = await puppeteer.connect({
				browserWSEndpoint: `ws://127.0.0.1:${LIGHTPANDA_PORT}`,
			});
			context = undefined;
		}
		if (!context || !context.browser().connected) {
			context = await browser.createBrowserContext();
		}
		try {
			page = await context.newPage();
			await page.goto(url, { waitUntil: "load" });
			const html = await page.content();
			return {
				content: [
					{
						type: "text",
						text: html,
					},
				],
			};
		} catch (err) {
			return {
				content: [
					{
						type: "text",
						text: `Error: ${err}`,
					},
				],
			};
		}
	}
);

server.tool("get_all_links", "Get all links from the webpage", {}, async () => {
	if (!page) {
		return {
			isError: true,
			content: [
				{
					type: "text",
					text: `The browser does not have a page currently open, first navigate to a page`,
				},
			],
		};
	}
	try {
		const links = await page.evaluate(() => {
			return Array.from(document.querySelectorAll("a")).map(row => {
				return row.getAttribute("href");
			});
		});
		return {
			content: [
				{
					type: "text",
					text: links.join("\n"),
				},
			],
		};
	} catch (err) {
		return {
			content: [
				{
					type: "text",
					text: `Error: ${err}`,
				},
			],
		};
	}
});

server.tool(
	"read_text",
	"Read inner text of a specific DOM element inside the webpage",
	{
		selector: z.string().describe("CSS selector for element to read"),
	},
	async ({ selector }) => {
		if (!page) {
			return {
				isError: true,
				content: [
					{
						type: "text",
						text: `The browser does not have a page currently open, first navigate to a page`,
					},
				],
			};
		}
		try {
			const content = await page.evaluate(() => {
				return document.querySelector(selector)?.textContent;
			});
			return {
				content: [
					{
						type: "text",
						text: content ?? "no content found",
					},
				],
			};
		} catch (err) {
			return {
				content: [
					{
						type: "text",
						text: `Error: ${err}`,
					},
				],
			};
		}
	}
);

async function main() {
	await setTimeout(500);
	browser = await puppeteer.connect({
		browserWSEndpoint: `ws://127.0.0.1:${LIGHTPANDA_PORT}`,
	});

	const transport = new StdioServerTransport();
	await server.connect(transport);
}

main().catch(error => {
	console.error("Fatal error in main():", error);
	process.exit(1);
});