feat: Phase 1 — Authentik auth, MCP endpoint, persistent memory

End-to-end Phase 1 of shared-memory: a logged-in Authentik user can sign into the Web UI (/me debug page), and an MCP client with an Authentik- issued bearer token can call memory.write / memory.list / memory.get / memory.delete plus project.identify against /api/mcp. Stack: - Next.js 15 (App Router) + React 19 + TypeScript, pnpm workspaces - Drizzle ORM + Postgres 16 + pgvector + pg_trgm - Auth.js v5 with Authentik provider (Web UI) - jose + Authentik JWKS for MCP bearer-token validation - JSON-RPC 2.0 dispatcher implementing the MCP wire protocol over plain HTTP POST (hand-rolled to fit Next.js App Router; switches to SSE in a later phase if server-initiated events are needed) - bge-small embeddings sidecar deferred to Phase 2; the schema already reserves the vector(384) column + IVFFlat index, FTS via a STORED tsvector column, and the visibility enum (private/shared/team) so cross-user memory sharing can be added without a future migration Deployment supports two modes (set in .env, never committed): - Behind an external reverse proxy (HAProxy / nginx / Cloudflare Tunnel / Traefik) — DEFAULT; the app exposes APP_PORT on the host with X-Forwarded-* trusted, no in-container TLS - Built-in TLS via Caddy — opt-in with `docker compose --profile tls up` Discovery endpoint at /.well-known/oauth-protected-resource (RFC 9728) points MCP clients at the Authentik authorization server after a 401. README walks through both Authentik providers (Web UI + MCP resource server), the audience scope mapping, redirect URIs, and includes a worked HAProxy config snippet. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-15 07:04:11 -07:00
parent d5be753cfb
commit 077d0a0825
37 changed files with 7294 additions and 8 deletions
@@ -0,0 +1,63 @@
+import { db } from "@/lib/db/client";
+import { users } from "@/lib/db/schema";
+import { and, eq } from "drizzle-orm";
+import type { AuthenticatedClaims } from "@/lib/auth/jwt";
+
+/**
+ * Per-request user context for MCP tool handlers.
+ *
+ * Resolves (or creates) the internal `users` row from the Authentik OIDC
+ * claims so tools work with stable UUID foreign keys rather than raw `sub`
+ * strings.
+ */
+export interface UserContext {
+  /** Internal users.id UUID. */
+  userId: string;
+  /** OIDC sub claim (stable identifier from Authentik). */
+  sub: string;
+  /** OIDC issuer. */
+  iss: string;
+  /** Optional profile fields if present in the access token. */
+  email: string | null;
+  name: string | null;
+}
+
+export async function userContextFromClaims(claims: AuthenticatedClaims): Promise<UserContext> {
+  const email = (claims.email as string | undefined) ?? null;
+  const name = (claims.name as string | undefined) ?? null;
+  const picture = (claims.picture as string | undefined) ?? null;
+
+  const row = await db
+    .insert(users)
+    .values({
+      oidcSub: claims.sub,
+      oidcIss: claims.iss,
+      email,
+      name,
+      picture,
+    })
+    .onConflictDoUpdate({
+      target: [users.oidcIss, users.oidcSub],
+      set: {
+        email,
+        name,
+        picture,
+        lastSeenAt: new Date(),
+      },
+    })
+    .returning({ id: users.id });
+
+  const userId = row[0]?.id;
+  if (!userId) {
+    // Race against another upsert — fall back to a select.
+    const existing = await db
+      .select({ id: users.id })
+      .from(users)
+      .where(and(eq(users.oidcIss, claims.iss), eq(users.oidcSub, claims.sub)))
+      .limit(1);
+    if (!existing[0]) throw new Error("user upsert failed and not found on re-read");
+    return { userId: existing[0].id, sub: claims.sub, iss: claims.iss, email, name };
+  }
+
+  return { userId, sub: claims.sub, iss: claims.iss, email, name };
+}
@@ -0,0 +1,132 @@
+import { tools, toolMap, type ToolResult } from "./tools";
+import type { UserContext } from "./context";
+
+/**
+ * Minimal JSON-RPC 2.0 dispatcher that implements the MCP wire protocol over
+ * HTTP. We intentionally don't depend on the SDK's `StreamableHTTPServerTransport`
+ * here because Next.js App Router uses Web `Request`/`Response`, not Node's
+ * `IncomingMessage`/`ServerResponse`, and a hand-rolled handler is simpler than
+ * a Node-stream shim. The protocol surface we cover for Phase 1 is:
+ *   - `initialize`            — handshake
+ *   - `notifications/initialized` — ack (no response)
+ *   - `tools/list`            — enumerate tools
+ *   - `tools/call`            — invoke a tool
+ *   - `ping`                  — liveness
+ *
+ * If we later need server-initiated events (notifications, sampling), we'll
+ * graduate to SSE responses; for now the protocol works as plain POST/JSON.
+ */
+
+const PROTOCOL_VERSION = "2025-06-18";
+const SERVER_INFO = {
+  name: "shared-memory",
+  version: "0.1.0",
+};
+
+type JsonRpcId = string | number | null;
+
+interface JsonRpcRequest {
+  jsonrpc: "2.0";
+  id?: JsonRpcId;
+  method: string;
+  params?: unknown;
+}
+
+interface JsonRpcSuccess {
+  jsonrpc: "2.0";
+  id: JsonRpcId;
+  result: unknown;
+}
+
+interface JsonRpcError {
+  jsonrpc: "2.0";
+  id: JsonRpcId;
+  error: { code: number; message: string; data?: unknown };
+}
+
+type JsonRpcResponse = JsonRpcSuccess | JsonRpcError;
+
+// JSON-RPC standard codes; MCP also defines server-error codes from -32000.
+const RPC = {
+  PARSE_ERROR: -32700,
+  INVALID_REQUEST: -32600,
+  METHOD_NOT_FOUND: -32601,
+  INVALID_PARAMS: -32602,
+  INTERNAL_ERROR: -32603,
+} as const;
+
+function makeError(id: JsonRpcId, code: number, message: string, data?: unknown): JsonRpcError {
+  return { jsonrpc: "2.0", id, error: { code, message, ...(data !== undefined && { data }) } };
+}
+
+function makeSuccess(id: JsonRpcId, result: unknown): JsonRpcSuccess {
+  return { jsonrpc: "2.0", id, result };
+}
+
+function isNotification(req: JsonRpcRequest): boolean {
+  return req.id === undefined;
+}
+
+export async function dispatchMcpMessage(
+  message: unknown,
+  ctx: UserContext,
+): Promise<JsonRpcResponse | null> {
+  if (!message || typeof message !== "object" || Array.isArray(message)) {
+    return makeError(null, RPC.INVALID_REQUEST, "request must be a JSON object");
+  }
+
+  const req = message as JsonRpcRequest;
+  if (req.jsonrpc !== "2.0" || typeof req.method !== "string") {
+    return makeError(req.id ?? null, RPC.INVALID_REQUEST, "invalid jsonrpc envelope");
+  }
+
+  const id = req.id ?? null;
+  const notification = isNotification(req);
+
+  try {
+    switch (req.method) {
+      case "initialize":
+        return makeSuccess(id, {
+          protocolVersion: PROTOCOL_VERSION,
+          capabilities: { tools: { listChanged: false } },
+          serverInfo: SERVER_INFO,
+        });
+
+      case "notifications/initialized":
+        // No response for notifications.
+        return null;
+
+      case "ping":
+        return makeSuccess(id, {});
+
+      case "tools/list":
+        return makeSuccess(id, {
+          tools: tools.map((t) => ({
+            name: t.name,
+            description: t.description,
+            inputSchema: t.inputSchema,
+          })),
+        });
+
+      case "tools/call": {
+        const params = (req.params ?? {}) as { name?: string; arguments?: unknown };
+        if (!params.name) {
+          return makeError(id, RPC.INVALID_PARAMS, "tools/call requires `name`");
+        }
+        const tool = toolMap[params.name];
+        if (!tool) {
+          return makeError(id, RPC.METHOD_NOT_FOUND, `unknown tool: ${params.name}`);
+        }
+        const result: ToolResult = await tool.handler(params.arguments ?? {}, ctx);
+        return makeSuccess(id, result);
+      }
+
+      default:
+        if (notification) return null; // ignore unknown notifications
+        return makeError(id, RPC.METHOD_NOT_FOUND, `unknown method: ${req.method}`);
+    }
+  } catch (e) {
+    const message = e instanceof Error ? e.message : "internal error";
+    return notification ? null : makeError(id, RPC.INTERNAL_ERROR, message);
+  }
+}
@@ -0,0 +1,310 @@
+import { and, desc, eq, isNull, sql } from "drizzle-orm";
+import { db } from "@/lib/db/client";
+import { memories, projects, auditLog } from "@/lib/db/schema";
+import {
+  MemoryIdInput,
+  MemoryListInput,
+  MemoryWriteInput,
+  ProjectIdentifyInput,
+} from "@shared-memory/schemas";
+import type { UserContext } from "./context";
+
+/**
+ * MCP tool definitions for v1 (Phase 1). Each tool has:
+ *   - name: dotted identifier exposed to clients
+ *   - description: shown to the model
+ *   - inputSchema: JSON Schema for the arguments object
+ *   - handler: async function that runs the tool
+ *
+ * Search (memory.search) and snippets come in later phases.
+ */
+
+export interface ToolResult {
+  content: Array<{ type: "text"; text: string }>;
+  isError?: boolean;
+  structuredContent?: unknown;
+}
+
+export interface ToolDef {
+  name: string;
+  description: string;
+  inputSchema: Record<string, unknown>;
+  handler: (args: unknown, ctx: UserContext) => Promise<ToolResult>;
+}
+
+// ---------- helpers ----------
+
+function ok(structured: unknown, summary: string): ToolResult {
+  return {
+    content: [{ type: "text", text: summary }],
+    structuredContent: structured,
+  };
+}
+
+function err(message: string): ToolResult {
+  return {
+    content: [{ type: "text", text: `error: ${message}` }],
+    isError: true,
+  };
+}
+
+async function resolveProjectId(
+  ctx: UserContext,
+  projectKey: string | undefined,
+): Promise<string | null> {
+  if (!projectKey) return null;
+  const row = await db
+    .select({ id: projects.id })
+    .from(projects)
+    .where(and(eq(projects.userId, ctx.userId), eq(projects.key, projectKey)))
+    .limit(1);
+  return row[0]?.id ?? null;
+}
+
+// ---------- tools ----------
+
+const projectIdentify: ToolDef = {
+  name: "project.identify",
+  description:
+    "Register or look up a project for this user by its stable key. Returns the project's internal ID and display name. Call once per session before writing project-scoped memories.",
+  inputSchema: {
+    type: "object",
+    properties: {
+      key: {
+        type: "string",
+        description:
+          "Stable project identifier. Recommended: repo name, repo URL, or any string the caller can reproduce across sessions.",
+      },
+      display_name: {
+        type: "string",
+        description: "Human-readable name shown in the Web UI. Optional.",
+      },
+    },
+    required: ["key"],
+  },
+  async handler(args, ctx) {
+    const parsed = ProjectIdentifyInput.safeParse(args);
+    if (!parsed.success) return err(parsed.error.message);
+
+    const row = await db
+      .insert(projects)
+      .values({
+        userId: ctx.userId,
+        key: parsed.data.key,
+        displayName: parsed.data.display_name ?? null,
+      })
+      .onConflictDoUpdate({
+        target: [projects.userId, projects.key],
+        set: {
+          displayName: parsed.data.display_name ?? sql`${projects.displayName}`,
+          updatedAt: new Date(),
+        },
+      })
+      .returning({
+        id: projects.id,
+        key: projects.key,
+        displayName: projects.displayName,
+        createdAt: projects.createdAt,
+      });
+
+    const p = row[0]!;
+    return ok(p, `project ${p.key} (${p.id})`);
+  },
+};
+
+const memoryWrite: ToolDef = {
+  name: "memory.write",
+  description:
+    "Persist a memory for this user. With scope='project' (default), the memory is attached to the named project. With scope='user', it's a user-global memory shared across all projects.",
+  inputSchema: {
+    type: "object",
+    properties: {
+      content: { type: "string", description: "Memory content (1–64,000 chars)." },
+      project: {
+        type: "string",
+        description: "Project key (required when scope='project').",
+      },
+      scope: {
+        type: "string",
+        enum: ["project", "user"],
+        description: "Scope of the memory. Defaults to 'project'.",
+      },
+      tags: {
+        type: "array",
+        items: { type: "string" },
+        description: "Optional tags for filtering/grouping.",
+      },
+    },
+    required: ["content"],
+  },
+  async handler(args, ctx) {
+    const parsed = MemoryWriteInput.safeParse(args);
+    if (!parsed.success) return err(parsed.error.message);
+
+    const scope = parsed.data.scope;
+    let projectId: string | null = null;
+    if (scope === "project") {
+      if (!parsed.data.project) return err("scope=project requires `project` key");
+      projectId = await resolveProjectId(ctx, parsed.data.project);
+      if (!projectId) {
+        return err(`unknown project '${parsed.data.project}'; call project.identify first`);
+      }
+    }
+
+    const inserted = await db
+      .insert(memories)
+      .values({
+        userId: ctx.userId,
+        projectId,
+        scope,
+        content: parsed.data.content,
+        tags: parsed.data.tags ?? [],
+      })
+      .returning({ id: memories.id, createdAt: memories.createdAt });
+
+    const m = inserted[0]!;
+    await db.insert(auditLog).values({
+      userId: ctx.userId,
+      actor: "mcp",
+      action: "memory.write",
+      entityType: "memory",
+      entityId: m.id,
+      payload: { scope, projectKey: parsed.data.project ?? null, tags: parsed.data.tags ?? [] },
+    });
+
+    return ok({ id: m.id, createdAt: m.createdAt }, `wrote memory ${m.id}`);
+  },
+};
+
+const memoryList: ToolDef = {
+  name: "memory.list",
+  description:
+    "List memories for this user, most recent first. Filter by project key and/or scope. Phase 2 will add memory.search for semantic + full-text lookup.",
+  inputSchema: {
+    type: "object",
+    properties: {
+      project: { type: "string", description: "Filter by project key." },
+      scope: { type: "string", enum: ["project", "user"], description: "Filter by scope." },
+      tags: {
+        type: "array",
+        items: { type: "string" },
+        description: "Require all of these tags.",
+      },
+      limit: { type: "integer", minimum: 1, maximum: 200, default: 50 },
+    },
+  },
+  async handler(args, ctx) {
+    const parsed = MemoryListInput.safeParse(args);
+    if (!parsed.success) return err(parsed.error.message);
+
+    const where = [eq(memories.userId, ctx.userId), isNull(memories.deletedAt)];
+
+    if (parsed.data.scope) where.push(eq(memories.scope, parsed.data.scope));
+
+    if (parsed.data.project) {
+      const projectId = await resolveProjectId(ctx, parsed.data.project);
+      if (!projectId) return ok({ items: [], next_cursor: null }, "0 results");
+      where.push(eq(memories.projectId, projectId));
+    }
+
+    if (parsed.data.tags && parsed.data.tags.length > 0) {
+      where.push(sql`${memories.tags} @> ${parsed.data.tags}::text[]`);
+    }
+
+    const rows = await db
+      .select({
+        id: memories.id,
+        scope: memories.scope,
+        projectId: memories.projectId,
+        content: memories.content,
+        tags: memories.tags,
+        createdAt: memories.createdAt,
+        updatedAt: memories.updatedAt,
+      })
+      .from(memories)
+      .where(and(...where))
+      .orderBy(desc(memories.createdAt))
+      .limit(parsed.data.limit);
+
+    return ok({ items: rows, next_cursor: null }, `${rows.length} result(s)`);
+  },
+};
+
+const memoryGet: ToolDef = {
+  name: "memory.get",
+  description: "Fetch a single memory by its UUID.",
+  inputSchema: {
+    type: "object",
+    properties: { id: { type: "string", format: "uuid" } },
+    required: ["id"],
+  },
+  async handler(args, ctx) {
+    const parsed = MemoryIdInput.safeParse(args);
+    if (!parsed.success) return err(parsed.error.message);
+
+    const row = await db
+      .select()
+      .from(memories)
+      .where(
+        and(
+          eq(memories.id, parsed.data.id),
+          eq(memories.userId, ctx.userId),
+          isNull(memories.deletedAt),
+        ),
+      )
+      .limit(1);
+
+    if (!row[0]) return err("not found");
+    return ok(row[0], `memory ${row[0].id}`);
+  },
+};
+
+const memoryDelete: ToolDef = {
+  name: "memory.delete",
+  description: "Soft-delete a memory (sets deleted_at; preserved for audit).",
+  inputSchema: {
+    type: "object",
+    properties: { id: { type: "string", format: "uuid" } },
+    required: ["id"],
+  },
+  async handler(args, ctx) {
+    const parsed = MemoryIdInput.safeParse(args);
+    if (!parsed.success) return err(parsed.error.message);
+
+    const updated = await db
+      .update(memories)
+      .set({ deletedAt: new Date() })
+      .where(
+        and(
+          eq(memories.id, parsed.data.id),
+          eq(memories.userId, ctx.userId),
+          isNull(memories.deletedAt),
+        ),
+      )
+      .returning({ id: memories.id });
+
+    if (!updated[0]) return err("not found");
+
+    await db.insert(auditLog).values({
+      userId: ctx.userId,
+      actor: "mcp",
+      action: "memory.delete",
+      entityType: "memory",
+      entityId: updated[0].id,
+    });
+
+    return ok({ id: updated[0].id, deleted: true }, `deleted memory ${updated[0].id}`);
+  },
+};
+
+export const tools: ToolDef[] = [
+  projectIdentify,
+  memoryWrite,
+  memoryList,
+  memoryGet,
+  memoryDelete,
+];
+
+export const toolMap: Record<string, ToolDef> = Object.fromEntries(
+  tools.map((t) => [t.name, t]),
+);