From f66f0c191da8398c8b339d4eeb402d882c7e5a09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adri=C3=A0=20Arrufat?= Date: Sun, 12 Jul 2026 14:57:24 +0200 Subject: [PATCH] mcp: add HTTP transport and multi-session support Introduces an HTTP transport option to serve multipleagents from a single process. Each connection is routed to its own isolated browsing session using the `Mcp-Session-Id` header. Also adds new session management tools (`session_new`, `session_list`, `session_close`) and refactors the MCP server to support multiple V8 isolates with parking. --- README.md | 22 +++ src/Config.zig | 2 + src/help.zon | 17 +- src/main.zig | 21 +++ src/mcp.zig | 1 + src/mcp/HttpServer.zig | 347 +++++++++++++++++++++++++++++++++++++++++ src/mcp/Server.zig | 200 ++++++++++++++++++++---- src/mcp/Transport.zig | 7 + src/mcp/resources.zig | 2 +- src/mcp/tools.zig | 207 +++++++++++++++++++++--- 10 files changed, 771 insertions(+), 55 deletions(-) create mode 100644 src/mcp/HttpServer.zig diff --git a/README.md b/README.md index 806f24d0c..9ea1552c8 100644 --- a/README.md +++ b/README.md @@ -200,6 +200,28 @@ Add to your MCP configuration: } ``` +#### HTTP transport and independent sessions + +For serving several agents from one process, start the MCP server over HTTP +instead of stdio by giving it a port (add `--host 0.0.0.0` to accept remote +connections): + +```bash +lightpanda mcp --port 9223 +``` + +Clients POST JSON-RPC to `http://host:9223/mcp`. Each connection is routed to +its own **browsing session** — its own page, cookies and memory — so agents no +longer clobber each other's page: + +- A client that `initialize`s without an `Mcp-Session-Id` header is assigned a + fresh session; the id comes back in the response's `Mcp-Session-Id` header. + Send it on subsequent requests to stay on that session (**isolation**). +- Two agents that send the **same** `Mcp-Session-Id` share one browsing context + (**sharing** — e.g. a workflow where several agents work the same page). +- The `session_new`, `session_list` and `session_close` tools manage sessions + explicitly. Sending `DELETE /mcp` with an `Mcp-Session-Id` closes that session. + [Read full documentation](https://lightpanda.io/docs/open-source/guides/mcp-server) A skill is available in [lightpanda-io/agent-skill](https://github.com/lightpanda-io/agent-skill). diff --git a/src/Config.zig b/src/Config.zig index cc825192b..4adc6949f 100644 --- a/src/Config.zig +++ b/src/Config.zig @@ -241,6 +241,8 @@ const Commands = cli.Builder(.{ .{ .name = "mcp", .options = .{ + .{ .name = "port", .type = ?u16 }, + .{ .name = "host", .type = []const u8, .default = "127.0.0.1" }, .{ .name = "cdp_port", .type = ?u16 }, }, .shared_options = CommonOptions, diff --git a/src/help.zon b/src/help.zon index 1163f001c..a8db5cee9 100644 --- a/src/help.zon +++ b/src/help.zon @@ -7,7 +7,7 @@ \\ agent starts an interactive AI agent that can browse the web \\ fetch fetches the specified URL \\ help displays this message - \\ mcp starts an MCP (Model Context Protocol) server over stdio + \\ mcp starts an MCP (Model Context Protocol) server (stdio or HTTP) \\ serve starts a WebSocket CDP server \\ version displays the version of {0s} \\ @@ -120,15 +120,28 @@ .mcp = \\usage: {0s} mcp [OPTIONS] [COMMON_OPTIONS] \\ - \\Starts an MCP (Model Context Protocol) server over stdio. + \\Starts an MCP (Model Context Protocol) server. Defaults to stdio; pass + \\--port to serve over HTTP with an independent browsing session per client. \\ \\options: + \\ --cdp-port + \\ Also run a CDP (WebSocket) server on this port. Cannot be + \\ combined with --port (they share one network listener). + \\ Defaults to disabled. \\ --cookie \\ Path to a JSON file to load cookies from (read-only). \\ Defaults to no cookie loading. \\ --cookie-jar \\ Path to a JSON file to save cookies to on exit (write-only). \\ Defaults to no cookie saving. + \\ --host + \\ Host to bind when --port is set (e.g. 0.0.0.0 for remote access). + \\ Defaults to "127.0.0.1". + \\ --port + \\ Serve MCP over HTTP on this port instead of stdio. Clients POST + \\ JSON-RPC to /mcp; route requests to a session with the + \\ Mcp-Session-Id header. + \\ Defaults to stdio (no HTTP server). , .agent = \\agent command diff --git a/src/main.zig b/src/main.zig index 4799eb3ef..f54e16cff 100644 --- a/src/main.zig +++ b/src/main.zig @@ -207,6 +207,27 @@ fn run(allocator: Allocator, main_arena: Allocator) !void { log.opts.format = .logfmt; + // --port serves MCP over HTTP instead of stdio. It and --cdp-port + // both need app.network's single listener, so they can't combine. + if (opts.port) |port| { + if (opts.cdp_port != null) { + log.fatal(.mcp, "port conflicts with cdp-port", .{ .hint = "both need the single network listener" }); + return; + } + const address = std.net.Address.parseIp(opts.host, port) catch |err| { + log.fatal(.mcp, "invalid address", .{ .err = err, .host = opts.host, .port = port }); + return; + }; + const http_server = try lp.mcp.HttpServer.init(allocator, app); + defer http_server.deinit(); + // Shutdown rides the already-registered Network.stop handler: + // a signal stops the accept loop, run() returns, deinit joins. + http_server.run(address) catch |err| { + log.fatal(.mcp, "mcp http error", .{ .err = err }); + }; + return; + } + var cdp_server: ?*lp.Server = null; if (opts.cdp_port) |port| { const address = std.net.Address.parseIp("127.0.0.1", port) catch |err| { diff --git a/src/mcp.zig b/src/mcp.zig index de013bc93..328a8153f 100644 --- a/src/mcp.zig +++ b/src/mcp.zig @@ -4,6 +4,7 @@ pub const protocol = @import("mcp/protocol.zig"); pub const Version = protocol.Version; pub const router = @import("mcp/router.zig"); pub const Server = @import("mcp/Server.zig"); +pub const HttpServer = @import("mcp/HttpServer.zig"); test { std.testing.refAllDecls(@This()); diff --git a/src/mcp/HttpServer.zig b/src/mcp/HttpServer.zig new file mode 100644 index 000000000..e1ff3d661 --- /dev/null +++ b/src/mcp/HttpServer.zig @@ -0,0 +1,347 @@ +// Copyright (C) 2023-2026 Lightpanda (Selecy SAS) +// +// Francis Bouvier +// Pierre Tachoire +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +//! HTTP (MCP "Streamable HTTP") transport for the browser-tools MCP server. +//! +//! Lets many clients drive one process, each on its own browsing session. +//! Threading rule: V8 isolates are thread-affine, so ALL browser work — every +//! session's Browser/Session — lives on a single worker thread that owns the +//! `Server`. Connection threads never touch a browser; they parse HTTP, +//! marshal a `Job` to the worker over a queue, block on its completion, then +//! write the response. Session routing follows the `Mcp-Session-Id` header: +//! an `initialize` without one mints a fresh session (isolation by default); +//! reusing an id joins that session (sharing on purpose). + +const std = @import("std"); +const lp = @import("lightpanda"); + +const App = @import("../App.zig"); +const Server = @import("Server.zig"); +const router = @import("router.zig"); + +const log = lp.log; +const posix = std.posix; + +const HttpServer = @This(); + +const ns_per_ms = std.time.ns_per_ms; + +/// Cap on a single JSON-RPC request body. Generous: agent tool payloads +/// (e.g. a `save` script) can be large, but this bounds a hostile client. +const max_request_bytes = 16 * 1024 * 1024; + +/// One unit of work handed from a connection thread to the browser worker. +/// Allocated on the connection thread's stack — safe because that thread +/// blocks on `done` for the whole time the worker reads `body`/`session_id` +/// and writes `out`. +const Job = struct { + kind: Kind, + body: []const u8, + session_id: ?[]const u8, + /// Where the worker writes the response. The connection thread owns the + /// backing buffer and reads it back once `done` fires. + out: *std.Io.Writer, + + /// Session id the worker actually routed to; echoed back as + /// `Mcp-Session-Id`. Stored in a fixed buffer (not a worker arena) so it + /// outlives the worker moving on to the next job. + assigned_buf: [128]u8 = undefined, + assigned_len: usize = 0, + + done: std.Thread.ResetEvent = .{}, + next: ?*Job = null, + + const Kind = enum { rpc, close }; + + fn assigned(self: *const Job) []const u8 { + return self.assigned_buf[0..self.assigned_len]; + } + + fn setAssigned(self: *Job, id: []const u8) void { + const n = @min(id.len, self.assigned_buf.len); + @memcpy(self.assigned_buf[0..n], id[0..n]); + self.assigned_len = n; + } +}; + +const Queue = struct { + mutex: std.Thread.Mutex = .{}, + cond: std.Thread.Condition = .{}, + head: ?*Job = null, + tail: ?*Job = null, + + fn push(self: *Queue, job: *Job) void { + self.mutex.lock(); + defer self.mutex.unlock(); + job.next = null; + if (self.tail) |t| t.next = job else self.head = job; + self.tail = job; + self.cond.signal(); + } + + /// Pop the next job, waiting at most `timeout_ms`. Returns null on timeout + /// (or spurious wakeup) so the worker can pump idle sessions and retry. + fn pop(self: *Queue, timeout_ms: u64) ?*Job { + self.mutex.lock(); + defer self.mutex.unlock(); + if (self.head == null) { + self.cond.timedWait(&self.mutex, timeout_ms * ns_per_ms) catch {}; + } + const job = self.head orelse return null; + self.head = job.next; + if (self.head == null) self.tail = null; + return job; + } +}; + +allocator: std.mem.Allocator, +app: *App, + +queue: Queue = .{}, +stopping: std.atomic.Value(bool) = .init(false), + +// The worker owns `server`; other threads must not touch it. +worker_thread: std.Thread = undefined, +worker_ready: std.Thread.ResetEvent = .{}, +worker_ok: bool = false, + +/// Create the server and start its browser worker thread. Returns once the +/// worker's default session is up (or errors if it failed to initialize). +pub fn init(allocator: std.mem.Allocator, app: *App) !*HttpServer { + const self = try allocator.create(HttpServer); + errdefer allocator.destroy(self); + self.* = .{ + .allocator = allocator, + .app = app, + }; + + self.worker_thread = try std.Thread.spawn(.{}, worker, .{self}); + self.worker_ready.wait(); + if (!self.worker_ok) { + self.worker_thread.join(); + return error.WorkerInitFailed; + } + return self; +} + +pub fn deinit(self: *HttpServer) void { + self.stopping.store(true, .release); + self.queue.cond.signal(); // nudge the worker off its idle wait + self.worker_thread.join(); + self.allocator.destroy(self); +} + +/// Accept MCP-over-HTTP connections until the network loop is stopped +/// (e.g. by the signal handler). Reuses the shared accept infrastructure; +/// blocks the calling thread in `Network.run`. +pub fn run(self: *HttpServer, address: std.net.Address) !void { + var bound = address; + try self.app.network.bind(&bound, self, onAccept); + log.note(.mcp, "mcp http server running", .{ .address = bound }); + self.app.network.run(); +} + +/// Network hands us a nonblocking accepted socket; each connection is served +/// by its own thread doing blocking IO, so we clear O_NONBLOCK first. +fn onAccept(ctx: *anyopaque, socket: posix.socket_t) void { + const self: *HttpServer = @ptrCast(@alignCast(ctx)); + + const flags = posix.fcntl(socket, posix.F.GETFL, 0) catch { + posix.close(socket); + return; + }; + _ = posix.fcntl(socket, posix.F.SETFL, flags & ~@as(u32, @bitCast(posix.O{ .NONBLOCK = true }))) catch { + posix.close(socket); + return; + }; + + const thread = std.Thread.spawn(.{}, handleConn, .{ self, socket }) catch |err| { + log.warn(.mcp, "mcp spawn", .{ .err = err }); + posix.close(socket); + return; + }; + thread.detach(); +} + +fn worker(self: *HttpServer) void { + var placeholder: std.Io.Writer.Allocating = .init(self.allocator); + defer placeholder.deinit(); + + const server = Server.init(self.allocator, self.app, &placeholder.writer) catch |err| { + log.err(.mcp, "mcp http server init", .{ .err = err }); + self.worker_ready.set(); + return; + }; + defer server.deinit(); + + server.enableIsolateParking(); + + self.worker_ok = true; + self.worker_ready.set(); + + var arena: std.heap.ArenaAllocator = .init(self.allocator); + defer arena.deinit(); + + while (!self.stopping.load(.acquire)) { + const wait_ms = server.idle(); + const job = self.queue.pop(wait_ms) orelse continue; + _ = arena.reset(.retain_capacity); + process(server, arena.allocator(), job); + job.done.set(); + } +} + +fn process(server: *Server, arena: std.mem.Allocator, job: *Job) void { + server.transport.retarget(job.out); + + if (job.kind == .close) { + if (job.session_id) |sid| _ = server.closeSession(sid); + return; + } + + const chosen = resolveSession(server, arena, job) catch |err| { + log.err(.mcp, "mcp session routing", .{ .err = err }); + return; + }; + job.setAssigned(chosen); + + router.handleMessage(server, arena, job.body) catch |err| { + log.err(.mcp, "mcp handle", .{ .err = err }); + }; +} + +/// Decide which session a request targets and make it active. An explicit +/// `Mcp-Session-Id` wins; otherwise `initialize` mints a new session and +/// everything else falls back to the default. +fn resolveSession(server: *Server, arena: std.mem.Allocator, job: *Job) ![]const u8 { + if (job.session_id) |sid| { + if (sid.len > 0) { + _ = try server.useSession(sid); + return sid; + } + } + + if (isInitialize(arena, job.body)) { + const sid = try server.nextSessionId(arena); + _ = try server.useSession(sid); + return sid; + } + + _ = try server.useSession(null); + return Server.default_session_id; +} + +fn isInitialize(arena: std.mem.Allocator, body: []const u8) bool { + const Peek = struct { method: ?[]const u8 = null }; + const peek = std.json.parseFromSliceLeaky(Peek, arena, body, .{ .ignore_unknown_fields = true }) catch return false; + const method = peek.method orelse return false; + return std.mem.eql(u8, method, "initialize"); +} + +fn handleConn(self: *HttpServer, socket: posix.socket_t) void { + const stream: std.net.Stream = .{ .handle = socket }; + defer stream.close(); + + var recv_buf: [16 * 1024]u8 = undefined; + var send_buf: [16 * 1024]u8 = undefined; + var stream_reader = stream.reader(&recv_buf); + var stream_writer = stream.writer(&send_buf); + var http_server = std.http.Server.init(stream_reader.interface(), &stream_writer.interface); + + var arena: std.heap.ArenaAllocator = .init(self.allocator); + defer arena.deinit(); + // Reused across requests (served serially), not reallocated per request. + var out: std.Io.Writer.Allocating = .init(self.allocator); + defer out.deinit(); + + while (!self.stopping.load(.acquire)) { + var request = http_server.receiveHead() catch return; // peer closed or bad head + _ = arena.reset(.retain_capacity); + out.clearRetainingCapacity(); + self.serve(&out.writer, arena.allocator(), &request) catch return; + if (!request.head.keep_alive) return; + } +} + +/// Handle one request: marshal it to the browser worker and write the reply. +/// MCP Streamable HTTP — POST carries a JSON-RPC message; DELETE closes the +/// session named by `Mcp-Session-Id`. std.http.Server owns the framing. +fn serve(self: *HttpServer, out: *std.Io.Writer, arena: std.mem.Allocator, request: *std.http.Server.Request) !void { + const method = request.head.method; + if (method != .POST and method != .DELETE) { + return request.respond("", .{ .status = .method_not_allowed, .keep_alive = false }); + } + if (request.head.expect != null) { + return request.respond("", .{ .status = .expectation_failed, .keep_alive = false }); + } + + // Read the session header and keep_alive before the body reader + // invalidates the head's string memory. + const session_id = try sessionHeader(arena, request); + const keep_alive = request.head.keep_alive; + + var body_buf: [8 * 1024]u8 = undefined; + const body = request.readerExpectNone(&body_buf).allocRemaining(arena, .limited(max_request_bytes)) catch { + return request.respond("", .{ .status = .payload_too_large, .keep_alive = false }); + }; + + var job: Job = .{ + .kind = if (method == .DELETE) .close else .rpc, + .body = body, + .session_id = session_id, + .out = out, + }; + self.queue.push(&job); + job.done.wait(); + + const resp = out.buffered(); + var headers: [2]std.http.Header = undefined; + var n: usize = 0; + headers[n] = .{ .name = "content-type", .value = "application/json" }; + n += 1; + if (job.assigned_len > 0) { + headers[n] = .{ .name = "mcp-session-id", .value = job.assigned() }; + n += 1; + } + return request.respond(resp, .{ + // An empty body means a notification (or a close): 202, no content. + .status = if (resp.len == 0) .accepted else .ok, + .keep_alive = keep_alive, + .extra_headers = headers[0..n], + }); +} + +/// Duplicate the `Mcp-Session-Id` request header into `arena`, or null. +fn sessionHeader(arena: std.mem.Allocator, request: *std.http.Server.Request) !?[]const u8 { + var it = request.iterateHeaders(); + while (it.next()) |header| { + if (std.ascii.eqlIgnoreCase(header.name, "mcp-session-id")) { + return try arena.dupe(u8, header.value); + } + } + return null; +} + +test "HttpServer - initialize is detected for session minting" { + var arena: std.heap.ArenaAllocator = .init(std.testing.allocator); + defer arena.deinit(); + const aa = arena.allocator(); + try std.testing.expect(isInitialize(aa, "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\"}")); + try std.testing.expect(!isInitialize(aa, "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\"}")); + try std.testing.expect(!isInitialize(aa, "not json")); +} diff --git a/src/mcp/Server.zig b/src/mcp/Server.zig index 9861a4f16..0455f7d26 100644 --- a/src/mcp/Server.zig +++ b/src/mcp/Server.zig @@ -13,61 +13,194 @@ const CDPNode = @import("../cdp/Node.zig"); const Self = @This(); +/// Session every un-scoped request lands on. Over stdio there is only ever +/// this one; the HTTP transport routes to it when a client sends no +/// `Mcp-Session-Id`. +pub const default_session_id = "default"; + +/// One isolated browsing context. Each owns its own V8 isolate (via +/// `Browser`), so two agents driving different sessions never touch the same +/// page. Heap-allocated and never moved after `init`: `Browser` registers +/// self-pointers (watchdog, http_client) that must stay stable. +pub const Session = struct { + id: []const u8, + browser: lp.Browser, + session: *lp.Session, + notification: *lp.Notification, + node_registry: CDPNode.Registry, + + fn isDefault(self: *const Session) bool { + return std.mem.eql(u8, self.id, default_session_id); + } +}; + allocator: std.mem.Allocator, app: *App, -notification: *lp.Notification, -browser: lp.Browser, -session: *lp.Session, -node_registry: CDPNode.Registry, - +sessions: std.StringHashMapUnmanaged(*Session) = .empty, +/// Monotonic counter backing auto-generated session ids (`s1`, `s2`, …). +session_seq: u32 = 0, +/// When several sessions (each its own V8 isolate) share one thread, V8's +/// "current isolate" is a per-thread stack, so an isolate must be *entered* +/// around any use of it and left un-entered otherwise. The HTTP transport +/// sets this; stdio (one isolate, permanently entered by `Env`) leaves it +/// false and keeps its historical behavior. See `enterIsolate`/`exitIsolate`. +park_isolates: bool = false, +/// The session the request currently being handled targets. Safe as a single +/// field because every request is dispatched on one thread, one at a time; +/// the transport sets it (via `useSession`) before each dispatch. Tools and +/// resources read it rather than threading a session through every call. +active_session: *Session = undefined, transport: Transport, pub fn init(allocator: std.mem.Allocator, app: *App, writer: *std.io.Writer) !*Self { - const notification = try lp.Notification.init(allocator); - errdefer notification.deinit(); - const self = try allocator.create(Self); errdefer allocator.destroy(self); self.* = .{ .allocator = allocator, .app = app, - .browser = undefined, .transport = .init(allocator, writer), - .notification = notification, - .session = undefined, - .node_registry = CDPNode.Registry.init(allocator), }; + errdefer self.transport.deinit(); - try self.browser.init(app, .{}, null); - errdefer self.browser.deinit(); - - self.session = try self.browser.newSession(self.notification); - try self.session.enableConsoleCapture(); - - if (app.config.cookieFile()) |cookie_path| { - lp.cookies.loadFromFile(self.session, cookie_path); - } - + self.active_session = try self.createSession(default_session_id); return self; } pub fn deinit(self: *Self) void { - if (self.app.config.cookieJarFile()) |cookie_jar_path| { - lp.cookies.saveToFile(&self.session.cookie_jar, cookie_jar_path); - } + var it = self.sessions.valueIterator(); + while (it.next()) |entry| self.destroySession(entry.*); + self.sessions.deinit(self.allocator); - self.node_registry.deinit(); self.transport.deinit(); - self.browser.deinit(); - self.notification.deinit(); - self.allocator.destroy(self); } +/// Create the session named `id`, or return the existing one. The `id` is +/// duped, so the caller keeps ownership of its slice. +pub fn createSession(self: *Self, id: []const u8) !*Session { + if (self.sessions.get(id)) |existing| return existing; + + const owned_id = try self.allocator.dupe(u8, id); + errdefer self.allocator.free(owned_id); + + const entry = try self.allocator.create(Session); + errdefer self.allocator.destroy(entry); + + const notification = try lp.Notification.init(self.allocator); + errdefer notification.deinit(); + + entry.* = .{ + .id = owned_id, + .browser = undefined, + .session = undefined, + .notification = notification, + .node_registry = CDPNode.Registry.init(self.allocator), + }; + errdefer entry.node_registry.deinit(); + + try entry.browser.init(self.app, .{}, null); + errdefer entry.browser.deinit(); + + entry.session = try entry.browser.newSession(notification); + try entry.session.enableConsoleCapture(); + + // Only the default session is backed by the on-disk cookie file; named + // sessions start clean so agents stay isolated by default. + if (entry.isDefault()) { + if (self.app.config.cookieFile()) |cookie_path| { + lp.cookies.loadFromFile(entry.session, cookie_path); + } + } + + try self.sessions.put(self.allocator, owned_id, entry); + // Browser.init left the isolate entered; park it (see park_isolates). + self.exitIsolate(entry); + return entry; +} + +/// Switch to the multi-isolate discipline: park the default (which `Server.init` +/// left entered) and require every use to bracket with `enterIsolate`. The HTTP +/// transport calls this on its worker thread before serving anyone. +pub fn enableIsolateParking(self: *Self) void { + self.park_isolates = true; + self.exitIsolate(self.defaultSession()); +} + +/// Make `entry`'s isolate the current one for this thread. Must bracket any +/// use of its Browser/Session (dispatch, idle pumping, teardown). No-op under +/// stdio, where the single isolate is permanently current. +pub fn enterIsolate(self: *Self, entry: *Session) void { + if (self.park_isolates) entry.browser.env.isolate.enter(); +} + +pub fn exitIsolate(self: *Self, entry: *Session) void { + if (self.park_isolates) entry.browser.env.isolate.exit(); +} + +/// Tear down the session named `id`. Returns false if no such session, or if +/// it is the default (which lives for the whole process). +pub fn closeSession(self: *Self, id: []const u8) bool { + if (std.mem.eql(u8, id, default_session_id)) return false; + const entry = self.sessions.fetchRemove(id) orelse return false; + if (self.active_session == entry.value) self.active_session = self.defaultSession(); + self.destroySession(entry.value); + return true; +} + +fn destroySession(self: *Self, entry: *Session) void { + if (entry.isDefault()) { + if (self.app.config.cookieJarFile()) |cookie_jar_path| { + lp.cookies.saveToFile(&entry.session.cookie_jar, cookie_jar_path); + } + } + + // Re-enter so `Browser.deinit`'s `Env.deinit` exit stays balanced against + // a parked isolate (and operates on the current one). + self.enterIsolate(entry); + entry.node_registry.deinit(); + entry.browser.deinit(); + entry.notification.deinit(); + self.allocator.free(entry.id); + self.allocator.destroy(entry); +} + +/// The session an un-scoped (stdio, or header-less HTTP) request targets. +pub fn defaultSession(self: *Self) *Session { + return self.sessions.get(default_session_id).?; +} + +/// Point subsequent tool/resource dispatch at the session named `id`, creating +/// it on first use. A null or empty `id` selects the default. +pub fn useSession(self: *Self, id: ?[]const u8) !*Session { + const wanted = id orelse ""; + self.active_session = if (wanted.len == 0) self.defaultSession() else try self.createSession(wanted); + return self.active_session; +} + +/// A session id that is not currently in use, formatted into `arena`. +pub fn nextSessionId(self: *Self, arena: std.mem.Allocator) ![]const u8 { + while (true) { + self.session_seq += 1; + const candidate = try std.fmt.allocPrint(arena, "s{d}", .{self.session_seq}); + if (!self.sessions.contains(candidate)) return candidate; + } +} + +/// Pump every live session's pending transfers and return the shortest time +/// the caller may block before pumping again. See `Session.idleSlice`. pub fn idle(self: *Self) u31 { - return self.session.idleSlice(); + var wait: u31 = std.math.maxInt(u31); + var it = self.sessions.valueIterator(); + while (it.next()) |entry| { + // Pumping may resume JS (e.g. a completed script fetch), so it needs + // the session's isolate current. + self.enterIsolate(entry.*); + wait = @min(wait, entry.*.session.idleSlice()); + self.exitIsolate(entry.*); + } + return wait; } pub fn sendError(self: *Self, id: std.json.Value, code: protocol.ErrorCode, message: []const u8) !void { @@ -96,6 +229,10 @@ pub fn handleToolList(self: *Self, arena: std.mem.Allocator, req: protocol.Reque } pub fn handleToolCall(self: *Self, arena: std.mem.Allocator, req: protocol.Request) !void { + // Dispatch runs page JS, so enter the target isolate around it. + const entry = self.active_session; + self.enterIsolate(entry); + defer self.exitIsolate(entry); return tools.handleCall(self, arena, req); } @@ -104,6 +241,9 @@ pub fn handleResourceList(self: *Self, req: protocol.Request) !void { } pub fn handleResourceRead(self: *Self, arena: std.mem.Allocator, req: protocol.Request) !void { + const entry = self.active_session; + self.enterIsolate(entry); + defer self.exitIsolate(entry); return resources.handleRead(self, arena, req); } diff --git a/src/mcp/Transport.zig b/src/mcp/Transport.zig index c9df2ee1f..430a45be5 100644 --- a/src/mcp/Transport.zig +++ b/src/mcp/Transport.zig @@ -36,6 +36,13 @@ pub fn deinit(self: *Self) void { self.aw.deinit(); } +/// Point subsequent responses at a different sink. The HTTP transport uses +/// this to capture each request's response into its own buffer; safe because +/// the browser worker retargets and writes on a single thread. +pub fn retarget(self: *Self, writer: *std.io.Writer) void { + self.writer = writer; +} + pub fn sendResponse(self: *Self, response: anytype) !void { self.mutex.lock(); defer self.mutex.unlock(); diff --git a/src/mcp/resources.zig b/src/mcp/resources.zig index ed413c339..348a0c2b2 100644 --- a/src/mcp/resources.zig +++ b/src/mcp/resources.zig @@ -86,7 +86,7 @@ pub fn handleRead(server: *Server, arena: std.mem.Allocator, req: protocol.Reque return server.sendError(req_id, .InvalidRequest, "Resource not found"); }; - const frame = server.session.currentFrame() orelse { + const frame = server.active_session.session.currentFrame() orelse { return server.sendError(req_id, .FrameNotLoaded, "Page not loaded"); }; diff --git a/src/mcp/tools.zig b/src/mcp/tools.zig index f03cc35f5..a795f708b 100644 --- a/src/mcp/tools.zig +++ b/src/mcp/tools.zig @@ -36,12 +36,46 @@ const save_schema = browser_tools.minify( \\} ); +const session_new_schema = browser_tools.minify( + \\{ + \\ "type": "object", + \\ "properties": { + \\ "name": { "type": "string", "description": "Optional id for the new session. Omit to get an auto-generated one. Reusing an existing id returns that session (a way to share one browsing context between agents)." } + \\ } + \\} +); + +const session_id_schema = browser_tools.minify( + \\{ + \\ "type": "object", + \\ "properties": { + \\ "id": { "type": "string", "description": "The session id." } + \\ }, + \\ "required": ["id"] + \\} +); + const extra_tools = [_]McpTool{ .{ .name = "save", .description = "Save the session as a reusable Lightpanda agent script. You hold the conversation, so synthesize the `script` yourself — `const page = new Page(); await page.goto(url);` then call the builtins you used as tools (extract, click, fill, …) as methods on `page` with the same object arguments. Keep `$LP_*` placeholders; never inline a resolved secret.\n\n" ++ browser_tools.save_synthesis_prompt ++ "\n\n" ++ browser_tools.save_script_rules, .inputSchema = save_schema, }, + .{ + .name = "session_new", + .description = "Create a new isolated browser session (its own page, cookies and memory) and return its id. Use it to give a separate agent its own browsing context, or to obtain an id to share. Pass that id back as the `Mcp-Session-Id` header to route calls to it.", + .inputSchema = session_new_schema, + }, + .{ + .name = "session_list", + .description = "List the active browser sessions with their id and current URL. The `default` session always exists.", + .inputSchema = browser_tools.minify("{ \"type\": \"object\", \"properties\": {} }"), + }, + .{ + .name = "session_close", + .description = "Close a browser session, freeing its page and memory. The `default` session cannot be closed.", + .inputSchema = session_id_schema, + }, }; const all_tools = browser_tool_list ++ extra_tools; @@ -49,6 +83,9 @@ const all_tools = browser_tool_list ++ extra_tools; /// Tools that bypass the browser-tool dispatch and have their own handlers. const ExtraTool = enum { save, + session_new, + session_list, + session_close, }; pub fn handleList(server: *Server, arena: std.mem.Allocator, req: protocol.Request) !void { @@ -68,6 +105,9 @@ pub fn handleCall(server: *Server, arena: std.mem.Allocator, req: protocol.Reque if (std.meta.stringToEnum(ExtraTool, call_params.name)) |tool| { return switch (tool) { .save => handleSave(server, arena, id, call_params.arguments), + .session_new => handleSessionNew(server, arena, id, call_params.arguments), + .session_list => handleSessionList(server, arena, id), + .session_close => handleSessionClose(server, arena, id, call_params.arguments), }; } @@ -85,7 +125,8 @@ fn dispatchBrowserTool( return server.sendError(id, .MethodNotFound, "Tool not found"); }; - const result = browser_tools.call(arena, server.session, &server.node_registry, name, arguments) catch |err| { + const active = server.active_session; + const result = browser_tools.call(arena, active.session, &active.node_registry, name, arguments) catch |err| { // evaluate/extract surface failures in-band so the LLM can self-correct; // other tools' operational failures are protocol-level. if (surfacesErrorInBand(tool)) { @@ -138,6 +179,72 @@ fn handleSave(server: *Server, arena: std.mem.Allocator, id: std.json.Value, arg try sendToolResultText(server, id, msg, false); } +/// The session tools require the HTTP transport's parked-isolate discipline: +/// a second session means a second V8 isolate, only safe when isolates are +/// entered around use. Over stdio (one permanently-entered isolate) they are +/// all unsupported, kept uniform so clients see one consistent rule. +fn requireMultiSession(server: *Server, id: std.json.Value) !bool { + if (server.park_isolates) return true; + try sendToolResultText(server, id, "multiple sessions require the HTTP transport (start with --http-port)", true); + return false; +} + +fn handleSessionNew(server: *Server, arena: std.mem.Allocator, id: std.json.Value, arguments: ?std.json.Value) !void { + if (!try requireMultiSession(server, id)) return; + const Args = struct { name: ?[]const u8 = null }; + const args = browser_tools.parseArgsOrDefault(Args, arena, arguments) catch { + return server.sendError(id, .InvalidParams, "expected { name?: string }"); + }; + + const requested: ?[]const u8 = if (args.name) |n| (if (n.len > 0) n else null) else null; + const sid = requested orelse (server.nextSessionId(arena) catch + return sendErrorContent(server, id, "out of memory")); + + _ = server.createSession(sid) catch |err| + return sendErrorContent(server, id, @errorName(err)); + + return sendToolResultFmt(server, arena, id, "session {s}", .{sid}); +} + +fn handleSessionList(server: *Server, arena: std.mem.Allocator, id: std.json.Value) !void { + if (!try requireMultiSession(server, id)) return; + const Entry = struct { id: []const u8, url: ?[]const u8 }; + var list: std.ArrayList(Entry) = .empty; + + var it = server.sessions.valueIterator(); + while (it.next()) |entry| { + const url: ?[]const u8 = if (entry.*.session.currentFrame()) |frame| frame.url else null; + list.append(arena, .{ .id = entry.*.id, .url = url }) catch + return sendErrorContent(server, id, "out of memory"); + } + + const json = std.json.Stringify.valueAlloc(arena, list.items, .{ .emit_null_optional_fields = false }) catch + return sendErrorContent(server, id, "out of memory"); + try sendToolResultText(server, id, json, false); +} + +fn handleSessionClose(server: *Server, arena: std.mem.Allocator, id: std.json.Value, arguments: ?std.json.Value) !void { + if (!try requireMultiSession(server, id)) return; + const Args = struct { id: []const u8 }; + const args = browser_tools.parseArgs(Args, arena, arguments) catch { + return server.sendError(id, .InvalidParams, "expected { id: string }"); + }; + + if (std.mem.eql(u8, args.id, Server.default_session_id)) { + return sendErrorContent(server, id, "the default session cannot be closed"); + } + // Closing the session serving this very call would tear down the isolate + // mid-dispatch; require the client to be elsewhere first. + if (std.mem.eql(u8, args.id, server.active_session.id)) { + return sendErrorContent(server, id, "cannot close the session you are attached to"); + } + if (!server.closeSession(args.id)) { + return sendErrorContent(server, id, "no such session"); + } + + return sendToolResultFmt(server, arena, id, "closed session {s}", .{args.id}); +} + fn writeScript(path: []const u8, content: []const u8) !void { const file = try std.fs.cwd().createFile(path, .{ .truncate = true }); defer file.close(); @@ -154,6 +261,12 @@ fn sendErrorContent(server: *Server, id: std.json.Value, msg: []const u8) !void return sendToolResultText(server, id, msg, true); } +fn sendToolResultFmt(server: *Server, arena: std.mem.Allocator, id: std.json.Value, comptime fmt: []const u8, args: anytype) !void { + const msg = std.fmt.allocPrint(arena, fmt, args) catch + return sendErrorContent(server, id, "out of memory"); + return sendToolResultText(server, id, msg, false); +} + const router = @import("router.zig"); const testing = @import("../testing.zig"); @@ -861,11 +974,11 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked const server = try testLoadPage("http://localhost:9582/src/browser/tests/mcp_actions.html", &out.writer); defer server.deinit(); - const frame = server.session.currentFrame().?; + const frame = server.active_session.session.currentFrame().?; { const btn = frame.document.getElementById("btn", frame).?.asNode(); - const btn_id = (try server.node_registry.register(btn)).id; + const btn_id = (try server.active_session.node_registry.register(btn)).id; var btn_id_buf: [12]u8 = undefined; const btn_id_str = std.fmt.bufPrint(&btn_id_buf, "{d}", .{btn_id}) catch unreachable; const click_msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"click\",\"arguments\":{\"backendNodeId\":", btn_id_str, "}}}" }); @@ -877,7 +990,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const inp = frame.document.getElementById("inp", frame).?.asNode(); - const inp_id = (try server.node_registry.register(inp)).id; + const inp_id = (try server.active_session.node_registry.register(inp)).id; var inp_id_buf: [12]u8 = undefined; const inp_id_str = std.fmt.bufPrint(&inp_id_buf, "{d}", .{inp_id}) catch unreachable; const fill_msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/call\",\"params\":{\"name\":\"fill\",\"arguments\":{\"backendNodeId\":", inp_id_str, ",\"value\":\"hello\"}}}" }); @@ -889,7 +1002,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const sel = frame.document.getElementById("sel", frame).?.asNode(); - const sel_id = (try server.node_registry.register(sel)).id; + const sel_id = (try server.active_session.node_registry.register(sel)).id; var sel_id_buf: [12]u8 = undefined; const sel_id_str = std.fmt.bufPrint(&sel_id_buf, "{d}", .{sel_id}) catch unreachable; const fill_sel_msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"tools/call\",\"params\":{\"name\":\"fill\",\"arguments\":{\"backendNodeId\":", sel_id_str, ",\"value\":\"opt2\"}}}" }); @@ -901,7 +1014,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const scrollbox = frame.document.getElementById("scrollbox", frame).?.asNode(); - const scrollbox_id = (try server.node_registry.register(scrollbox)).id; + const scrollbox_id = (try server.active_session.node_registry.register(scrollbox)).id; var scroll_id_buf: [12]u8 = undefined; const scroll_id_str = std.fmt.bufPrint(&scroll_id_buf, "{d}", .{scrollbox_id}) catch unreachable; const scroll_msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":4,\"method\":\"tools/call\",\"params\":{\"name\":\"scroll\",\"arguments\":{\"backendNodeId\":", scroll_id_str, ",\"y\":50}}}" }); @@ -912,7 +1025,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const el = frame.document.getElementById("hoverTarget", frame).?.asNode(); - const el_id = (try server.node_registry.register(el)).id; + const el_id = (try server.active_session.node_registry.register(el)).id; var id_buf: [12]u8 = undefined; const id_str = std.fmt.bufPrint(&id_buf, "{d}", .{el_id}) catch unreachable; const msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":5,\"method\":\"tools/call\",\"params\":{\"name\":\"hover\",\"arguments\":{\"backendNodeId\":", id_str, "}}}" }); @@ -923,7 +1036,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const el = frame.document.getElementById("keyTarget", frame).?.asNode(); - const el_id = (try server.node_registry.register(el)).id; + const el_id = (try server.active_session.node_registry.register(el)).id; var id_buf: [12]u8 = undefined; const id_str = std.fmt.bufPrint(&id_buf, "{d}", .{el_id}) catch unreachable; const msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":6,\"method\":\"tools/call\",\"params\":{\"name\":\"press\",\"arguments\":{\"key\":\"Enter\",\"backendNodeId\":", id_str, "}}}" }); @@ -934,7 +1047,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const el = frame.document.getElementById("sel2", frame).?.asNode(); - const el_id = (try server.node_registry.register(el)).id; + const el_id = (try server.active_session.node_registry.register(el)).id; var id_buf: [12]u8 = undefined; const id_str = std.fmt.bufPrint(&id_buf, "{d}", .{el_id}) catch unreachable; const msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":7,\"method\":\"tools/call\",\"params\":{\"name\":\"selectOption\",\"arguments\":{\"backendNodeId\":", id_str, ",\"value\":\"b\"}}}" }); @@ -945,7 +1058,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const el = frame.document.getElementById("chk", frame).?.asNode(); - const el_id = (try server.node_registry.register(el)).id; + const el_id = (try server.active_session.node_registry.register(el)).id; var id_buf: [12]u8 = undefined; const id_str = std.fmt.bufPrint(&id_buf, "{d}", .{el_id}) catch unreachable; const msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":8,\"method\":\"tools/call\",\"params\":{\"name\":\"setChecked\",\"arguments\":{\"backendNodeId\":", id_str, ",\"checked\":true}}}" }); @@ -956,7 +1069,7 @@ test "MCP - Actions: click, fill, scroll, hover, press, selectOption, setChecked { const el = frame.document.getElementById("rad", frame).?.asNode(); - const el_id = (try server.node_registry.register(el)).id; + const el_id = (try server.active_session.node_registry.register(el)).id; var id_buf: [12]u8 = undefined; const id_str = std.fmt.bufPrint(&id_buf, "{d}", .{el_id}) catch unreachable; const msg = try std.mem.concat(aa, u8, &.{ "{\"jsonrpc\":\"2.0\",\"id\":9,\"method\":\"tools/call\",\"params\":{\"name\":\"setChecked\",\"arguments\":{\"backendNodeId\":", id_str, ",\"checked\":true}}}" }); @@ -999,10 +1112,10 @@ test "MCP - click that navigates clears node registry" { const server = try testLoadPage("http://localhost:9582/src/browser/tests/mcp_nav.html", &out.writer); defer server.deinit(); - const before_frame = server.session.currentFrame().?; + const before_frame = server.active_session.session.currentFrame().?; const link = before_frame.document.getElementById("navlink", before_frame).?.asNode(); - const link_id = (try server.node_registry.register(link)).id; - try testing.expect(server.node_registry.lookup_by_id.contains(link_id)); + const link_id = (try server.active_session.node_registry.register(link)).id; + try testing.expect(server.active_session.node_registry.lookup_by_id.contains(link_id)); var id_buf: [12]u8 = undefined; const id_str = std.fmt.bufPrint(&id_buf, "{d}", .{link_id}) catch unreachable; @@ -1013,8 +1126,8 @@ test "MCP - click that navigates clears node registry" { }); try router.handleMessage(server, aa, click_msg); - try testing.expect(server.session.currentFrame().? != before_frame); - try testing.expect(!server.node_registry.lookup_by_id.contains(link_id)); + try testing.expect(server.active_session.session.currentFrame().? != before_frame); + try testing.expect(!server.active_session.node_registry.lookup_by_id.contains(link_id)); } test "MCP - Actions by selector: hover, selectOption, setChecked" { @@ -1026,7 +1139,7 @@ test "MCP - Actions by selector: hover, selectOption, setChecked" { defer server.deinit(); // Single-page test: reach straight into the live page. - const page = server.session.pages.items[0]; + const page = server.active_session.session.pages.items[0]; { const msg = @@ -1291,8 +1404,8 @@ test "MCP - getCookies: defaults to current page, url filter, all flag" { const server = try testLoadPage("http://localhost:9582/src/browser/tests/mcp_press_form.htm", &out.writer); defer server.deinit(); - try server.session.cookie_jar.populateFromResponse("http://localhost:9582", "session=abc; Path=/"); - try server.session.cookie_jar.populateFromResponse("http://other.test/", "tracking=xyz; Path=/"); + try server.active_session.session.cookie_jar.populateFromResponse("http://localhost:9582", "session=abc; Path=/"); + try server.active_session.session.cookie_jar.populateFromResponse("http://other.test/", "tracking=xyz; Path=/"); const default_msg = \\{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"getCookies"}} @@ -1331,7 +1444,7 @@ test "MCP - getCookies without a loaded page refuses instead of dumping the jar" var server = try Server.init(testing.allocator, testing.test_app, &out.writer); defer server.deinit(); - try server.session.cookie_jar.populateFromResponse("http://example.com/", "session=abc; Path=/"); + try server.active_session.session.cookie_jar.populateFromResponse("http://example.com/", "session=abc; Path=/"); const msg = \\{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"getCookies"}} @@ -1358,14 +1471,64 @@ test "MCP - waitForState with bad state surfaces rich error" { try testing.expect(std.mem.indexOf(u8, written, "isError\":true") != null); } +test "MCP - sessions: new, list, attach isolation, close" { + defer testing.reset(); + const aa = testing.arena_allocator; + var out: std.io.Writer.Allocating = .init(aa); + var server = try Server.init(testing.allocator, testing.test_app, &out.writer); + defer server.deinit(); + // Session tools require the HTTP transport's parked-isolate discipline. + server.enableIsolateParking(); + + try router.handleMessage(server, aa, + \\{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"session_new","arguments":{"name":"a"}}} + ); + try testing.expect(std.mem.indexOf(u8, out.written(), "session a") != null); + try testing.expect(server.sessions.contains("a")); + + out.clearRetainingCapacity(); + try router.handleMessage(server, aa, + \\{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"session_list"}} + ); + // The listing is JSON nested in the tool-result text, so its quotes are + // escaped (\"default\"). + try testing.expect(std.mem.indexOf(u8, out.written(), "\\\"default\\\"") != null); + try testing.expect(std.mem.indexOf(u8, out.written(), "\\\"a\\\"") != null); + + // Routing a request to "a" (as the Mcp-Session-Id header does) and loading + // a page there leaves the default untouched, proving the two are isolated. + _ = try server.useSession("a"); + try router.handleMessage(server, aa, + \\{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"goto","arguments":{"url":"about:blank"}}} + ); + try testing.expect(server.sessions.get("a").?.session.currentFrame() != null); + try testing.expect(server.defaultSession().session.currentFrame() == null); + + // Route back to the default before closing "a" (the active session can't be closed). + _ = try server.useSession(null); + + out.clearRetainingCapacity(); + try router.handleMessage(server, aa, + \\{"jsonrpc":"2.0","id":6,"method":"tools/call","params":{"name":"session_close","arguments":{"id":"default"}}} + ); + try testing.expect(std.mem.indexOf(u8, out.written(), "cannot be closed") != null); + + out.clearRetainingCapacity(); + try router.handleMessage(server, aa, + \\{"jsonrpc":"2.0","id":7,"method":"tools/call","params":{"name":"session_close","arguments":{"id":"a"}}} + ); + try testing.expect(std.mem.indexOf(u8, out.written(), "closed session a") != null); + try testing.expect(!server.sessions.contains("a")); +} + fn testLoadPage(url: [:0]const u8, writer: *std.Io.Writer) !*Server { var server = try Server.init(testing.allocator, testing.test_app, writer); errdefer server.deinit(); - const page = try server.session.createPage(); + const page = try server.active_session.session.createPage(); try page.navigate(url, .{}); - var runner = server.session.runner(.{}); + var runner = server.active_session.session.runner(.{}); try runner.waitForFrame(page.frame_id, 2000, .{ .until = .done }); return server; }