feat(server): MCP tools that start and drive threads

apps/server/src/mcp/McpActor.ts

@@ -0,0 +1,54 @@
+import type { ThreadId } from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as Fiber from "effect/Fiber";
+import * as Option from "effect/Option";
+
+import type * as McpInvocationContext from "./McpInvocationContext.ts";
+
+/**
+ * Who is calling an MCP tool. An agent inside a thread holds that thread's
+ * provider-session credential on `/mcp`; an agent outside T3 Code holds an
+ * environment access token on `/mcp/operate`. Tools that act across threads
+ * read this to decide what the caller may do, and to record who did it.
+ */
+export type McpActorValue =
+  | {
+      readonly kind: "thread";
+      readonly threadId: ThreadId;
+      readonly capabilities: ReadonlySet<McpInvocationContext.McpCapability>;
+    }
+  | {
+      readonly kind: "token";
+      /** The token's label from Settings → Connections → Agent access. */
+      readonly label: string;
+    };
+
+export class McpActor extends Context.Service<McpActor, McpActorValue>()("t3/mcp/McpActor") {}
+
+/** The thread actor for a `/mcp` credential. */
+export const fromInvocation = (
+  invocation: McpInvocationContext.McpInvocationScope,
+): McpActorValue => ({
+  kind: "thread",
+  threadId: invocation.threadId,
+  capabilities: invocation.capabilities,
+});
+
+/** Whether the actor may drive threads: a token on `/mcp/operate`, or a thread granted control. */
+export const canOperate = (actor: McpActorValue): boolean =>
+  actor.kind === "token" || actor.capabilities.has("orchestration");
+
+/**
+ * The `EnabledWhen` predicate for operate tools. MCP lists tools without
+ * passing the request through, but the list and call handlers run on the
+ * request's fiber, where the auth middleware put the actor. Reading it there
+ * means an agent whose credential cannot operate never sees these tools, so
+ * they cost it nothing in its context.
+ */
+export const operateToolsVisible = (): boolean => {
+  const fiber = Fiber.getCurrent();
+  if (fiber === undefined) return false;
+  return Context.getOption(fiber.context, McpActor).pipe(
+    Option.match({ onNone: () => false, onSome: canOperate }),
+  );
+};

apps/server/src/mcp/McpHttpServer.test.ts

@@ -10,9 +10,20 @@ import * as Path from "effect/Path";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
 import { McpProtocol, McpSchema, McpServer } from "effect/unstable/ai";
-import { HttpBody, HttpClient, HttpRouter, HttpServerResponse } from "effect/unstable/http";
-
+import {
+  HttpBody,
+  HttpClient,
+  HttpRouter,
+  HttpServerRequest,
+  HttpServerResponse,
+} from "effect/unstable/http";
+
+import * as GitWorkflowService from "../git/GitWorkflowService.ts";
+import * as ClientCommandDispatcher from "../orchestration/ClientCommandDispatcher.ts";
 import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts";
+import * as ProviderRegistry from "../provider/Services/ProviderRegistry.ts";
+import * as ServerSettings from "../serverSettings.ts";
+import * as McpActor from "./McpActor.ts";
 import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";
 import * as ServerConfig from "../config.ts";
 import * as McpHttpServer from "./McpHttpServer.ts";
@@ -870,3 +881,104 @@ it.effect("registers annotated tools and preserves authenticated request context
     }),
   ).pipe(Effect.provide(TestLayer)),
 );
+
+it.effect("lists thread-control tools only to credentials that may drive threads", () =>
+  Effect.scoped(
+    Effect.gen(function* () {
+      // Stands in for the /mcp credential check: the header picks the thread's capabilities.
+      const actorMiddleware = HttpRouter.middleware<{ provides: McpActor.McpActor }>()(
+        Effect.succeed((httpEffect) =>
+          Effect.gen(function* () {
+            const request = yield* HttpServerRequest.HttpServerRequest;
+            const control = request.headers["x-thread-control"] === "on";
+            return yield* httpEffect.pipe(
+              Effect.provideService(McpActor.McpActor, {
+                kind: "thread",
+                threadId,
+                capabilities: new Set<McpInvocationContext.McpCapability>(
+                  control ? ["orchestration"] : [],
+                ),
+              }),
+              Effect.map(McpHttpServer.normalizeMcpHttpResponse),
+            );
+          }),
+        ),
+      ).layer;
+      const serverLayer = McpHttpServer.OperateToolkitRegistrationLive.pipe(
+        Layer.provideMerge(
+          McpServer.layerHttp({
+            name: "MCP visibility test",
+            version: "1.0.0",
+            path: "/mcp",
+            protocols: [McpProtocol.v2025_06_18],
+          }).pipe(Layer.provide(actorMiddleware)),
+        ),
+        Layer.provide(
+          Layer.mergeAll(
+            Layer.mock(ClientCommandDispatcher.ClientCommandDispatcher)({
+              forOrigin: () => ({ dispatch: () => Effect.die("unused") }),
+            }),
+            Layer.mock(OrchestrationEngineService)({}),
+            Layer.mock(ProjectionSnapshotQuery)({}),
+            Layer.mock(ProviderRegistry.ProviderRegistry)({}),
+            Layer.mock(ServerSettings.ServerSettingsService)({}),
+            Layer.mock(GitWorkflowService.GitWorkflowService)({}),
+            NodeServices.layer,
+          ),
+        ),
+      );
+      yield* HttpRouter.serve(serverLayer, { disableListenLog: true, disableLogger: true }).pipe(
+        Layer.build,
+      );
+      const httpClient = yield* HttpClient.HttpClient;
+
+      const listToolNames = (control: "on" | "off") =>
+        Effect.gen(function* () {
+          const headers = {
+            accept: "application/json, text/event-stream",
+            "x-thread-control": control,
+          };
+          const initialized = yield* httpClient.post("/mcp", {
+            headers,
+            body: HttpBody.text(
+              `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"mcp-test","version":"1.0.0"}}}`,
+              "application/json",
+            ),
+          });
+          const sessionId = initialized.headers["mcp-session-id"]!;
+          const listed = yield* httpClient.post("/mcp", {
+            headers: {
+              ...headers,
+              "mcp-session-id": sessionId,
+              "mcp-protocol-version": "2025-06-18",
+            },
+            body: HttpBody.text(
+              `{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}`,
+              "application/json",
+            ),
+          });
+          const text = yield* listed.text;
+          expect(listed.status).toBe(200);
+          const json = text.trimStart().startsWith("{")
+            ? text
+            : (text
+                .split("\n")
+                .find((line) => line.startsWith("data:"))
+                ?.slice(5) ?? "{}");
+          const body = decodeJsonText(json) as {
+            readonly result?: { readonly tools?: ReadonlyArray<{ readonly name: string }> };
+          };
+          return (body.result?.tools ?? []).map((tool) => tool.name).sort();
+        });
+
+      expect(yield* listToolNames("off")).toEqual([]);
+      expect(yield* listToolNames("on")).toEqual([
+        "create_thread",
+        "interrupt_turn",
+        "list_models",
+        "send_message",
+        "wait_for_thread",
+      ]);
+    }),
+  ).pipe(Effect.provide(NodeHttpServer.layerTest)),
+);

apps/server/src/mcp/McpHttpServer.ts

@@ -18,6 +18,7 @@ import { PreviewAutomationError } from "@t3tools/contracts";
 import packageJson from "../../package.json" with { type: "json" };
 import * as ServerConfig from "../config.ts";
 import * as DeviceService from "../device/DeviceService.ts";
+import * as McpActor from "./McpActor.ts";
 import * as McpInvocationContext from "./McpInvocationContext.ts";
 import * as McpSessionRegistry from "./McpSessionRegistry.ts";
 import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
@@ -30,6 +31,8 @@ import {
   PreviewSnapshotToolkit,
   PreviewStandardToolkit,
 } from "./toolkits/preview/tools.ts";
+import { OperateToolkitHandlersLive } from "./toolkits/operate/handlers.ts";
+import { OperateToolkit } from "./toolkits/operate/tools.ts";
 import { PullRequestsToolkitHandlersLive } from "./toolkits/pullRequests/handlers.ts";
 import { PullRequestsToolkit } from "./toolkits/pullRequests/tools.ts";
 import {
@@ -59,7 +62,7 @@ const unauthorized = HttpServerResponse.jsonUnsafe(
 type AuthenticatedHttpEffect = Effect.Effect<
   HttpServerResponse.HttpServerResponse,
   Types.unhandled,
-  McpInvocationContext.McpInvocationContext
+  McpInvocationContext.McpInvocationContext | McpActor.McpActor
 >;
 
 type McpAuthMiddleware = (
@@ -103,6 +106,7 @@ const makeMcpAuthMiddleware = McpSessionRegistry.McpSessionRegistry.pipe(
       }
       return yield* httpEffect.pipe(
         Effect.provideService(McpInvocationContext.McpInvocationContext, invocation),
+        Effect.provideService(McpActor.McpActor, McpActor.fromInvocation(invocation)),
         Effect.map(normalizeMcpHttpResponse),
       );
     }),
@@ -111,7 +115,7 @@ const makeMcpAuthMiddleware = McpSessionRegistry.McpSessionRegistry.pipe(
 );
 
 const McpAuthMiddlewareLive = HttpRouter.middleware<{
-  provides: McpInvocationContext.McpInvocationContext;
+  provides: McpInvocationContext.McpInvocationContext | McpActor.McpActor;
 }>()(makeMcpAuthMiddleware).layer;
 
 /**
@@ -647,6 +651,11 @@ export const PullRequestsToolkitRegistrationLive = McpServer.toolkit(PullRequest
   Layer.provide(PullRequestsToolkitHandlersLive),
 );
 
+/** Listed only to credentials that may drive threads; see McpActor.operateToolsVisible. */
+export const OperateToolkitRegistrationLive = McpServer.toolkit(OperateToolkit).pipe(
+  Layer.provide(OperateToolkitHandlersLive),
+);
+
 const DeviceStandardToolkitRegistrationLive = McpServer.toolkit(DeviceStandardToolkit).pipe(
   Layer.provide(DeviceStandardToolkitHandlersLive),
 );
@@ -671,4 +680,5 @@ export const layer = Layer.mergeAll(
   PreviewToolkitRegistrationLive,
   PullRequestsToolkitRegistrationLive,
   DeviceToolkitRegistrationLive,
+  OperateToolkitRegistrationLive,
 ).pipe(Layer.provideMerge(McpTransportLive));

apps/server/src/mcp/McpInvocationContext.ts

@@ -8,7 +8,7 @@ import {
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 
-export type McpCapability = "preview" | "device" | "pull-requests";
+export type McpCapability = "preview" | "device" | "pull-requests" | "orchestration";
 
 export interface McpInvocationScope {
   readonly environmentId: EnvironmentId;