feat(cli): MCP tools to read and write session channels

Add read_session_channel and write_session_channel MCP tools so an MCP
client can observe a session's realtime streams (a named side channel or
the reserved transcript pair) and send control input to a running agent
on a channel's .in without waking a run. Read is a point-in-time drain
with cursor pagination; write appends one record to a channel's .in.

Author: ericallam

packages/cli-v3/src/mcp/config.ts

@@ -241,4 +241,16 @@ export const toolsMetadata = {
     description:
       "Close an agent chat conversation. The agent exits its loop gracefully. Without this, the agent will close on its own when its idle timeout expires.",
   },
+  read_session_channel: {
+    name: "read_session_channel",
+    title: "Read Session Channel",
+    description:
+      "Read records from a session's realtime stream: a named side channel (pass `channel`) or the reserved chat transcript pair (omit `channel`). Returns whatever records currently exist after the optional cursor and closes, so it is a point-in-time drain, not a live subscription. Read `out` for the producer's feed (e.g. a screencast) or `in` for what clients have sent. Use the returned nextCursor as `afterEventId` to page forward.",
+  },
+  write_session_channel: {
+    name: "write_session_channel",
+    title: "Write Session Channel",
+    description:
+      "Append one record to a named side channel's `in` stream on a session. Use this to send control input to a running agent (e.g. a pause/viewport command) without waking or triggering a run. Requires a `channel` name; the reserved transcript and the `out` side are not writable here (`out` is producer-only). Pass a JSON string as `value` for structured records.",
+  },
 };

packages/cli-v3/src/mcp/tools.ts

@@ -32,6 +32,7 @@ import {
 } from "./tools/prompts.js";
 import { listAgentsTool } from "./tools/agents.js";
 import { startAgentChatTool, sendAgentMessageTool, closeAgentChatTool } from "./tools/agentChat.js";
+import { readSessionChannelTool, writeSessionChannelTool } from "./tools/sessionChannels.js";
 import { respondWithError } from "./utils.js";
 
 /** Tool names that perform write/mutating operations. */
@@ -49,6 +50,7 @@ const WRITE_TOOLS = new Set([
   startAgentChatTool.name,
   sendAgentMessageTool.name,
   closeAgentChatTool.name,
+  writeSessionChannelTool.name,
 ]);
 
 export function registerTools(context: McpContext) {
@@ -90,6 +92,8 @@ export function registerTools(context: McpContext) {
     startAgentChatTool,
     sendAgentMessageTool,
     closeAgentChatTool,
+    readSessionChannelTool,
+    writeSessionChannelTool,
     getReportTool,
   ];
 

packages/cli-v3/src/mcp/tools/sessionChannels.ts

@@ -0,0 +1,161 @@
+import { z } from "zod";
+import { toolsMetadata } from "../config.js";
+import { CommonProjectsInput } from "../schemas.js";
+import { respondWithError, toolHandler } from "../utils.js";
+
+const SESSION_CHANNEL_NAME_REGEX = /^[A-Za-z0-9._-]{1,128}$/;
+
+const ReadSessionChannelInput = CommonProjectsInput.extend({
+  sessionId: z
+    .string()
+    .describe("The session id (session_* friendlyId) or the externalId it was created with."),
+  channel: z
+    .string()
+    .describe(
+      "The named side channel to read. Omit to read the session's reserved chat transcript pair."
+    )
+    .optional(),
+  io: z
+    .enum(["out", "in"])
+    .describe("Which side to read: `out` (producer feed) or `in` (client input).")
+    .default("out"),
+  afterEventId: z
+    .string()
+    .describe(
+      "Cursor: only return records after this event id. Use the nextCursor from a prior read."
+    )
+    .optional(),
+  maxRecords: z
+    .number()
+    .int()
+    .positive()
+    .max(500)
+    .describe("Maximum records to return (default 100).")
+    .default(100),
+});
+
+export const readSessionChannelTool = {
+  name: toolsMetadata.read_session_channel.name,
+  title: toolsMetadata.read_session_channel.title,
+  description: toolsMetadata.read_session_channel.description,
+  inputSchema: ReadSessionChannelInput.shape,
+  handler: toolHandler(ReadSessionChannelInput.shape, async (input, { ctx }) => {
+    ctx.logger?.log("calling read_session_channel", { input });
+
+    if (ctx.options.devOnly && input.environment !== "dev") {
+      return respondWithError(`This MCP server is only available for the dev environment.`);
+    }
+
+    if (input.channel !== undefined && !SESSION_CHANNEL_NAME_REGEX.test(input.channel)) {
+      return respondWithError(
+        `Invalid channel name "${input.channel}": use 1-128 chars from [A-Za-z0-9._-].`
+      );
+    }
+
+    const projectRef = await ctx.getProjectRef({
+      projectRef: input.projectRef,
+      cwd: input.configPath,
+    });
+
+    const apiClient = await ctx.getApiClient({
+      projectRef,
+      environment: input.environment,
+      scopes: ["read:sessions"],
+      branch: input.branch,
+    });
+
+    const { records } = await apiClient.readSessionStreamRecords(input.sessionId, input.io, {
+      channel: input.channel,
+      afterEventId: input.afterEventId,
+    });
+
+    const limited = records.slice(0, input.maxRecords);
+    const hasMore = records.length > limited.length;
+    const nextCursor = limited.at(-1)?.seqNum;
+
+    const label = input.channel ? `channel "${input.channel}"` : "reserved pair";
+    const header = `Session ${input.sessionId} ${label} .${input.io}: ${limited.length} record${
+      limited.length === 1 ? "" : "s"
+    }${hasMore ? ` (more available)` : ""}`;
+
+    const lines = limited.map((record) => {
+      const data = typeof record.data === "string" ? record.data : JSON.stringify(record.data);
+      return `#${record.seqNum} ${data}`;
+    });
+
+    const footer =
+      nextCursor !== undefined && hasMore
+        ? `\n\nMore records available. Read again with afterEventId "${nextCursor}" to continue.`
+        : "";
+
+    return {
+      content: [
+        {
+          type: "text",
+          text: [header, "", ...lines].join("\n") + footer,
+        },
+      ],
+    };
+  }),
+};
+
+const WriteSessionChannelInput = CommonProjectsInput.extend({
+  sessionId: z
+    .string()
+    .describe("The session id (session_* friendlyId) or the externalId it was created with."),
+  channel: z.string().describe("The named side channel to write to."),
+  value: z
+    .string()
+    .describe(
+      "The record to append to the channel's `in` stream. Pass a JSON string for structured records (e.g. '{\"paused\":true}')."
+    ),
+});
+
+export const writeSessionChannelTool = {
+  name: toolsMetadata.write_session_channel.name,
+  title: toolsMetadata.write_session_channel.title,
+  description: toolsMetadata.write_session_channel.description,
+  inputSchema: WriteSessionChannelInput.shape,
+  handler: toolHandler(WriteSessionChannelInput.shape, async (input, { ctx }) => {
+    ctx.logger?.log("calling write_session_channel", { input });
+
+    if (ctx.options.devOnly && input.environment !== "dev") {
+      return respondWithError(`This MCP server is only available for the dev environment.`);
+    }
+
+    if (!SESSION_CHANNEL_NAME_REGEX.test(input.channel)) {
+      return respondWithError(
+        `Invalid channel name "${input.channel}": use 1-128 chars from [A-Za-z0-9._-].`
+      );
+    }
+
+    const projectRef = await ctx.getProjectRef({
+      projectRef: input.projectRef,
+      cwd: input.configPath,
+    });
+
+    const apiClient = await ctx.getApiClient({
+      projectRef,
+      environment: input.environment,
+      scopes: ["write:sessions"],
+      branch: input.branch,
+    });
+
+    await apiClient.appendToSessionStream(
+      input.sessionId,
+      "in",
+      input.value,
+      undefined,
+      input.channel
+    );
+
+    return {
+      content: [
+        {
+          type: "text",
+          text: `Wrote 1 record to session ${input.sessionId} channel "${input.channel}" .in. This does not wake or trigger a run.`,
+        },
+      ],
+    };
+  }),
+};