feat(mcp): oauth schema + minimum admin form

Adds the data model and admin UI surface required by the OAuth-MCP flow.
Phase 2 of the per-user delegation initiative.

Schema:
- migration 049 creates mcp_user_tokens (PK user_id, server_name) and
  mcp_oauth_pending (PK state, indexed by created_at)
- eight new columns on mcp_servers: auth_type ('none' / 'static' /
  'oauth_user', NOT NULL DEFAULT 'static') plus six oauth_* config
  fields and oauth_as_issuer_cached
- post-upgrade UPDATE normalises auth_type to 'none' for streamable-http
  rows whose headers are NULL/empty/'{}'; stdio rows are left at the
  'static' default (auth_type is HTTP-auth-only)
- _schema.py kept in lockstep with the migration so metadata.create_all
  and alembic upgrade produce identical shapes
- mcp_user_tokens / mcp_oauth_pending TypedDicts in _protocol.py for
  Phase 3/4 use (no CRUD methods yet)

Storage / API:
- create_mcp_server gains the eight kwargs across protocol + sqlite +
  postgresql
- MCP_SERVER_MUTABLE picks up auth_type and the six text oauth_* fields;
  oauth_client_secret_ct is intentionally NOT in the whitelist — Phase 3
  will own ciphertext writes via a dedicated method
- McpServerInfo + Create/Update Pydantic schemas extended; oauth_client_secret
  accepted as plaintext input but discarded (Phase 3 wires encryption)

Admin handlers:
- _parse_auth_type validates against {'none', 'static', 'oauth_user'} and
  rejects empty / unknown values; shared between create and update
- when auth_type changes away from 'oauth_user', the oauth_* config
  columns are explicitly nulled in the same UPDATE so the row stays
  consistent
- _clean_oauth_text caps text fields at 512 chars (URLs at 2048) to bound
  admin write surface
- _mask_mcp_secrets now masks oauth_client_secret_ct to '***' regardless
  of reveal=true (write-only field)
- audit detail dict redacts oauth_client_secret if present

Frontend:
- new "Multitenant Authorization" fieldset on the MCP-server modal with
  three radio buttons (None / Shared / Per-user OAuth 2.1)
- conditional OAuth subform: AS URL, registration mode (preregistered /
  dcr; cimd is future), client ID, client secret, scopes, audience
- secret input is autocomplete=off and never round-trips on edit
- audience auto-populates from the MCP server URL on blur
- headers textarea hidden and submitted as {} when auth_type is 'none' or
  'oauth_user' so flipping the radio cleans up server-side state

Tests: storage round-trip for the new columns, oauth_pending table smoke,
migration 049 upgrade/downgrade with stdio-vs-http normalisation, four
admin-API tests for auth_type validation and oauth_*-clear-on-flip-away.
Suite passes 5284 (matched pre-Phase-2 baseline 5267 + 17 new).

Stacks on Phase 0; no behavioural change for existing rows.

tests/test_mcp_admin_api.py

@@ -319,6 +319,49 @@ class TestCreateMcpServer:
         assert r.status_code == 400
         assert "name" in r.json()["error"].lower()
 
+    def test_admin_create_oauth_server(self, client):
+        """Phase 2: admin can POST a server with auth_type=oauth_user
+        and the seven OAuth text fields round-trip via GET."""
+        r = client.post(
+            "/v1/api/admin/mcp-servers",
+            json={
+                "name": "oauth-srv",
+                "transport": "streamable-http",
+                "url": "https://mcp.example.com/sse",
+                "auth_type": "oauth_user",
+                "oauth_client_id": "cli_abc",
+                "oauth_client_secret": "should-be-discarded",
+                "oauth_scopes": "openid profile",
+                "oauth_audience": "https://mcp.example.com",
+                "oauth_registration_mode": "preregistered",
+                "oauth_authorization_server_url": "https://auth.example.com",
+            },
+        )
+        assert r.status_code == 200, r.text
+        data = r.json()
+        assert data["auth_type"] == "oauth_user"
+        assert data["oauth_client_id"] == "cli_abc"
+        assert data["oauth_scopes"] == "openid profile"
+        assert data["oauth_audience"] == "https://mcp.example.com"
+        assert data["oauth_registration_mode"] == "preregistered"
+        assert data["oauth_authorization_server_url"] == "https://auth.example.com"
+        # Phase 2 discards the plaintext secret — ciphertext stays NULL,
+        # masked to None in the response.
+        assert data["oauth_client_secret_ct"] is None
+
+    def test_create_invalid_auth_type(self, client):
+        r = client.post(
+            "/v1/api/admin/mcp-servers",
+            json={
+                "name": "bad-auth",
+                "transport": "stdio",
+                "command": "x",
+                "auth_type": "magic",
+            },
+        )
+        assert r.status_code == 400
+        assert "auth_type" in r.json()["error"].lower()
+
 
 # ---------------------------------------------------------------------------
 # Get single
@@ -401,6 +444,130 @@ class TestUpdateMcpServer:
         assert r.status_code == 400
         assert "transport" in r.json()["error"].lower()
 
+    def test_admin_update_auth_type_static_to_oauth(self, client):
+        """Phase 2: an existing static row can be flipped to oauth_user
+        with OAuth fields supplied alongside."""
+        created = _create_server(
+            client,
+            name="flip-to-oauth",
+            transport="streamable-http",
+            url="http://mcp.example.com/sse",
+        )
+        sid = created["server_id"]
+        assert created["auth_type"] == "static"
+
+        r = client.put(
+            f"/v1/api/admin/mcp-servers/{sid}",
+            json={
+                "auth_type": "oauth_user",
+                "oauth_client_id": "cli_xyz",
+                "oauth_audience": "https://mcp.example.com",
+                "oauth_registration_mode": "dcr",
+            },
+        )
+        assert r.status_code == 200, r.text
+        data = r.json()
+        assert data["auth_type"] == "oauth_user"
+        assert data["oauth_client_id"] == "cli_xyz"
+        assert data["oauth_audience"] == "https://mcp.example.com"
+        assert data["oauth_registration_mode"] == "dcr"
+
+    def test_update_invalid_auth_type(self, client):
+        created = _create_server(client, name="bad-auth-update")
+        sid = created["server_id"]
+        r = client.put(
+            f"/v1/api/admin/mcp-servers/{sid}",
+            json={"auth_type": "wat"},
+        )
+        assert r.status_code == 400
+        assert "auth_type" in r.json()["error"].lower()
+
+    def test_update_empty_auth_type_rejected(self, client):
+        """Empty-string auth_type is rejected (no silent coercion to 'static')."""
+        created = _create_server(client, name="empty-auth-update")
+        sid = created["server_id"]
+        r = client.put(
+            f"/v1/api/admin/mcp-servers/{sid}",
+            json={"auth_type": ""},
+        )
+        assert r.status_code == 400
+        assert "auth_type" in r.json()["error"].lower()
+
+    def test_create_empty_auth_type_rejected(self, client):
+        """Empty-string auth_type on create is rejected too."""
+        r = client.post(
+            "/v1/api/admin/mcp-servers",
+            json={
+                "name": "empty-auth-create",
+                "transport": "stdio",
+                "command": "x",
+                "auth_type": "",
+            },
+        )
+        assert r.status_code == 400
+        assert "auth_type" in r.json()["error"].lower()
+
+    def test_update_auth_type_oauth_to_static_clears_oauth_fields(self, client):
+        """Flipping auth_type away from oauth_user clears the oauth_* text
+        columns so a stale client_id / audience can't leak back."""
+        # Seed an oauth_user row with all fields populated.
+        r = client.post(
+            "/v1/api/admin/mcp-servers",
+            json={
+                "name": "flip-away",
+                "transport": "streamable-http",
+                "url": "https://mcp.example.com/sse",
+                "auth_type": "oauth_user",
+                "oauth_client_id": "cli_seed",
+                "oauth_scopes": "openid",
+                "oauth_audience": "https://mcp.example.com",
+                "oauth_registration_mode": "preregistered",
+                "oauth_authorization_server_url": "https://auth.example.com",
+            },
+        )
+        assert r.status_code == 200, r.text
+        sid = r.json()["server_id"]
+
+        # Flip to static — server should clear all oauth_* text fields.
+        r = client.put(
+            f"/v1/api/admin/mcp-servers/{sid}",
+            json={"auth_type": "static"},
+        )
+        assert r.status_code == 200, r.text
+        data = r.json()
+        assert data["auth_type"] == "static"
+        assert data["oauth_client_id"] is None
+        assert data["oauth_scopes"] is None
+        assert data["oauth_audience"] is None
+        assert data["oauth_registration_mode"] is None
+        assert data["oauth_authorization_server_url"] is None
+
+    def test_update_auth_type_oauth_to_none_clears_oauth_fields(self, client):
+        """Same clear behavior when flipping to 'none'."""
+        r = client.post(
+            "/v1/api/admin/mcp-servers",
+            json={
+                "name": "flip-to-none",
+                "transport": "streamable-http",
+                "url": "https://mcp.example.com/sse",
+                "auth_type": "oauth_user",
+                "oauth_client_id": "cli_seed2",
+                "oauth_audience": "https://mcp.example.com",
+            },
+        )
+        assert r.status_code == 200, r.text
+        sid = r.json()["server_id"]
+
+        r = client.put(
+            f"/v1/api/admin/mcp-servers/{sid}",
+            json={"auth_type": "none"},
+        )
+        assert r.status_code == 200, r.text
+        data = r.json()
+        assert data["auth_type"] == "none"
+        assert data["oauth_client_id"] is None
+        assert data["oauth_audience"] is None
+
 
 # ---------------------------------------------------------------------------
 # Delete

tests/test_mcp_oauth_pending_storage.py

@@ -0,0 +1,133 @@
+"""Smoke tests for the new OAuth-MCP storage tables.
+
+Phase 2 only adds the schema — token CRUD lands in Phase 3 and pending-
+state CRUD in Phase 4.  These tests verify the tables exist after
+``init_storage`` and accept the documented row shape via raw SQL.
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+
+from turnstone.core.storage._schema import mcp_oauth_pending, mcp_user_tokens
+
+
+class TestMcpUserTokensTable:
+    def test_table_exists_and_accepts_row(self, backend) -> None:
+        with backend._engine.connect() as conn:
+            conn.execute(
+                sa.insert(mcp_user_tokens),
+                {
+                    "user_id": "u1",
+                    "server_name": "srv-a",
+                    "access_token_ct": b"\x00ciphertext-a",
+                    "refresh_token_ct": b"\x00ciphertext-r",
+                    "expires_at": "2026-05-04T12:00:00",
+                    "scopes": "openid profile",
+                    "as_issuer": "https://auth.example.com",
+                    "audience": "https://mcp.example.com",
+                    "created": "2026-05-04T11:00:00",
+                    "last_refreshed": None,
+                },
+            )
+            conn.commit()
+            row = conn.execute(
+                sa.select(mcp_user_tokens).where(
+                    (mcp_user_tokens.c.user_id == "u1") & (mcp_user_tokens.c.server_name == "srv-a")
+                )
+            ).one()
+        assert row.access_token_ct == b"\x00ciphertext-a"
+        assert row.refresh_token_ct == b"\x00ciphertext-r"
+        assert row.scopes == "openid profile"
+        assert row.audience == "https://mcp.example.com"
+
+    def test_composite_pk_distinguishes_user_server(self, backend) -> None:
+        """Same user, different server => two rows; same (user, server) => conflict."""
+        with backend._engine.connect() as conn:
+            conn.execute(
+                sa.insert(mcp_user_tokens),
+                [
+                    {
+                        "user_id": "u1",
+                        "server_name": "srv-a",
+                        "access_token_ct": b"a",
+                        "refresh_token_ct": None,
+                        "expires_at": None,
+                        "scopes": None,
+                        "as_issuer": "https://auth.example.com",
+                        "audience": "https://a.example.com",
+                        "created": "2026-05-04T11:00:00",
+                        "last_refreshed": None,
+                    },
+                    {
+                        "user_id": "u1",
+                        "server_name": "srv-b",
+                        "access_token_ct": b"b",
+                        "refresh_token_ct": None,
+                        "expires_at": None,
+                        "scopes": None,
+                        "as_issuer": "https://auth.example.com",
+                        "audience": "https://b.example.com",
+                        "created": "2026-05-04T11:00:00",
+                        "last_refreshed": None,
+                    },
+                ],
+            )
+            conn.commit()
+            count = conn.execute(sa.select(sa.func.count()).select_from(mcp_user_tokens)).scalar()
+        assert count == 2
+
+
+class TestMcpOauthPendingTable:
+    def test_table_exists_and_accepts_row(self, backend) -> None:
+        with backend._engine.connect() as conn:
+            conn.execute(
+                sa.insert(mcp_oauth_pending),
+                {
+                    "state": "rand-state-xyz",
+                    "user_id": "u1",
+                    "server_name": "srv-a",
+                    "code_verifier": "verifier-blob",
+                    "return_url": "/admin/mcp-servers",
+                    "created_at": "2026-05-04T11:00:00",
+                },
+            )
+            conn.commit()
+            row = conn.execute(
+                sa.select(mcp_oauth_pending).where(mcp_oauth_pending.c.state == "rand-state-xyz")
+            ).one()
+        assert row.user_id == "u1"
+        assert row.server_name == "srv-a"
+        assert row.return_url == "/admin/mcp-servers"
+
+    def test_state_pk_unique(self, backend) -> None:
+        """A second insert with the same state value raises IntegrityError."""
+        with backend._engine.connect() as conn:
+            conn.execute(
+                sa.insert(mcp_oauth_pending),
+                {
+                    "state": "dup-state",
+                    "user_id": "u1",
+                    "server_name": "srv-a",
+                    "code_verifier": "v",
+                    "return_url": "/x",
+                    "created_at": "2026-05-04T11:00:00",
+                },
+            )
+            conn.commit()
+        import pytest
+        from sqlalchemy.exc import IntegrityError
+
+        with pytest.raises(IntegrityError), backend._engine.connect() as conn:
+            conn.execute(
+                sa.insert(mcp_oauth_pending),
+                {
+                    "state": "dup-state",
+                    "user_id": "u2",
+                    "server_name": "srv-b",
+                    "code_verifier": "v",
+                    "return_url": "/y",
+                    "created_at": "2026-05-04T11:01:00",
+                },
+            )
+            conn.commit()

tests/test_migration_049.py

@@ -0,0 +1,166 @@
+"""Tests for alembic migration 049 (OAuth-MCP schema).
+
+Drives ``command.upgrade`` from a programmatic Alembic config against
+an isolated SQLite database per test, then asserts:
+
+* the two new tables (``mcp_user_tokens``, ``mcp_oauth_pending``) exist,
+* the eight new ``mcp_servers`` columns exist,
+* the post-upgrade ``UPDATE mcp_servers`` normalization rewrites rows
+  with empty / missing headers to ``auth_type='none'`` while leaving
+  rows with non-empty headers at ``auth_type='static'``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import sqlalchemy as sa
+from alembic import command
+from alembic.config import Config
+
+_MIGRATIONS_DIR = str(
+    Path(__file__).resolve().parent.parent / "turnstone" / "core" / "storage" / "migrations"
+)
+
+
+def _alembic_cfg(db_path: Path) -> Config:
+    cfg = Config()
+    cfg.set_main_option("script_location", _MIGRATIONS_DIR)
+    cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
+    return cfg
+
+
+class TestMigration049:
+    def test_creates_new_tables_and_columns(self, tmp_path: Path) -> None:
+        db_path = tmp_path / "049.db"
+        cfg = _alembic_cfg(db_path)
+
+        # Walk forward through 048 first, then explicitly to 049 so we
+        # exercise the *upgrade* function (not just the schema's `head`).
+        command.upgrade(cfg, "048")
+        command.upgrade(cfg, "049")
+
+        engine = sa.create_engine(f"sqlite:///{db_path}")
+        try:
+            inspector = sa.inspect(engine)
+            tables = set(inspector.get_table_names())
+            assert "mcp_user_tokens" in tables
+            assert "mcp_oauth_pending" in tables
+
+            mcp_cols = {c["name"] for c in inspector.get_columns("mcp_servers")}
+            new_cols = {
+                "auth_type",
+                "oauth_client_id",
+                "oauth_client_secret_ct",
+                "oauth_scopes",
+                "oauth_audience",
+                "oauth_registration_mode",
+                "oauth_authorization_server_url",
+                "oauth_as_issuer_cached",
+            }
+            assert new_cols.issubset(mcp_cols), new_cols - mcp_cols
+
+            # Index check on mcp_oauth_pending.
+            indexes = {ix["name"] for ix in inspector.get_indexes("mcp_oauth_pending")}
+            assert "idx_mcp_pending_created" in indexes
+        finally:
+            engine.dispose()
+
+    def test_normalizes_empty_headers_to_none(self, tmp_path: Path) -> None:
+        """Streamable-http rows with NULL / '' / '{}' headers become
+        auth_type='none'; rows with non-empty headers stay 'static'.
+        Stdio rows always stay 'static' regardless of headers — the
+        column value is opaque when there is no HTTP transport."""
+        db_path = tmp_path / "049-norm.db"
+        cfg = _alembic_cfg(db_path)
+
+        # Apply everything up to 048, seed rows, then apply 049.
+        command.upgrade(cfg, "048")
+
+        engine = sa.create_engine(f"sqlite:///{db_path}")
+        try:
+            with engine.begin() as conn:
+                conn.execute(
+                    sa.text(
+                        """
+                        INSERT INTO mcp_servers (
+                            server_id, name, transport, command, args, url,
+                            headers, env, auto_approve, enabled, created_by,
+                            registry_name, registry_version, registry_meta,
+                            created, updated
+                        ) VALUES (
+                            :sid, :name, :transport, '', '[]',
+                            'https://x', :headers, '{}', 0, 1, '', NULL, '',
+                            '{}', '2026-05-04T11:00:00', '2026-05-04T11:00:00'
+                        )
+                        """
+                    ),
+                    [
+                        {
+                            "sid": "s-empty-str",
+                            "name": "empty-str",
+                            "transport": "streamable-http",
+                            "headers": "",
+                        },
+                        {
+                            "sid": "s-empty-obj",
+                            "name": "empty-obj",
+                            "transport": "streamable-http",
+                            "headers": "{}",
+                        },
+                        {
+                            "sid": "s-with-headers",
+                            "name": "with-headers",
+                            "transport": "streamable-http",
+                            "headers": '{"Authorization":"Bearer x"}',
+                        },
+                        # Stdio rows must keep the 'static' default, even
+                        # though their headers are empty — auth_type is
+                        # opaque for stdio.
+                        {
+                            "sid": "s-stdio-empty",
+                            "name": "stdio-empty",
+                            "transport": "stdio",
+                            "headers": "{}",
+                        },
+                        {
+                            "sid": "s-stdio-null",
+                            "name": "stdio-null",
+                            "transport": "stdio",
+                            "headers": "",
+                        },
+                    ],
+                )
+
+            command.upgrade(cfg, "049")
+
+            with engine.connect() as conn:
+                rows = dict(conn.execute(sa.text("SELECT name, auth_type FROM mcp_servers")).all())
+            assert rows["empty-str"] == "none"
+            assert rows["empty-obj"] == "none"
+            assert rows["with-headers"] == "static"
+            # Stdio rows must remain at the 'static' column default even
+            # when headers are empty — the migration only touches HTTP
+            # rows where auth_type is semantically meaningful.
+            assert rows["stdio-empty"] == "static"
+            assert rows["stdio-null"] == "static"
+        finally:
+            engine.dispose()
+
+    def test_full_chain_to_head(self, tmp_path: Path) -> None:
+        """Sanity: running ``upgrade head`` on a fresh DB yields the
+        same end-state column set as ``_schema.metadata``."""
+        db_path = tmp_path / "049-head.db"
+        cfg = _alembic_cfg(db_path)
+        command.upgrade(cfg, "head")
+
+        engine = sa.create_engine(f"sqlite:///{db_path}")
+        try:
+            from turnstone.core.storage._schema import mcp_servers
+
+            inspector = sa.inspect(engine)
+            actual = {c["name"] for c in inspector.get_columns("mcp_servers")}
+            expected = {c.name for c in mcp_servers.columns}
+            assert expected.issubset(actual), expected - actual
+        finally:
+            engine.dispose()

tests/test_storage_sqlite.py

@@ -939,6 +939,64 @@ class TestTouchWorkstream:
         backend.touch_workstream("nonexistent")  # must not raise
 
 
+# -- MCP OAuth columns ---------------------------------------------------------
+
+
+class TestMcpServerOauthColumns:
+    def test_mcp_servers_oauth_columns_round_trip(self, backend: Any) -> None:
+        """An oauth_user row round-trips through create -> get with all
+        seven OAuth text columns intact."""
+        sid = "oauth-srv-1"
+        backend.create_mcp_server(
+            server_id=sid,
+            name="oauth-srv",
+            transport="streamable-http",
+            url="https://mcp.example.com/sse",
+            auth_type="oauth_user",
+            oauth_client_id="cli_abc123",
+            oauth_scopes="openid profile",
+            oauth_audience="https://mcp.example.com",
+            oauth_registration_mode="preregistered",
+            oauth_authorization_server_url="https://auth.example.com",
+            oauth_as_issuer_cached="https://auth.example.com",
+        )
+        s = backend.get_mcp_server(sid)
+        assert s is not None
+        assert s["auth_type"] == "oauth_user"
+        assert s["oauth_client_id"] == "cli_abc123"
+        assert s["oauth_scopes"] == "openid profile"
+        assert s["oauth_audience"] == "https://mcp.example.com"
+        assert s["oauth_registration_mode"] == "preregistered"
+        assert s["oauth_authorization_server_url"] == "https://auth.example.com"
+        assert s["oauth_as_issuer_cached"] == "https://auth.example.com"
+        # Phase 2 leaves the ciphertext slot NULL even when other oauth
+        # fields are populated; Phase 3 wires the encryption write path.
+        assert s["oauth_client_secret_ct"] is None
+
+    def test_update_auth_type_static_to_oauth(self, backend: Any) -> None:
+        sid = "oauth-srv-2"
+        backend.create_mcp_server(
+            server_id=sid,
+            name="static-then-oauth",
+            transport="streamable-http",
+            url="https://mcp.example.com/sse",
+        )
+        assert backend.get_mcp_server(sid)["auth_type"] == "static"
+
+        ok = backend.update_mcp_server(
+            sid,
+            auth_type="oauth_user",
+            oauth_client_id="cli_after",
+            oauth_audience="https://mcp.example.com",
+        )
+        assert ok is True
+        s = backend.get_mcp_server(sid)
+        assert s is not None
+        assert s["auth_type"] == "oauth_user"
+        assert s["oauth_client_id"] == "cli_after"
+        assert s["oauth_audience"] == "https://mcp.example.com"
+
+
 # -- Lifecycle -----------------------------------------------------------------
 
 

turnstone/api/console_schemas.py

@@ -674,6 +674,16 @@ class McpServerInfo(BaseModel):
     registry_name: str | None = None
     registry_version: str = ""
     registry_meta: str = "{}"
+    auth_type: str = "static"
+    oauth_client_id: str | None = None
+    oauth_scopes: str | None = None
+    oauth_audience: str | None = None
+    oauth_registration_mode: str | None = None
+    oauth_authorization_server_url: str | None = None
+    oauth_as_issuer_cached: str | None = None
+    # Fernet ciphertext; never decrypted on the read path.  Responses
+    # carry the masked ``"***"`` sentinel via ``_mask_mcp_secrets``.
+    oauth_client_secret_ct: str | None = None
     created: str
     updated: str
 
@@ -704,6 +714,16 @@ class CreateMcpServerRequest(BaseModel):
     env: dict[str, str] = Field(default_factory=dict)
     auto_approve: bool = False
     enabled: bool = True
+    # OAuth-MCP: one of 'none' | 'static' | 'oauth_user'.
+    # ``oauth_client_secret`` is plaintext input; never persisted,
+    # redacted in audit log.
+    auth_type: str = "static"
+    oauth_client_id: str | None = None
+    oauth_client_secret: str | None = None
+    oauth_scopes: str | None = None
+    oauth_audience: str | None = None
+    oauth_registration_mode: str | None = None
+    oauth_authorization_server_url: str | None = None
 
 
 class UpdateMcpServerRequest(BaseModel):
@@ -716,6 +736,13 @@ class UpdateMcpServerRequest(BaseModel):
     env: dict[str, str] | None = None
     auto_approve: bool | None = None
     enabled: bool | None = None
+    auth_type: str | None = None
+    oauth_client_id: str | None = None
+    oauth_client_secret: str | None = None
+    oauth_scopes: str | None = None
+    oauth_audience: str | None = None
+    oauth_registration_mode: str | None = None
+    oauth_authorization_server_url: str | None = None
 
 
 class ListMcpServersResponse(BaseModel):

turnstone/console/server.py

@@ -7965,6 +7965,42 @@ async def admin_registry_install(request: Request) -> JSONResponse:
 
 _MCP_NAME_RE = re.compile(r"^[a-zA-Z0-9._-]+$")
 _MCP_MAX_SERVERS = 200  # fallback; prefer cluster.mcp_max_servers from storage
+_MCP_AUTH_TYPES = frozenset({"none", "static", "oauth_user"})
+
+
+def _clean_oauth_text(value: Any, *, max_length: int = 512) -> str | None:
+    """Normalize an admin form OAuth text field — empty string -> None.
+
+    Caps the input to ``max_length`` characters to bound DB row size on
+    the admin.mcp write path.  Pass a larger ``max_length`` (e.g. 2048)
+    for URL fields where the default would otherwise truncate valid
+    long URLs.
+    """
+    if value is None:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    return text[:max_length]
+
+
+def _parse_auth_type(body: dict[str, Any]) -> tuple[str | None, JSONResponse | None]:
+    """Validate ``auth_type`` from a request body.
+
+    Returns ``(value, None)`` for a valid value, ``(None, error)`` for
+    a present-but-invalid value (caller returns ``error``), or
+    ``(None, None)`` when ``auth_type`` is absent (caller skips the
+    update / falls back to a default).
+    """
+    if "auth_type" not in body:
+        return None, None
+    auth_type = str(body["auth_type"]).strip()
+    if auth_type not in _MCP_AUTH_TYPES:
+        return None, JSONResponse(
+            {"error": "auth_type must be 'none', 'static', or 'oauth_user'"},
+            status_code=400,
+        )
+    return auth_type, None
 
 
 def _get_mcp_max_servers(request: Request) -> int:
@@ -7976,10 +8012,20 @@ def _get_mcp_max_servers(request: Request) -> int:
 
 
 def _mask_mcp_secrets(server: dict[str, Any], reveal: bool = False) -> dict[str, Any]:
-    """Replace env/headers values with '***' unless reveal is True."""
-    if reveal:
-        return server
+    """Mask secret fields on an MCP server response dict.
+
+    ``env`` and ``headers`` are masked only when ``reveal`` is False.
+    ``oauth_client_secret_ct`` is always masked regardless of ``reveal``
+    — it's a write-only field (the admin form accepts plaintext but
+    the response just signals presence-or-absence as ``"***"`` /
+    ``None``).
+    """
     s = dict(server)
+    # OAuth client secret ciphertext is write-only at every read path.
+    raw_secret = s.get("oauth_client_secret_ct")
+    s["oauth_client_secret_ct"] = "***" if raw_secret is not None else None
+    if reveal:
+        return s
     if s.get("env") and s["env"] != "{}":
         try:
             env_dict = json.loads(s["env"]) if isinstance(s["env"], str) else s["env"]
@@ -8100,6 +8146,7 @@ async def admin_list_mcp_servers(request: Request) -> JSONResponse:
                 "auto_approve": False,
                 "enabled": True,
                 "created_by": "",
+                "auth_type": "static",
                 "created": "",
                 "updated": "",
                 "source": "config",
@@ -8155,6 +8202,12 @@ async def admin_create_mcp_server(request: Request) -> JSONResponse:
             {"error": "url is required for streamable-http transport"}, status_code=400
         )
 
+    auth_type_value, err_resp = _parse_auth_type(body)
+    if err_resp is not None:
+        return err_resp
+    # Helper returns None when key is absent — fall back to the default.
+    auth_type = auth_type_value if auth_type_value is not None else "static"
+
     # Check max servers
     existing = storage.list_mcp_servers()
     max_servers = _get_mcp_max_servers(request)
@@ -8190,15 +8243,26 @@ async def admin_create_mcp_server(request: Request) -> JSONResponse:
         auto_approve=bool(body.get("auto_approve", False)),
         enabled=bool(body.get("enabled", True)),
         created_by=audit_uid,
+        auth_type=auth_type,
+        oauth_client_id=_clean_oauth_text(body.get("oauth_client_id")),
+        oauth_scopes=_clean_oauth_text(body.get("oauth_scopes")),
+        oauth_audience=_clean_oauth_text(body.get("oauth_audience"), max_length=2048),
+        oauth_registration_mode=_clean_oauth_text(body.get("oauth_registration_mode")),
+        oauth_authorization_server_url=_clean_oauth_text(
+            body.get("oauth_authorization_server_url"), max_length=2048
+        ),
     )
 
+    audit_detail: dict[str, Any] = {"name": name, "auth_type": auth_type}
+    if "oauth_client_secret" in body:
+        audit_detail["oauth_client_secret"] = "(redacted)"
     record_audit(
         storage,
         audit_uid,
         "mcp_server.create",
         "mcp_server",
         server_id,
-        {"name": name},
+        audit_detail,
         ip,
     )
 
@@ -8302,6 +8366,41 @@ async def admin_update_mcp_server(request: Request) -> JSONResponse:
         updates["auto_approve"] = bool(body["auto_approve"])
     if "enabled" in body:
         updates["enabled"] = bool(body["enabled"])
+    auth_type_value, err_resp = _parse_auth_type(body)
+    if err_resp is not None:
+        return err_resp
+    if auth_type_value is not None:
+        updates["auth_type"] = auth_type_value
+    for _oauth_key in (
+        "oauth_client_id",
+        "oauth_scopes",
+        "oauth_registration_mode",
+    ):
+        if _oauth_key in body:
+            updates[_oauth_key] = _clean_oauth_text(body[_oauth_key])
+    for _oauth_url_key in (
+        "oauth_audience",
+        "oauth_authorization_server_url",
+    ):
+        if _oauth_url_key in body:
+            updates[_oauth_url_key] = _clean_oauth_text(body[_oauth_url_key], max_length=2048)
+
+    # When auth_type is changed away from oauth_user, clear the OAuth
+    # columns so a stale client_id / audience can't leak back if the
+    # row is later flipped to a different oauth_user provider.
+    # ``oauth_client_secret_ct`` is owned by a dedicated write path
+    # (not the generic update); its clear-on-change lives there.
+    if updates.get("auth_type") and updates["auth_type"] != "oauth_user":
+        updates.update(
+            {
+                "oauth_client_id": None,
+                "oauth_scopes": None,
+                "oauth_audience": None,
+                "oauth_registration_mode": None,
+                "oauth_authorization_server_url": None,
+                "oauth_as_issuer_cached": None,
+            }
+        )
 
     if updates:
         storage.update_mcp_server(server_id, **updates)
@@ -8311,6 +8410,8 @@ async def admin_update_mcp_server(request: Request) -> JSONResponse:
     for _secret_key in ("env", "headers"):
         if _secret_key in audit_detail:
             audit_detail[_secret_key] = "(updated)"
+    if "oauth_client_secret" in body:
+        audit_detail["oauth_client_secret"] = "(redacted)"
     record_audit(
         storage,
         audit_uid,

turnstone/console/static/admin.js

@@ -3465,6 +3465,46 @@ function toggleMcpTransport() {
     v === "stdio" ? "" : "none";
   document.getElementById("mcp-http-fields").style.display =
     v === "streamable-http" ? "" : "none";
+  // Re-evaluate auth-field visibility because the headers row lives
+  // inside mcp-http-fields and gets toggled there.
+  toggleMcpAuthFields();
+}
+
+function _selectedMcpAuthType() {
+  var radios = document.getElementsByName("mcp-auth-type");
+  for (var i = 0; i < radios.length; i++) {
+    if (radios[i].checked) return radios[i].value;
+  }
+  return "static";
+}
+
+function toggleMcpAuthFields() {
+  var authType = _selectedMcpAuthType();
+  var oauthDiv = document.getElementById("mcp-oauth-fields");
+  if (oauthDiv) {
+    oauthDiv.style.display = authType === "oauth_user" ? "" : "none";
+  }
+  // The "Headers" textarea (inside mcp-http-fields) is only meaningful
+  // for static auth; hide it for 'none' / 'oauth_user' so operators
+  // don't accidentally configure stale credentials.
+  var headersInput = document.getElementById("mcp-headers");
+  if (headersInput) {
+    var headersLabel = document.querySelector('label[for="mcp-headers"]');
+    var show = authType === "static";
+    headersInput.style.display = show ? "" : "none";
+    if (headersLabel) headersLabel.style.display = show ? "" : "none";
+  }
+}
+
+function _wireMcpAudienceAutofill() {
+  // Idempotent — only attach the listener once per page lifetime.
+  var urlInput = document.getElementById("mcp-url");
+  if (!urlInput || urlInput.dataset.audAutofill === "1") return;
+  urlInput.dataset.audAutofill = "1";
+  urlInput.addEventListener("blur", function () {
+    var aud = document.getElementById("mcp-oauth-audience");
+    if (aud && !aud.value.trim()) aud.value = urlInput.value.trim();
+  });
 }
 
 function showCreateMcpModal() {
@@ -3483,8 +3523,20 @@ function showCreateMcpModal() {
   document.getElementById("mcp-headers").value = "";
   document.getElementById("mcp-auto-approve").checked = false;
   document.getElementById("mcp-enabled").checked = true;
+  // Reset auth radios + OAuth subfields to the 'static' default.
+  document.getElementById("mcp-auth-static").checked = true;
+  document.getElementById("mcp-auth-none").checked = false;
+  document.getElementById("mcp-auth-oauth").checked = false;
+  document.getElementById("mcp-oauth-as-url").value = "";
+  document.getElementById("mcp-oauth-registration").value = "preregistered";
+  document.getElementById("mcp-oauth-client-id").value = "";
+  document.getElementById("mcp-oauth-client-secret").value = "";
+  document.getElementById("mcp-oauth-scopes").value = "";
+  document.getElementById("mcp-oauth-audience").value = "";
   document.getElementById("mcp-create-error").style.display = "none";
   toggleMcpTransport();
+  toggleMcpAuthFields();
+  _wireMcpAudienceAutofill();
   document.getElementById("mcp-name").focus();
   _mcpCreateTrap = _installTrap("mcp-create-overlay", "mcp-create-box");
 }
@@ -3535,7 +3587,25 @@ function showEditMcpModal(serverId) {
       document.getElementById("mcp-auto-approve").checked =
         s.auto_approve || false;
       document.getElementById("mcp-enabled").checked = s.enabled !== false;
+      var authType = s.auth_type || "static";
+      document.getElementById("mcp-auth-none").checked = authType === "none";
+      document.getElementById("mcp-auth-static").checked =
+        authType === "static";
+      document.getElementById("mcp-auth-oauth").checked =
+        authType === "oauth_user";
+      document.getElementById("mcp-oauth-as-url").value =
+        s.oauth_authorization_server_url || "";
+      document.getElementById("mcp-oauth-registration").value =
+        s.oauth_registration_mode || "preregistered";
+      document.getElementById("mcp-oauth-client-id").value =
+        s.oauth_client_id || "";
+      // Secret field always blank — write-only, never read back.
+      document.getElementById("mcp-oauth-client-secret").value = "";
+      document.getElementById("mcp-oauth-scopes").value = s.oauth_scopes || "";
+      document.getElementById("mcp-oauth-audience").value =
+        s.oauth_audience || "";
       toggleMcpTransport();
+      toggleMcpAuthFields();
     })
     .catch(function () {
       showToast("Failed to load server details");
@@ -3557,11 +3627,13 @@ function _parseMcpForm() {
     return { error: "Name must match [a-zA-Z0-9._-]+" };
   if (name.indexOf("__") >= 0) return { error: "Name must not contain '__'" };
 
+  var authType = _selectedMcpAuthType();
   var payload = {
     name: name,
     transport: transport,
     auto_approve: document.getElementById("mcp-auto-approve").checked,
     enabled: document.getElementById("mcp-enabled").checked,
+    auth_type: authType,
   };
 
   if (transport === "stdio") {
@@ -3587,19 +3659,46 @@ function _parseMcpForm() {
     payload.env = envObj;
   } else {
     payload.url = document.getElementById("mcp-url").value.trim();
-    var hdrText = document.getElementById("mcp-headers").value.trim();
-    var hdrObj = {};
-    if (hdrText) {
-      hdrText.split("\n").forEach(function (line) {
-        var colon = line.indexOf(":");
-        if (colon > 0)
-          hdrObj[line.substring(0, colon).trim()] = line
-            .substring(colon + 1)
-            .trim();
-      });
+    if (authType === "static") {
+      var hdrText = document.getElementById("mcp-headers").value.trim();
+      var hdrObj = {};
+      if (hdrText) {
+        hdrText.split("\n").forEach(function (line) {
+          var colon = line.indexOf(":");
+          if (colon > 0)
+            hdrObj[line.substring(0, colon).trim()] = line
+              .substring(colon + 1)
+              .trim();
+        });
+      }
+      payload.headers = hdrObj;
+    } else {
+      // 'none' / 'oauth_user' — clear server-side static headers state.
+      payload.headers = {};
     }
-    payload.headers = hdrObj;
   }
+
+  if (authType === "oauth_user") {
+    payload.oauth_authorization_server_url = document
+      .getElementById("mcp-oauth-as-url")
+      .value.trim();
+    payload.oauth_registration_mode = document.getElementById(
+      "mcp-oauth-registration",
+    ).value;
+    payload.oauth_client_id = document
+      .getElementById("mcp-oauth-client-id")
+      .value.trim();
+    payload.oauth_scopes = document
+      .getElementById("mcp-oauth-scopes")
+      .value.trim();
+    payload.oauth_audience = document
+      .getElementById("mcp-oauth-audience")
+      .value.trim();
+    var secret = document.getElementById("mcp-oauth-client-secret").value;
+    // Submit only when the operator typed a value; redacted in audit log.
+    if (secret) payload.oauth_client_secret = secret;
+  }
+
   return payload;
 }
 

turnstone/console/static/index.html

@@ -3677,6 +3677,114 @@
             placeholder="Authorization: Bearer ..."
           ></textarea>
         </div>
+        <fieldset
+          id="mcp-auth-section"
+          style="
+            border: 1px solid var(--border);
+            padding: 10px 12px;
+            margin-top: 12px;
+          "
+        >
+          <legend style="font-size: 12px; padding: 0 6px">
+            Multitenant Authorization
+          </legend>
+          <label
+            style="display: block; margin: 4px 0; font-weight: 400; font-size: 13px"
+          >
+            <input
+              type="radio"
+              name="mcp-auth-type"
+              id="mcp-auth-none"
+              value="none"
+              onchange="toggleMcpAuthFields()"
+              style="margin-right: 6px"
+            />No authorization
+          </label>
+          <label
+            style="display: block; margin: 4px 0; font-weight: 400; font-size: 13px"
+          >
+            <input
+              type="radio"
+              name="mcp-auth-type"
+              id="mcp-auth-static"
+              value="static"
+              onchange="toggleMcpAuthFields()"
+              checked
+              style="margin-right: 6px"
+            />Static headers (single shared identity)
+          </label>
+          <label
+            style="display: block; margin: 4px 0; font-weight: 400; font-size: 13px"
+          >
+            <input
+              type="radio"
+              name="mcp-auth-type"
+              id="mcp-auth-oauth"
+              value="oauth_user"
+              onchange="toggleMcpAuthFields()"
+              style="margin-right: 6px"
+            />Per-user OAuth 2.1 (recommended)
+          </label>
+          <div id="mcp-oauth-fields" style="display: none; margin-top: 8px">
+            <label for="mcp-oauth-as-url"
+              >Authorization Server URL
+              <span style="font-weight: 400; text-transform: none"
+                >(optional — override discovery when MCP server URL is not the
+                OAuth issuer)</span
+              ></label
+            >
+            <input
+              type="text"
+              id="mcp-oauth-as-url"
+              placeholder="https://auth.example.com"
+            />
+            <label for="mcp-oauth-registration">Client Registration</label>
+            <select id="mcp-oauth-registration">
+              <option value="preregistered">preregistered</option>
+              <option value="dcr">dcr (Dynamic Client Registration)</option>
+            </select>
+            <label for="mcp-oauth-client-id">Client ID</label>
+            <input
+              type="text"
+              id="mcp-oauth-client-id"
+              placeholder="(operator-issued client_id)"
+            />
+            <label for="mcp-oauth-client-secret"
+              >Client Secret
+              <span style="font-weight: 400; text-transform: none"
+                >(write-only, never displayed)</span
+              ></label
+            >
+            <input
+              type="password"
+              id="mcp-oauth-client-secret"
+              placeholder="***"
+              autocomplete="off"
+            />
+            <label for="mcp-oauth-scopes"
+              >Scopes
+              <span style="font-weight: 400; text-transform: none"
+                >(space-separated)</span
+              ></label
+            >
+            <input
+              type="text"
+              id="mcp-oauth-scopes"
+              placeholder="openid profile"
+            />
+            <label for="mcp-oauth-audience"
+              >Audience
+              <span style="font-weight: 400; text-transform: none"
+                >(auto-populated from URL)</span
+              ></label
+            >
+            <input
+              type="text"
+              id="mcp-oauth-audience"
+              placeholder="https://mcp.example.com"
+            />
+          </div>
+        </fieldset>
         <div style="display: flex; gap: 20px; margin-top: 14px">
           <label style="margin: 0; font-size: 12px; color: var(--fg-dim)"
             ><input

turnstone/core/storage/_postgresql.py

@@ -3705,6 +3705,14 @@ class PostgreSQLBackend:
         registry_name: str | None = None,
         registry_version: str = "",
         registry_meta: str = "{}",
+        auth_type: str = "static",
+        oauth_client_id: str | None = None,
+        oauth_client_secret_ct: bytes | None = None,
+        oauth_scopes: str | None = None,
+        oauth_audience: str | None = None,
+        oauth_registration_mode: str | None = None,
+        oauth_authorization_server_url: str | None = None,
+        oauth_as_issuer_cached: str | None = None,
     ) -> None:
         from sqlalchemy.dialects import postgresql
 
@@ -3727,6 +3735,14 @@ class PostgreSQLBackend:
                     registry_name=registry_name,
                     registry_version=registry_version,
                     registry_meta=registry_meta,
+                    auth_type=auth_type,
+                    oauth_client_id=oauth_client_id,
+                    oauth_client_secret_ct=oauth_client_secret_ct,
+                    oauth_scopes=oauth_scopes,
+                    oauth_audience=oauth_audience,
+                    oauth_registration_mode=oauth_registration_mode,
+                    oauth_authorization_server_url=oauth_authorization_server_url,
+                    oauth_as_issuer_cached=oauth_as_issuer_cached,
                     created=now,
                     updated=now,
                 )

turnstone/core/storage/_protocol.py

@@ -39,6 +39,38 @@ class OIDCPendingState(TypedDict):
     created_at: str
 
 
+class MCPUserToken(TypedDict):
+    """Row shape returned by per-(user, MCP server) OAuth token lookups.
+
+    See ``docs/design/oauth-mcp.md`` §5.1.  ``access_token_ct`` and
+    ``refresh_token_ct`` are Fernet ciphertext blobs; the storage layer
+    returns them verbatim and ``MCPTokenStore`` (Phase 3) handles
+    encrypt/decrypt.
+    """
+
+    user_id: str
+    server_name: str
+    access_token_ct: bytes
+    refresh_token_ct: bytes | None
+    expires_at: str | None
+    scopes: str | None
+    as_issuer: str
+    audience: str
+    created: str
+    last_refreshed: str | None
+
+
+class MCPOAuthPendingState(TypedDict):
+    """Row shape returned when popping a pending MCP OAuth flow state."""
+
+    state: str
+    user_id: str
+    server_name: str
+    code_verifier: str
+    return_url: str
+    created_at: str
+
+
 @runtime_checkable
 class StorageBackend(Protocol):
     """Protocol that every storage backend adapter must implement.
@@ -1585,6 +1617,14 @@ class StorageBackend(Protocol):
         registry_name: str | None = None,
         registry_version: str = "",
         registry_meta: str = "{}",
+        auth_type: str = "static",
+        oauth_client_id: str | None = None,
+        oauth_client_secret_ct: bytes | None = None,
+        oauth_scopes: str | None = None,
+        oauth_audience: str | None = None,
+        oauth_registration_mode: str | None = None,
+        oauth_authorization_server_url: str | None = None,
+        oauth_as_issuer_cached: str | None = None,
     ) -> None:
         """Create an MCP server definition. No-op if server_id already exists."""
         ...

turnstone/core/storage/_schema.py

@@ -607,6 +607,19 @@ mcp_servers = sa.Table(
     sa.Column("registry_name", sa.Text, nullable=True),
     sa.Column("registry_version", sa.Text, nullable=False, server_default=""),
     sa.Column("registry_meta", sa.Text, nullable=False, server_default="{}"),
+    # Per-(user, server) OAuth 2.1 — see docs/design/oauth-mcp.md §5.2.
+    # `auth_type` is one of: 'none', 'static', 'oauth_user'.  The other
+    # `oauth_*` columns are NULL when auth_type != 'oauth_user'.
+    # `oauth_client_secret_ct` is Fernet ciphertext; never decrypted on
+    # the read path (write-only field, masked as "***" in responses).
+    sa.Column("auth_type", sa.Text, nullable=False, server_default="static"),
+    sa.Column("oauth_client_id", sa.Text, nullable=True),
+    sa.Column("oauth_client_secret_ct", sa.LargeBinary, nullable=True),
+    sa.Column("oauth_scopes", sa.Text, nullable=True),
+    sa.Column("oauth_audience", sa.Text, nullable=True),
+    sa.Column("oauth_registration_mode", sa.Text, nullable=True),
+    sa.Column("oauth_authorization_server_url", sa.Text, nullable=True),
+    sa.Column("oauth_as_issuer_cached", sa.Text, nullable=True),
     sa.Column("created", sa.Text, nullable=False),
     sa.Column("updated", sa.Text, nullable=False),
 )
@@ -693,6 +706,41 @@ oidc_pending_states = sa.Table(
     sa.Column("created_at", sa.Text, nullable=False),
 )
 
+# ---------------------------------------------------------------------------
+# MCP per-(user, server) OAuth tokens and pending authorization-flow state.
+# See docs/design/oauth-mcp.md §5.1.  No FKs at the schema level (matches
+# `oidc_*` tables; tests avoid orphan rows via fixtures).
+# ---------------------------------------------------------------------------
+
+mcp_user_tokens = sa.Table(
+    "mcp_user_tokens",
+    metadata,
+    sa.Column("user_id", sa.Text, nullable=False),
+    sa.Column("server_name", sa.Text, nullable=False),
+    sa.Column("access_token_ct", sa.LargeBinary, nullable=False),
+    sa.Column("refresh_token_ct", sa.LargeBinary, nullable=True),
+    sa.Column("expires_at", sa.Text, nullable=True),
+    sa.Column("scopes", sa.Text, nullable=True),
+    sa.Column("as_issuer", sa.Text, nullable=False),
+    sa.Column("audience", sa.Text, nullable=False),
+    sa.Column("created", sa.Text, nullable=False),
+    sa.Column("last_refreshed", sa.Text, nullable=True),
+    sa.PrimaryKeyConstraint("user_id", "server_name"),
+)
+
+mcp_oauth_pending = sa.Table(
+    "mcp_oauth_pending",
+    metadata,
+    sa.Column("state", sa.Text, primary_key=True),
+    sa.Column("user_id", sa.Text, nullable=False),
+    sa.Column("server_name", sa.Text, nullable=False),
+    sa.Column("code_verifier", sa.Text, nullable=False),
+    sa.Column("return_url", sa.Text, nullable=False),
+    sa.Column("created_at", sa.Text, nullable=False),
+)
+
+sa.Index("idx_mcp_pending_created", mcp_oauth_pending.c.created_at)
+
 # ── TLS / ACME (lacme integration) ──────────────────────────────────────────
 
 tls_account_keys = sa.Table(

turnstone/core/storage/_sqlite.py

@@ -3856,6 +3856,14 @@ class SQLiteBackend:
         registry_name: str | None = None,
         registry_version: str = "",
         registry_meta: str = "{}",
+        auth_type: str = "static",
+        oauth_client_id: str | None = None,
+        oauth_client_secret_ct: bytes | None = None,
+        oauth_scopes: str | None = None,
+        oauth_audience: str | None = None,
+        oauth_registration_mode: str | None = None,
+        oauth_authorization_server_url: str | None = None,
+        oauth_as_issuer_cached: str | None = None,
     ) -> None:
 
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
@@ -3877,6 +3885,14 @@ class SQLiteBackend:
                     "registry_name": registry_name,
                     "registry_version": registry_version,
                     "registry_meta": registry_meta,
+                    "auth_type": auth_type,
+                    "oauth_client_id": oauth_client_id,
+                    "oauth_client_secret_ct": oauth_client_secret_ct,
+                    "oauth_scopes": oauth_scopes,
+                    "oauth_audience": oauth_audience,
+                    "oauth_registration_mode": oauth_registration_mode,
+                    "oauth_authorization_server_url": oauth_authorization_server_url,
+                    "oauth_as_issuer_cached": oauth_as_issuer_cached,
                     "created": now,
                     "updated": now,
                 },

turnstone/core/storage/_utils.py

@@ -167,6 +167,13 @@ MCP_SERVER_MUTABLE = frozenset(
         "registry_name",
         "registry_version",
         "registry_meta",
+        "auth_type",
+        "oauth_client_id",
+        "oauth_scopes",
+        "oauth_audience",
+        "oauth_registration_mode",
+        "oauth_authorization_server_url",
+        "oauth_as_issuer_cached",
     }
 )
 MODEL_DEFINITION_MUTABLE = frozenset(

turnstone/core/storage/migrations/versions/049_mcp_oauth_schema.py

@@ -0,0 +1,101 @@
+"""Add OAuth-MCP schema (Phase 2).
+
+Adds two new tables (``mcp_user_tokens``, ``mcp_oauth_pending``) and
+eight new columns on ``mcp_servers`` to support per-(user, server)
+OAuth 2.1 + PKCE authorization for MCP servers.  See
+``docs/design/oauth-mcp.md`` §5.1 / §5.2.
+
+Existing rows continue working: every new column is either nullable or
+defaults to ``'static'`` (the existing behavior).  After the new
+``auth_type`` column lands, rows with empty ``headers`` are normalized
+to ``auth_type='none'`` so the operator UI can hide the static-headers
+field for those.
+
+Revision ID: 049
+Revises: 048
+Create Date: 2026-05-04
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "049"
+down_revision = "048"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "mcp_user_tokens",
+        sa.Column("user_id", sa.Text, nullable=False),
+        sa.Column("server_name", sa.Text, nullable=False),
+        sa.Column("access_token_ct", sa.LargeBinary, nullable=False),
+        sa.Column("refresh_token_ct", sa.LargeBinary, nullable=True),
+        sa.Column("expires_at", sa.Text, nullable=True),
+        sa.Column("scopes", sa.Text, nullable=True),
+        sa.Column("as_issuer", sa.Text, nullable=False),
+        sa.Column("audience", sa.Text, nullable=False),
+        sa.Column("created", sa.Text, nullable=False),
+        sa.Column("last_refreshed", sa.Text, nullable=True),
+        sa.PrimaryKeyConstraint("user_id", "server_name"),
+    )
+
+    op.create_table(
+        "mcp_oauth_pending",
+        sa.Column("state", sa.Text, primary_key=True),
+        sa.Column("user_id", sa.Text, nullable=False),
+        sa.Column("server_name", sa.Text, nullable=False),
+        sa.Column("code_verifier", sa.Text, nullable=False),
+        sa.Column("return_url", sa.Text, nullable=False),
+        sa.Column("created_at", sa.Text, nullable=False),
+    )
+    op.create_index("idx_mcp_pending_created", "mcp_oauth_pending", ["created_at"])
+
+    op.add_column(
+        "mcp_servers",
+        sa.Column("auth_type", sa.Text, nullable=False, server_default="static"),
+    )
+    op.add_column("mcp_servers", sa.Column("oauth_client_id", sa.Text, nullable=True))
+    op.add_column(
+        "mcp_servers",
+        sa.Column("oauth_client_secret_ct", sa.LargeBinary, nullable=True),
+    )
+    op.add_column("mcp_servers", sa.Column("oauth_scopes", sa.Text, nullable=True))
+    op.add_column("mcp_servers", sa.Column("oauth_audience", sa.Text, nullable=True))
+    op.add_column("mcp_servers", sa.Column("oauth_registration_mode", sa.Text, nullable=True))
+    op.add_column(
+        "mcp_servers",
+        sa.Column("oauth_authorization_server_url", sa.Text, nullable=True),
+    )
+    op.add_column("mcp_servers", sa.Column("oauth_as_issuer_cached", sa.Text, nullable=True))
+
+    # Normalize the no-static-headers case after the column exists.
+    # Restricted to streamable-http rows — for stdio rows the column
+    # value is opaque (auth_type is meaningless when there is no HTTP
+    # transport to attach headers to), so leave them at the 'static'
+    # default.
+    # NOTE: lossy on downgrade — once a row is rewritten to 'none', the
+    # previous distinction (server_default 'static' vs explicit 'none')
+    # cannot be recovered from the DB alone.
+    op.execute(
+        "UPDATE mcp_servers SET auth_type = 'none' "
+        "WHERE transport = 'streamable-http' "
+        "AND (headers IS NULL OR headers = '' OR headers = '{}')"
+    )
+
+
+def downgrade() -> None:
+    # Reverse order of upgrade; column drops mirror add_column calls.
+    op.drop_column("mcp_servers", "oauth_as_issuer_cached")
+    op.drop_column("mcp_servers", "oauth_authorization_server_url")
+    op.drop_column("mcp_servers", "oauth_registration_mode")
+    op.drop_column("mcp_servers", "oauth_audience")
+    op.drop_column("mcp_servers", "oauth_scopes")
+    op.drop_column("mcp_servers", "oauth_client_secret_ct")
+    op.drop_column("mcp_servers", "oauth_client_id")
+    op.drop_column("mcp_servers", "auth_type")
+
+    op.drop_index("idx_mcp_pending_created", table_name="mcp_oauth_pending")
+    op.drop_table("mcp_oauth_pending")
+    op.drop_table("mcp_user_tokens")