feat: serve workspace MCP over the web app HTTP endpoint

2026-10-07 20:46:08 +09:00
parent e917cafd88
commit f41d0c1964
18 changed files with 1004 additions and 145 deletions
@@ -1,21 +1,15 @@
 import { z } from "zod";
 import type { Connection } from "../connections/model";
-import { requireTwitterConnection } from "../connections/repository.server";
 import { type Deck, type DeckColumn, deckSchema } from "../decks/model";
 import { listDecks, loadDeck } from "../decks/repository.server";
 import { emptyToolInput, prepareDeck, setDeckInput } from "../decks/webmcp-contracts";
-import {
-  fetchMastodonLists,
-  fetchMastodonPage,
-  MastodonFeedError,
-} from "../platforms/mastodon-feed.server";
-import { mapTwitterPost } from "../platforms/twitter";
+import { MastodonFeedError } from "../platforms/mastodon-feed.server";
 import type { ResearchPage, ResearchPost } from "../platforms/types";
-import { getBirdReader } from "../posts/bird-client.server";
 import { InputError, listChoicesInputSchema } from "../posts/inputs";
-import { loadListChoices, loadListPage, loadUserPage, searchPage } from "../posts/post-service";
 import { ProfileUnavailableError } from "../profiles/errors";

+import { fetchPage, publicPost, SourceFailure, listSourceLists } from "./sources.server";
+
 const MAX_FETCHES = 12;
 const POSTS_PER_FETCH = 20;
 const openInput = setDeckInput.omit({ deckId: true, expectedRevision: true }).strict();
@@ -28,50 +22,6 @@ const fetchInput = z
 const getDeckInput = z.object({ deckId: z.string().min(1).max(128) }).strict();
 const listInput = listChoicesInputSchema.strict();
 type FetchPage = (column: DeckColumn, cursor?: string) => Promise<ResearchPage>;
-class SourceFailure extends Error {
-  constructor(readonly code: string) {
-    super("The source could not be loaded.");
-  }
-}
-
-async function fetchPage(column: DeckColumn, cursor?: string): Promise<ResearchPage> {
-  if (column.source.platform === "mastodon")
-    return fetchMastodonPage({
-      connectionId: column.connectionId,
-      source: column.source,
-      cursor,
-    });
-  const reader = getBirdReader(await requireTwitterConnection(column.connectionId));
-  const input = { ...column.source, cursor };
-  const result =
-    input.kind === "search"
-      ? await searchPage(reader, input)
-      : input.kind === "user"
-        ? await loadUserPage(reader, input)
-        : await loadListPage(reader, input);
-  if (!result.ok) throw new SourceFailure(result.error.code);
-  return {
-    posts: result.page.tweets.map(mapTwitterPost),
-    nextCursor: result.page.nextCursor,
-  };
-}
-
-function publicPost(post: ResearchPost) {
-  const url = new URL(post.url);
-  if (!["https:", "http:"].includes(url.protocol) || url.username || url.password)
-    throw new SourceFailure("invalid-source-result");
-  return {
-    key: post.key,
-    nativeId: post.nativeId,
-    platform: post.platform,
-    url: url.href,
-    text: post.text,
-    author: { name: post.author.name, handle: post.author.handle },
-    ...(post.createdAt ? { createdAt: post.createdAt } : {}),
-    ...(post.contentWarning ? { contentWarning: post.contentWarning } : {}),
-    ...(post.sensitive ? { sensitive: true } : {}),
-  };
-}
 function failure(code: string, message: string) {
  return { ok: false as const, error: { code, message } };
 }
@@ -191,22 +141,7 @@ export function createResearchTools(
          if (fetches >= MAX_FETCHES)
            return failure("budget-exhausted", "This research has used its 12 fetch requests.");
          fetches += 1;
-          let lists: {
-            id: string;
-            name: string;
-            isPrivate?: boolean;
-            description?: string;
-            memberCount?: number;
-          }[];
-          if (connection.platform === "mastodon") {
-            lists = await fetchMastodonLists(connectionId);
-          } else {
-            const result = await loadListChoices(
-              getBirdReader(await requireTwitterConnection(connectionId)),
-            );
-            if (!result.ok) throw new SourceFailure(result.error.code);
-            lists = result.lists;
-          }
+          const lists = await listSourceLists(connectionId, connection.platform);
          return {
            ok: true,
            connectionId,
@@ -1,13 +1,32 @@
 // @vitest-environment node
 import { mkdtemp, readFile, rm } from "node:fs/promises";
+import { createServer } from "node:http";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { fileURLToPath } from "node:url";
-import { expect, it, vi } from "vitest";
+import { assert, expect, it, vi } from "vitest";
+import { handleMcpRequest } from "../mcp/http.server";
 import { executeCodexResearch } from "./codex-provider.server";

 it("streams and resumes through the real SDK provider with a local MCP roundtrip", async () => {
  const directory = await mkdtemp(join(tmpdir(), "codex-provider-"));
+  const server = createServer(async (incoming, outgoing) => {
+    const chunks: Buffer[] = [];
+    for await (const chunk of incoming) chunks.push(Buffer.from(chunk));
+    const response = await handleMcpRequest(
+      new Request(`http://127.0.0.1${incoming.url}`, {
+        method: incoming.method,
+        headers: incoming.headers as Record<string, string>,
+        body: incoming.method === "POST" ? Buffer.concat(chunks) : undefined,
+      }),
+    );
+    outgoing.writeHead(response.status, Object.fromEntries(response.headers));
+    outgoing.end(Buffer.from(await response.arrayBuffer()));
+  });
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address();
+  assert.isNotNull(address);
+  assert(typeof address === "object");
  try {
    const execute = vi.fn<(name: string, input: unknown) => Promise<unknown>>(async () => ({
      ok: true,
@@ -20,6 +39,7 @@ it("streams and resumes through the real SDK provider with a local MCP roundtrip
        model: "fixture-model",
        reportRoot: directory,
        codexPath: fileURLToPath(new URL("./fixtures/codex-app-server.mjs", import.meta.url)),
+        mcpUrl: `http://127.0.0.1:${address.port}/mcp`,
      },
      cwd: directory,
      runId: "fixture-run",
@@ -68,6 +88,9 @@ it("streams and resumes through the real SDK provider with a local MCP roundtrip
    expect(methods).toContain("thread/resume");
    expect(methods).toContain("turn/interrupt");
  } finally {
+    await new Promise<void>((resolve, reject) =>
+      server.close((error) => (error ? reject(error) : resolve())),
+    );
    await rm(directory, { recursive: true, force: true });
  }
 }, 20_000);
@@ -3,13 +3,15 @@ import { homedir } from "node:os";
 import { resolve } from "node:path";
 import { promisify } from "node:util";
 import { smoothStream, streamText } from "ai";
-import { createCodexAppServer, createSdkMcpServer } from "ai-sdk-provider-codex-cli";
+import { createCodexAppServer } from "ai-sdk-provider-codex-cli";
+import { registerResearchTools } from "../mcp/http.server";

 const execFileAsync = promisify(execFile);
 export type CodexResearchConfig = {
  model: string;
  reportRoot: string;
  codexPath?: string;
+  mcpUrl?: string;
 };
 export type CodexResearchInput = {
  config: CodexResearchConfig;
@@ -92,56 +94,63 @@ export async function executeCodexResearch(input: CodexResearchInput): Promise<v
    input.signal.throwIfAborted();
    return result;
  }
-  const bridge = createSdkMcpServer({
-    name,
-    tools: input.tools.definitions.map((definition) => ({
-      ...definition,
-      execute: (args: unknown) => executeTool(definition.name, args),
-    })),
+  const url = new URL(
+    input.config.mcpUrl ||
+      process.env.TWITTER_LITE_MCP_URL ||
+      `http://127.0.0.1:${process.env.PORT || 3000}/mcp`,
+  );
+  const registration = registerResearchTools({
+    definitions: input.tools.definitions,
+    execute: executeTool,
  });
-  const provider = createCodexAppServer({
-    defaultSettings: {
-      codexPath: binary,
-      cwd: input.cwd,
-      env,
-      logger: false,
-      minCodexVersion: "0.156.0",
-      threadMode: "persistent",
-      resume: input.threadId,
-      approvalPolicy: "never",
-      sandboxPolicy: "workspace-write",
-      autoApprove: false,
-      developerInstructions: input.instructions,
-      mcpServers: { [name]: bridge },
-      serverRequests: {
-        onDynamicToolCall: async ({ params }) => {
-          const result = await executeTool(params.tool, params.arguments);
-          return {
-            contentItems: [{ type: "inputText", text: JSON.stringify(result) }],
-            success: !!result && typeof result === "object" && "ok" in result && result.ok === true,
-          };
+  url.searchParams.set("research", registration.id);
+  let provider: ReturnType<typeof createCodexAppServer> | undefined;
+  try {
+    provider = createCodexAppServer({
+      defaultSettings: {
+        codexPath: binary,
+        cwd: input.cwd,
+        env,
+        logger: false,
+        minCodexVersion: "0.156.0",
+        threadMode: "persistent",
+        resume: input.threadId,
+        approvalPolicy: "never",
+        sandboxPolicy: "workspace-write",
+        autoApprove: false,
+        developerInstructions: input.instructions,
+        mcpServers: { [name]: { transport: "http", url: url.toString() } },
+        serverRequests: {
+          onDynamicToolCall: async ({ params }) => {
+            const result = await executeTool(params.tool, params.arguments);
+            return {
+              contentItems: [{ type: "inputText", text: JSON.stringify(result) }],
+              success:
+                !!result && typeof result === "object" && "ok" in result && result.ok === true,
+            };
+          },
+        },
+        configOverrides: {
+          "shell_environment_policy.inherit": "none",
+          "features.multi_agent": false,
+          "features.hooks": false,
+          web_search: "disabled",
+          ...Object.fromEntries(
+            inherited.map((server) => [`mcp_servers.${server}.enabled`, false]),
+          ),
+          [`mcp_servers.${name}.enabled`]: true,
+          ...Object.fromEntries(
+            input.tools.definitions.map((tool) => [
+              `mcp_servers.${name}.tools.${tool.name}.approval_mode`,
+              "approve",
+            ]),
+          ),
+        },
+        onSessionCreated: (session) => {
+          if (!input.signal.aborted) input.onThread(session.threadId);
        },
      },
-      configOverrides: {
-        "shell_environment_policy.inherit": "none",
-        "features.multi_agent": false,
-        "features.hooks": false,
-        web_search: "disabled",
-        ...Object.fromEntries(inherited.map((server) => [`mcp_servers.${server}.enabled`, false])),
-        [`mcp_servers.${name}.enabled`]: true,
-        ...Object.fromEntries(
-          input.tools.definitions.map((tool) => [
-            `mcp_servers.${name}.tools.${tool.name}.approval_mode`,
-            "approve",
-          ]),
-        ),
-      },
-      onSessionCreated: (session) => {
-        if (!input.signal.aborted) input.onThread(session.threadId);
-      },
-    },
-  });
-  try {
+    });
    const result = streamText({
      model: provider(input.config.model),
      prompt: input.prompt,
@@ -165,6 +174,7 @@ export async function executeCodexResearch(input: CodexResearchInput): Promise<v
      throw new Error(step.rawFinishReason || "Codex could not complete the turn.");
    }
  } finally {
-    await provider.close();
+    registration.dispose();
+    await provider?.close();
  }
 }
@@ -2,12 +2,6 @@
 import { assert, beforeEach, expect, it, vi } from "vitest";
 import { type CodexResearchInput, executeCodexResearch } from "./codex-provider.server";

-type Bridge = {
-  name: string;
-  tools: (CodexResearchInput["tools"]["definitions"][number] & {
-    execute: (input: unknown) => Promise<unknown>;
-  })[];
-};
 type ProviderSettings = import("ai-sdk-provider-codex-cli").CodexAppServerSettings;
 const fake = vi.hoisted(() => ({
  inspect:
@@ -19,7 +13,8 @@ const fake = vi.hoisted(() => ({
      ) => Promise<{ stdout: string }>
    >(),
  create: vi.fn<(options: { defaultSettings: ProviderSettings }) => unknown>(),
-  bridge: vi.fn<(options: Bridge) => Bridge>(),
+  register: vi.fn<(tools: CodexResearchInput["tools"]) => { id: string; dispose: () => void }>(),
+  dispose: vi.fn<() => void>(),
  model: Object.assign(vi.fn<(id: string) => string>(), { close: vi.fn<() => Promise<void>>() }),
  stream: vi.fn<
    (options: unknown) => {
@@ -38,14 +33,15 @@ vi.mock("node:util", () => ({ promisify: (fn: unknown) => fn }));
 vi.mock("ai", () => ({ streamText: fake.stream, smoothStream: fake.smooth }));
 vi.mock("ai-sdk-provider-codex-cli", () => ({
  createCodexAppServer: fake.create,
-  createSdkMcpServer: fake.bridge,
 }));
+vi.mock("../mcp/http.server", () => ({ registerResearchTools: fake.register }));
 function input(): CodexResearchInput {
  return {
    config: {
      reportRoot: "/reports",
      model: "selected-model",
      codexPath: "/bin/codex",
+      mcpUrl: "http://127.0.0.1:3002/mcp",
    },
    cwd: "/reports/run",
    runId: "test-run",
@@ -72,7 +68,7 @@ beforeEach(() => {
    stdout: JSON.stringify([{ name: "external" }]),
  });
  fake.create.mockReturnValue(fake.model);
-  fake.bridge.mockImplementation((options) => options);
+  fake.register.mockReturnValue({ id: "turn-scope", dispose: fake.dispose });
  fake.model.mockReturnValue("model");
  fake.model.close.mockResolvedValue(undefined);
  fake.smooth.mockReturnValue("smoothing-transform");
@@ -121,19 +117,34 @@ it("uses the app-server provider for incremental Japanese streaming with persist
  expect(fake.model.close).toHaveBeenCalledOnce();
 });

-it("bridges the same validated research definitions into local MCP tools", async () => {
+it("connects Codex to the shared HTTP endpoint with this turn's scoped tools", async () => {
  const request = input();
  await executeCodexResearch(request);
-  const bridge = fake.bridge.mock.calls[0]?.[0];
-  assert.isDefined(bridge);
-  assert.isDefined(bridge.tools[0]);
-  expect(bridge.tools[0]).toMatchObject({
-    name: "list_decks",
-    inputSchema: { type: "object" },
+  const tools = fake.register.mock.calls[0]?.[0];
+  assert.isDefined(tools);
+  expect(tools.definitions).toEqual(request.tools.definitions);
+  expect(fake.create).toHaveBeenCalledWith({
+    defaultSettings: expect.objectContaining({
+      mcpServers: {
+        workspace_research_testrun: {
+          transport: "http",
+          url: "http://127.0.0.1:3002/mcp?research=turn-scope",
+        },
+      },
+    }),
  });
  const args = { example: "input" };
-  expect(await bridge.tools[0].execute(args)).toEqual({ ok: true, decks: [] });
+  expect(await tools.execute("list_decks", args)).toEqual({ ok: true, decks: [] });
  expect(request.tools.execute).toHaveBeenCalledWith("list_decks", args);
+  expect(fake.dispose).toHaveBeenCalledOnce();
+});
+
+it("removes the registered scope when provider startup fails", async () => {
+  fake.create.mockImplementationOnce(() => {
+    throw new Error("Startup failed");
+  });
+  await expect(executeCodexResearch(input())).rejects.toThrow("Startup failed");
+  expect(fake.dispose).toHaveBeenCalledOnce();
 });

 it("excludes web-app secrets and disables inherited MCP servers, hooks and extra tools", async () => {
@@ -235,10 +246,10 @@ it("propagates cancellation and prevents late bridge calls or metadata changes",
    controller.abort(new Error("Stopped")),
  );
  await expect(executeCodexResearch(request)).rejects.toThrow("Stopped");
-  const bridge = fake.bridge.mock.calls[0]?.[0];
-  assert.isDefined(bridge);
-  assert.isDefined(bridge.tools[0]);
-  await expect(bridge.tools[0].execute({})).rejects.toThrow("Stopped");
+  const tools = fake.register.mock.calls[0]?.[0];
+  assert.isDefined(tools);
+  await expect(tools.execute("list_decks", {})).rejects.toThrow("Stopped");
+  expect(fake.dispose).toHaveBeenCalledOnce();
  expect(request.tools.execute).not.toHaveBeenCalled();
  expect(request.onThread).not.toHaveBeenCalled();
  expect(fake.model.close).toHaveBeenCalledOnce();
@@ -62,12 +62,19 @@ events.on("line", async (line) => {
          method: "POST",
          headers: {
            "content-type": "application/json",
+            accept: "application/json, text/event-stream",
            ...config[`${prefix}.http_headers`],
          },
          body: JSON.stringify({ jsonrpc: "2.0", id: 1, method, params }),
        });
        if (!response.ok) throw new Error(`MCP HTTP ${response.status}`);
-        return response.json();
+        const text = await response.text();
+        if (response.headers.get("content-type")?.includes("text/event-stream")) {
+          const data = text.split("\n").find((line) => line.startsWith("data:"));
+          if (!data) throw new Error("MCP response missing SSE data");
+          return JSON.parse(data.slice(5).trim());
+        }
+        return JSON.parse(text);
      };
      const catalog = await rpc("tools/list", {});
      if (!catalog.result.tools.some((tool) => tool.name === "list_decks"))
@@ -0,0 +1,73 @@
+import { requireTwitterConnection } from "../connections/repository.server";
+import type { Connection } from "../connections/model";
+import type { DeckColumn } from "../decks/model";
+import { fetchMastodonLists, fetchMastodonPage } from "../platforms/mastodon-feed.server";
+import { mapTwitterPost } from "../platforms/twitter";
+import type { ResearchPage, ResearchPost } from "../platforms/types";
+import { getBirdReader } from "../posts/bird-client.server";
+import { loadListChoices, loadListPage, loadUserPage, searchPage } from "../posts/post-service";
+
+export class SourceFailure extends Error {
+  constructor(readonly code: string) {
+    super("The source could not be loaded.");
+  }
+}
+
+export async function fetchPage(column: DeckColumn, cursor?: string): Promise<ResearchPage> {
+  if (column.source.platform === "mastodon")
+    return fetchMastodonPage({
+      connectionId: column.connectionId,
+      source: column.source,
+      cursor,
+    });
+  const reader = getBirdReader(await requireTwitterConnection(column.connectionId));
+  const input = { ...column.source, cursor };
+  const result =
+    input.kind === "search"
+      ? await searchPage(reader, input)
+      : input.kind === "user"
+        ? await loadUserPage(reader, input)
+        : await loadListPage(reader, input);
+  if (!result.ok) throw new SourceFailure(result.error.code);
+  return {
+    posts: result.page.tweets.map(mapTwitterPost),
+    nextCursor: result.page.nextCursor,
+  };
+}
+
+export function publicPost(post: ResearchPost) {
+  const url = new URL(post.url);
+  if (!["https:", "http:"].includes(url.protocol) || url.username || url.password)
+    throw new SourceFailure("invalid-source-result");
+  return {
+    key: post.key,
+    nativeId: post.nativeId,
+    platform: post.platform,
+    url: url.href,
+    text: post.text,
+    author: { name: post.author.name, handle: post.author.handle },
+    ...(post.createdAt ? { createdAt: post.createdAt } : {}),
+    ...(post.contentWarning ? { contentWarning: post.contentWarning } : {}),
+    ...(post.sensitive ? { sensitive: true } : {}),
+  };
+}
+export async function listSourceLists(connectionId: string, platform: Connection["platform"]) {
+  let lists: {
+    id: string;
+    name: string;
+    isPrivate?: boolean;
+    description?: string;
+    memberCount?: number;
+  }[];
+  if (platform === "mastodon") {
+    lists = await fetchMastodonLists(connectionId);
+  } else {
+    const result = await loadListChoices(
+      getBirdReader(await requireTwitterConnection(connectionId)),
+    );
+    if (!result.ok) throw new SourceFailure(result.error.code);
+    lists = result.lists;
+  }
+
+  return lists;
+}