From 9b51048d01fbb1b92e9d3a2ca5a5de718e1fa0f5 Mon Sep 17 00:00:00 2001 From: HUANG Cheng Date: Thu, 27 Aug 2026 13:22:06 +0800 Subject: [PATCH] feat(connectors): add OAuth for custom MCP servers Probe-driven OAuth discovery from MCP URL, unified oauth/start API, encrypted token storage, and Custom Connectors UI for save-then-probe and one-click authorization. Fixes #337 --- CHANGELOG.md | 4 + dashboard/src/api/modules/connectors.ts | 71 ++- dashboard/src/locales/en.json | 15 +- dashboard/src/locales/zh.json | 15 +- .../Agent/Connectors/CustomMcpServerCard.tsx | 155 ++++- .../pages/Agent/Connectors/CustomMcpTab.tsx | 595 ++++++++++++++---- .../pages/Agent/Connectors/customMcpUtils.ts | 55 ++ .../pages/Agent/Connectors/index.module.less | 83 ++- .../src/pages/Agent/Connectors/index.tsx | 2 +- src/octop/api/routers/connectors.py | 250 ++++++-- src/octop/infra/connectors/custom_mcp.py | 121 +++- src/octop/infra/connectors/oauth/__init__.py | 2 + src/octop/infra/connectors/oauth/discovery.py | 190 ++++++ src/octop/infra/connectors/oauth/registry.py | 223 ++++++- src/octop/infra/connectors/probe.py | 40 +- src/octop/infra/connectors/service.py | 106 +++- tests/integration/test_connectors_api.py | 38 +- tests/unit/connectors/test_oauth_discovery.py | 106 ++++ 18 files changed, 1815 insertions(+), 256 deletions(-) create mode 100644 src/octop/infra/connectors/oauth/discovery.py create mode 100644 tests/unit/connectors/test_oauth_discovery.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f75bddd5..cf5806b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ ## [Unreleased] +### 新增 + +- 自定义 MCP 连接器支持 OAuth：从 MCP URL 自动发现授权端点，统一 `POST /connectors/oauth/start` 启动授权；保存后可探测、一键授权，Bearer 注入 harness spec；Custom Connectors UI 支持启用与「对话默认选中」分离、保存后可选探测 + ## [0.9.28] - 2026-08-26 ### 修复 diff --git a/dashboard/src/api/modules/connectors.ts b/dashboard/src/api/modules/connectors.ts index 376271fd..07100e8c 100644 --- a/dashboard/src/api/modules/connectors.ts +++ b/dashboard/src/api/modules/connectors.ts @@ -82,11 +82,23 @@ export interface ConnectorProbeResult { tool_count?: number; tools?: { name: string; description: string }[]; error?: string; + error_type?: string; status_code?: number; + oauth?: { + available: boolean; + issuer?: string; + resource?: string; + }; } export type CustomMcpTransport = "streamable_http" | "stdio"; +export interface CustomMcpOAuthPreview { + configured?: boolean; + required?: boolean; + expires_at?: number; +} + export interface CustomMcpServerSpec { transport: CustomMcpTransport; url?: string; @@ -99,8 +111,13 @@ export interface CustomMcpServerSpec { display_name?: string; /** When true, chat composer pre-selects this MCP server. */ default_open?: boolean; + oauth?: CustomMcpOAuthPreview; } +export type OAuthStartTarget = + | { type: "catalog"; kind: string } + | { type: "custom_mcp"; server_name: string }; + export type CustomMcpServers = Record; export interface ConnectorCliInstallResult { @@ -144,7 +161,9 @@ export const connectorsApi = { listInstances: () => request("/connector-instances"), getInstance: (instanceId: string) => - request(`/connector-instances/${instanceId}`), + request( + `/connector-instances/${encodeURIComponent(instanceId)}`, + ), createInstance: (body: { kind: string; @@ -158,23 +177,41 @@ export const connectorsApi = { }), deleteInstance: (instanceId: string) => - request(`/connector-instances/${instanceId}`, { method: "DELETE" }), + request(`/connector-instances/${encodeURIComponent(instanceId)}`, { + method: "DELETE", + }), patchInstance: ( instanceId: string, body: { status?: "active" | "disabled"; default_open?: boolean }, ) => - request(`/connector-instances/${instanceId}`, { - method: "PATCH", - body: JSON.stringify(body), - }), + request( + `/connector-instances/${encodeURIComponent(instanceId)}`, + { + method: "PATCH", + body: JSON.stringify(body), + }, + ), testInstance: (instanceId: string) => - request(`/connector-instances/${instanceId}/test`, { - method: "POST", - }), + request( + `/connector-instances/${encodeURIComponent(instanceId)}/test`, + { + method: "POST", + }, + ), - oauthStart: (kind: string, redirectAfter?: string) => + oauthStart: (target: OAuthStartTarget, redirectAfter?: string) => + request<{ authorize_url: string; state_id: string }>( + "/connectors/oauth/start", + { + method: "POST", + body: JSON.stringify({ target, redirect_after: redirectAfter }), + }, + ), + + /** @deprecated Prefer oauthStart with `{ type: "catalog", kind }`. */ + oauthStartCatalog: (kind: string, redirectAfter?: string) => request<{ authorize_url: string; state_id: string }>( `/connectors/oauth/${kind}/start`, { @@ -184,9 +221,12 @@ export const connectorsApi = { ), oauthPending: (stateId: string) => - request<{ kind: string; tokens: Record }>( - `/connectors/oauth/pending/${stateId}`, - ), + request<{ + kind: string; + tokens: Record; + server_name?: string; + applied?: boolean; + }>(`/connectors/oauth/pending/${stateId}`), authorizeUrl: (kind: string) => request<{ authorize_url: string | null }>( @@ -284,7 +324,10 @@ export const connectorsApi = { body: JSON.stringify({ servers }), }), - patchCustomMcpServer: (name: string, body: { enabled: boolean }) => + patchCustomMcpServer: ( + name: string, + body: { enabled?: boolean; default_open?: boolean }, + ) => request<{ servers: CustomMcpServers }>( `/connectors/custom-mcp/servers/${encodeURIComponent(name)}`, { diff --git a/dashboard/src/locales/en.json b/dashboard/src/locales/en.json index bce08fe8..e9da19df 100644 --- a/dashboard/src/locales/en.json +++ b/dashboard/src/locales/en.json @@ -4280,9 +4280,22 @@ "duplicateName": "Server IDs must be unique", "emptyName": "Server ID is required", "probeNeedConfig": "Fill in the configuration before probing", + "probeOnSave": "Probe connection after save", + "probeOnSaveHint": "Octop will request your MCP URL to verify connectivity. If sign-in is required, we will guide you through OAuth.", + "probeNeedsOAuth": "This MCP requires OAuth before it can be used", + "probeComplete": "Connection verified", + "oauthConfigured": "OAuth authorization completed", + "oauthAuthorizeHint": "One-click OAuth saves your config first, then opens sign-in; we verify the connection again afterward.", + "oauthBeforeEnable": "Complete OAuth authorization before enabling this server", + "enable": "Enable connector", + "enableHint": "When off, agents will not load tools from this MCP.", + "defaultOpen": "Selected by default in chat", + "defaultOpenHint": "When on, your Dashboard, IM, and Cron (with no manual picks) include this MCP by default.", + "defaultOpenRequiresEnable": "Enable the connector first.", + "authorizing": "Authorizing…", "probeOk": "Probe ok — found {{count}} tools", "deleteConfirm": "Delete MCP server \"{{name}}\"?", - "deleteConfirmHint": "Changes take effect after you click Save." + "deleteConfirmHint": "Saved servers are removed immediately; unsaved drafts are dropped from this editor only." }, "configuredBadge": "Connected", "clickToConnect": "Click to connect", diff --git a/dashboard/src/locales/zh.json b/dashboard/src/locales/zh.json index d94c3228..3290c9cf 100644 --- a/dashboard/src/locales/zh.json +++ b/dashboard/src/locales/zh.json @@ -4419,9 +4419,22 @@ "duplicateName": "服务器 ID 不能重复", "emptyName": "请填写服务器 ID", "probeNeedConfig": "请先填写完整配置再探测", + "probeOnSave": "保存后自动探测连接", + "probeOnSaveHint": "将向您填写的 MCP 地址发起请求以验证可用性；若需登录，会引导您完成 OAuth。", + "probeNeedsOAuth": "此 MCP 需要 OAuth 授权才能访问", + "probeComplete": "连接正常", + "oauthConfigured": "已完成 OAuth 授权", + "oauthAuthorizeHint": "点击「一键授权」将先保存配置再打开登录页；完成后我们会自动再次验证连接。", + "oauthBeforeEnable": "请先完成 OAuth 授权后再启用", + "enable": "启用连接器", + "enableHint": "关闭后 Agent 不会加载此 MCP 的工具。", + "defaultOpen": "对话默认选中", + "defaultOpenHint": "开启后，你的 Dashboard、IM 与 Cron（未手动选连接器时）会默认带上此 MCP。", + "defaultOpenRequiresEnable": "需先启用连接器。", + "authorizing": "授权中…", "probeOk": "探测成功，发现 {{count}} 个工具", "deleteConfirm": "确定删除 MCP 服务器「{{name}}」？", - "deleteConfirmHint": "删除后需点击保存才会生效。" + "deleteConfirmHint": "已保存的服务器将立即删除；尚未保存的配置只会从当前编辑中移除。" }, "configuredBadge": "已连接", "clickToConnect": "点击连接", diff --git a/dashboard/src/pages/Agent/Connectors/CustomMcpServerCard.tsx b/dashboard/src/pages/Agent/Connectors/CustomMcpServerCard.tsx index 0be341b7..96297a08 100644 --- a/dashboard/src/pages/Agent/Connectors/CustomMcpServerCard.tsx +++ b/dashboard/src/pages/Agent/Connectors/CustomMcpServerCard.tsx @@ -21,23 +21,31 @@ import styles from "./index.module.less"; interface CustomMcpServerCardProps { card: ServerCardState; probing: boolean; + authorizing?: boolean; + oauthAvailable?: boolean; probeTools?: { name: string; description: string }[]; transportOptions: { value: string; label: string }[]; onUpdate: (key: string, patch: Partial) => void; onToggleEnabled: (enabled: boolean) => void; - onRemove: () => void; + onRemove: () => void | Promise; onProbe: () => void; + onAuthorize?: () => void; + onDefaultOpenChange?: (defaultOpen: boolean) => void; } export function CustomMcpServerCard({ card, probing, + authorizing = false, + oauthAvailable = false, probeTools, transportOptions, onUpdate, onToggleEnabled, onRemove, onProbe, + onAuthorize, + onDefaultOpenChange, }: CustomMcpServerCardProps) { const { t } = useTranslation(); const isHttp = card.transport === "streamable_http"; @@ -63,20 +71,36 @@ export function CustomMcpServerCard({ }), content: t( "connectors.customMcp.deleteConfirmHint", - "删除后需点击保存才会生效。", + "已保存的服务器将立即删除；尚未保存的配置只会从当前编辑中移除。", ), okText: t("common.delete"), okButtonProps: { danger: true }, cancelText: t("common.cancel"), - onOk: onRemove, + onOk: () => Promise.resolve(onRemove()), }); }; + const authPending = isHttp && oauthAvailable && !card.oauthConfigured; + const effectiveEnabled = !authPending && card.enabled; + const showOAuthConnectLink = + isHttp && card.collapsed && authPending && onAuthorize; + const connectLabel = t("connectors.clickToConnect", "点击连接"); + + const handleConnectClick = () => { + if (oauthAvailable && onAuthorize) { + onAuthorize(); + } + }; + return (
- +
+ + {t("connectors.customMcp.enable", "启用连接器")} + + +
+ {showOAuthConnectLink ? ( +
+ +
+ ) : null} + {!card.collapsed ? (
@@ -262,46 +307,100 @@ export function CustomMcpServerCard({ )}
- +
- onUpdate(card.key, { defaultOpen: checked }) - } + disabled={!effectiveEnabled} + onChange={(checked) => { + if (onDefaultOpenChange) { + onDefaultOpenChange(checked); + return; + } + onUpdate(card.key, { defaultOpen: checked }); + }} /> - {!card.defaultOpen ? ( + {!effectiveEnabled ? ( {t( - "connectors.defaultOpenHint", - "关闭时需在对话中手动勾选才会注入工具。", + "connectors.customMcp.defaultOpenRequiresEnable", + "需先启用连接器。", + )} + + ) : !card.defaultOpen ? ( + + {t( + "connectors.customMcp.defaultOpenHint", + "开启后，你的 Dashboard、IM 与 Cron（未手动选连接器时）会默认带上此 MCP。", )} ) : null}
- {card.defaultOpen ? ( + {effectiveEnabled && card.defaultOpen ? ( ) : null}
-
- -
+ {isHttp && card.oauthConfigured ? ( + + ) : null} + + {isHttp && oauthAvailable && !card.oauthConfigured ? ( + + {t("connectors.oneClickOAuth", "一键授权")} + + ) : undefined + } + /> + ) : null} + + {isHttp ? ( +
+ +
+ ) : null} - {probeTools !== undefined ? ( + {isHttp && probeTools !== undefined ? (
([]); const [jsonText, setJsonText] = useState("{}"); const [jsonError, setJsonError] = useState(null); + const [probeOnSave, setProbeOnSave] = useState( + () => localStorage.getItem(PROBE_ON_SAVE_KEY) === "1", + ); + const [oauthAvailable, setOauthAvailable] = useState>( + {}, + ); + const [authorizingKey, setAuthorizingKey] = useState(null); + const [persistedNames, setPersistedNames] = useState>( + () => new Set(), + ); + + useEffect(() => { + localStorage.setItem(PROBE_ON_SAVE_KEY, probeOnSave ? "1" : "0"); + }, [probeOnSave]); + + const applySavedServers = ( + servers: CustomMcpServers, + prevCards: ServerCardState[], + ) => { + const nextCards = mergeCustomMcpCards(servers, prevCards); + setCards(nextCards); + setPersistedNames(new Set(Object.keys(servers))); + setJsonText(JSON.stringify(servers, null, 2)); + setOauthAvailable(oauthHintsFromServers(servers, nextCards)); + return nextCards; + }; const load = useCallback(async () => { setLoading(true); try { const { servers } = await connectorsApi.getCustomMcp(); - const nextCards = serversToCards(servers); - setCards(nextCards); + applySavedServers(servers, []); setProbeResults({}); - setJsonText(JSON.stringify(servers, null, 2)); setJsonError(null); } catch (e) { console.error(e); @@ -64,6 +92,35 @@ export function CustomMcpTab() { void load(); }, [load]); + const persistServerPatch = async ( + card: ServerCardState, + apiPatch: { enabled?: boolean; default_open?: boolean }, + localPatch: Partial, + ) => { + const optimisticCards = cards.map((item) => + item.key === card.key ? { ...item, ...localPatch } : item, + ); + updateCard(card.key, localPatch); + const name = card.name.trim(); + if (!name || !persistedNames.has(name)) { + return; + } + try { + const { servers } = await connectorsApi.patchCustomMcpServer( + name, + apiPatch, + ); + applySavedServers(servers, optimisticCards); + notifyConnectorsChanged(); + } catch (e) { + console.error(e); + message.error( + apiErrorMessage(e, t("connectors.customMcp.saveFailed", "保存失败"), t), + ); + await load(); + } + }; + const syncJsonFromCards = useCallback((nextCards: ServerCardState[]) => { try { const servers = cardsToServers(nextCards); @@ -138,18 +195,45 @@ export function CustomMcpTab() { setMode("visual"); }; - const handleRemove = (key: string) => { - setCards((prev) => { - const next = prev.filter((card) => card.key !== key); - syncJsonFromCards(next); + const handleRemove = async (key: string) => { + const card = cards.find((item) => item.key === key); + const serverName = card?.name.trim() ?? ""; + const nextCards = cards.filter((item) => item.key !== key); + setCards(nextCards); + syncJsonFromCards(nextCards); + setProbeResults((prev) => { + if (!(key in prev)) return prev; + const next = { ...prev }; + delete next[key]; return next; }); - setProbeResults((prev) => { + setOauthAvailable((prev) => { if (!(key in prev)) return prev; const next = { ...prev }; delete next[key]; return next; }); + + if (!serverName || !persistedNames.has(serverName)) { + return; + } + + try { + await connectorsApi.deleteInstance(`custom:${serverName}`); + setPersistedNames((prev) => { + const next = new Set(prev); + next.delete(serverName); + return next; + }); + notifyConnectorsChanged(); + message.success(t("connectors.deleteSuccess", "已删除")); + } catch (e) { + console.error(e); + message.error( + apiErrorMessage(e, t("connectors.deleteFailed", "删除失败"), t), + ); + await load(); + } }; const clearProbeResult = (key: string) => { @@ -159,6 +243,96 @@ export function CustomMcpTab() { delete next[key]; return next; }); + setOauthAvailable((prev) => { + if (!(key in prev)) return prev; + const next = { ...prev }; + delete next[key]; + return next; + }); + }; + + const applyProbeResult = ( + card: ServerCardState, + result: ConnectorProbeResult, + options: { fromSave?: boolean } = {}, + ) => { + const fromSave = options.fromSave === true; + if (result.ok) { + const tools = result.tools ?? []; + setProbeResults((prev) => ({ ...prev, [card.key]: tools })); + setOauthAvailable((prev) => ({ ...prev, [card.key]: false })); + if (!fromSave) { + updateCard(card.key, { collapsed: false }); + } + if (!fromSave) { + if (tools.length === 0) { + message.success( + t("connectors.probeToolsEmpty", "连接正常，但未发现可用工具"), + ); + } else { + message.success(t("connectors.customMcp.probeComplete", "连接正常")); + } + } + return; + } + setProbeResults((prev) => { + const next = { ...prev }; + delete next[card.key]; + return next; + }); + if (result.oauth?.available) { + setOauthAvailable((prev) => ({ ...prev, [card.key]: true })); + if (fromSave) { + updateCard(card.key, { enabled: false, defaultOpen: false }); + } else { + updateCard(card.key, { + collapsed: false, + enabled: false, + defaultOpen: false, + }); + } + if (!fromSave) { + message.warning( + t( + "connectors.customMcp.probeNeedsOAuth", + "此 MCP 需要 OAuth 授权才能访问", + ), + ); + } + return; + } + setOauthAvailable((prev) => ({ ...prev, [card.key]: false })); + if (!fromSave) { + message.error(result.error ?? t("connectors.probeFailed", "探测失败")); + } + }; + + const runProbe = async ( + card: ServerCardState, + options: { byName?: boolean; fromSave?: boolean } = {}, + ) => { + const byName = options.byName === true; + const fromSave = options.fromSave === true; + setProbingKey(card.key); + clearProbeResult(card.key); + try { + let result: ConnectorProbeResult; + if (byName || card.oauthConfigured) { + result = await connectorsApi.testCustomMcp({ name: card.name.trim() }); + } else { + const map = cardsToServers([card]); + const server = map[card.name.trim()]; + result = await connectorsApi.testCustomMcp({ server }); + } + applyProbeResult(card, result, { fromSave }); + } catch (e) { + console.error(e); + message.error( + apiErrorMessage(e, t("connectors.probeFailed", "探测失败"), t), + ); + } finally { + setProbingKey(null); + } }; const resolveServersForSave = (): CustomMcpServers | null => { @@ -207,13 +381,19 @@ export function CustomMcpTab() { setSaving(true); try { const saved = await connectorsApi.putCustomMcp(servers); - const nextCards = serversToCards(saved.servers); - setCards(nextCards); - setJsonText(JSON.stringify(saved.servers, null, 2)); + const nextCards = applySavedServers(saved.servers, cards); notifyConnectorsChanged(); message.success( t("connectors.customMcp.saveSuccess", "自定义 MCP 已保存"), ); + if (probeOnSave && showProbeSection) { + for (const card of nextCards) { + if (card.transport !== "streamable_http" || !card.url.trim()) { + continue; + } + await runProbe(card, { byName: true, fromSave: true }); + } + } } catch (e) { console.error(e); message.error( @@ -225,39 +405,145 @@ export function CustomMcpTab() { }; const handleProbe = async (card: ServerCardState) => { - let server: CustomMcpServerSpec; try { - const map = cardsToServers([card]); - server = map[card.name.trim()]; + cardsToServers([card]); } catch { message.warning( t("connectors.customMcp.probeNeedConfig", "请先填写完整配置再探测"), ); return; } - setProbingKey(card.key); - clearProbeResult(card.key); + await runProbe(card); + }; + + const handleOAuth = async (card: ServerCardState) => { + const serverName = card.name.trim(); + if (!serverName) { + message.warning(t("connectors.customMcp.emptyName", "请填写服务器名称")); + return; + } + const popup = window.open("", "octop-oauth", "width=520,height=720"); + if (!popup) { + message.error( + t( + "connectors.oauthPopupBlocked", + "授权窗口被浏览器拦截，请允许本站弹出窗口后重试", + ), + ); + return; + } + + setAuthorizingKey(card.key); + let settled = false; + let pollTimer: ReturnType | undefined; + let timeoutTimer: ReturnType | undefined; + let stateId = ""; + + const cleanup = () => { + if (pollTimer !== undefined) clearInterval(pollTimer); + if (timeoutTimer !== undefined) clearTimeout(timeoutTimer); + window.removeEventListener("message", onMessage); + }; + + const finish = async () => { + if (settled) return; + settled = true; + cleanup(); + try { + popup.close(); + } catch { + // ignore + } + setAuthorizingKey(null); + try { + const { servers } = await connectorsApi.getCustomMcp(); + const nextCards = applySavedServers(servers, cards); + const refreshed = nextCards.find((c) => c.name.trim() === serverName); + if (refreshed) { + await runProbe(refreshed, { byName: true }); + } + message.success( + t("connectors.oauthConfigured", "已授权，可直接探测或保存"), + ); + } catch (e) { + console.error(e); + message.error( + apiErrorMessage( + e, + t("connectors.oauthFailed", "获取授权结果失败"), + t, + ), + ); + } + }; + + const claimPending = async () => { + if (!stateId || settled) return; + try { + const pending = await connectorsApi.oauthPending(stateId); + if (pending.applied || pending.server_name) { + await finish(); + } + } catch { + // keep polling until timeout + } + }; + + const onMessage = (ev: MessageEvent) => { + if (ev.data?.type !== "octop:connector-oauth") return; + if (ev.data.state_id !== stateId) return; + void claimPending(); + }; + try { - const result = await connectorsApi.testCustomMcp({ server }); - if (result.ok) { - const tools = result.tools ?? []; - setProbeResults((prev) => ({ ...prev, [card.key]: tools })); - updateCard(card.key, { collapsed: false }); - if (tools.length === 0) { - message.success( - t("connectors.probeToolsEmpty", "连接正常，但未发现可用工具"), - ); + const servers = resolveServersForSave(); + if (!servers) { + try { + popup.close(); + } catch { + // ignore } - } else { - message.error(result.error ?? t("connectors.probeFailed", "探测失败")); + setAuthorizingKey(null); + return; } + const saved = await connectorsApi.putCustomMcp(servers); + applySavedServers(saved.servers, cards); + notifyConnectorsChanged(); + + const { authorize_url, state_id } = await connectorsApi.oauthStart( + { type: "custom_mcp", server_name: serverName }, + window.location.pathname, + ); + stateId = state_id; + window.addEventListener("message", onMessage); + pollTimer = setInterval(() => { + void claimPending(); + }, 1200); + timeoutTimer = setTimeout(() => { + if (settled) return; + settled = true; + cleanup(); + setAuthorizingKey(null); + message.error( + t("connectors.oauthTimedOut", "授权超时，请重试一键授权"), + ); + }, 120_000); + popup.location.replace(authorize_url); } catch (e) { - console.error(e); + cleanup(); + setAuthorizingKey(null); + try { + popup.close(); + } catch { + // ignore + } message.error( - apiErrorMessage(e, t("connectors.probeFailed", "探测失败"), t), + apiErrorMessage( + e, + t("connectors.oauthStartFailed", "无法启动 OAuth"), + t, + ), ); - } finally { - setProbingKey(null); } }; @@ -269,6 +555,21 @@ export function CustomMcpTab() { [], ); + const showProbeSection = useMemo(() => { + if (mode === "visual") { + return hasHttpProbeTargets(cards); + } + try { + const parsed = JSON.parse(jsonText) as unknown; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + return false; + } + return hasHttpProbeTargets(parsed as CustomMcpServers); + } catch { + return false; + } + }, [mode, cards, jsonText]); + if (loading) { return (
@@ -277,6 +578,39 @@ export function CustomMcpTab() { ); } + const saveFooter = ( +
+ {showProbeSection ? ( +
+ setProbeOnSave(e.target.checked)} + > + {t("connectors.customMcp.probeOnSave", "保存后自动探测连接")} + +

+ {t( + "connectors.customMcp.probeOnSaveHint", + "将向您填写的 MCP 地址发起请求以验证可用性；若需登录，会引导您完成 OAuth。", + )} +

+
+ ) : null} + +
+ ); + return (
@@ -307,97 +641,128 @@ export function CustomMcpTab() {
{mode === "json" ? ( -
- { - setJsonText(e.target.value); - setJsonError(null); - }} - autoSize={{ minRows: 16, maxRows: 32 }} - className={styles.customMcpJsonArea} - spellCheck={false} - /> - {jsonError ? ( -
{jsonError}
- ) : null} -
- ) : ( <> -
- - +
+
+ { + setJsonText(e.target.value); + setJsonError(null); + }} + autoSize={{ minRows: 16, maxRows: 32 }} + className={styles.customMcpJsonArea} + spellCheck={false} + /> + {jsonError ? ( +
{jsonError}
+ ) : null} +
- -
-
- {t("connectors.customMcp.listTitle", "已添加的服务器")} - {cards.length > 0 ? ( - - {cards.length} + {saveFooter} + + ) : ( + <> +
+
+ +
- {cards.length === 0 ? ( -
- {t( - "connectors.customMcp.emptyList", - "尚未添加自定义 MCP，点击上方按钮开始配置", - )} -
- ) : ( -
- {cards.map((card) => ( - - updateCard(card.key, { enabled }) - } - onRemove={() => handleRemove(card.key)} - onProbe={() => void handleProbe(card)} - /> - ))} +
+
+ {t("connectors.customMcp.listTitle", "已添加的服务器")} + {cards.length > 0 ? ( + + {cards.length} + + ) : null}
- )} + + {cards.length === 0 ? ( +
+ {t( + "connectors.customMcp.emptyList", + "尚未添加自定义 MCP，点击上方按钮开始配置", + )} +
+ ) : ( +
+ {cards.map((card) => ( + { + if ( + oauthAvailable[card.key] && + !card.oauthConfigured && + enabled + ) { + message.warning( + t( + "connectors.customMcp.oauthBeforeEnable", + "请先完成 OAuth 授权后再启用", + ), + ); + return; + } + void persistServerPatch( + card, + { + enabled, + ...(enabled ? {} : { default_open: false }), + }, + { + enabled, + ...(enabled ? {} : { defaultOpen: false }), + }, + ); + }} + onDefaultOpenChange={(defaultOpen) => { + void persistServerPatch( + card, + { default_open: defaultOpen }, + { defaultOpen }, + ); + }} + onRemove={() => handleRemove(card.key)} + onProbe={() => void handleProbe(card)} + onAuthorize={() => void handleOAuth(card)} + /> + ))} +
+ )} +
+ {saveFooter} )} - -
- -
); } diff --git a/dashboard/src/pages/Agent/Connectors/customMcpUtils.ts b/dashboard/src/pages/Agent/Connectors/customMcpUtils.ts index a5975468..bfc3850f 100644 --- a/dashboard/src/pages/Agent/Connectors/customMcpUtils.ts +++ b/dashboard/src/pages/Agent/Connectors/customMcpUtils.ts @@ -19,8 +19,12 @@ export interface ServerCardState { enabled: boolean; defaultOpen: boolean; collapsed: boolean; + oauthConfigured: boolean; + oauthExpiresAt?: number; } +export const PROBE_ON_SAVE_KEY = "octop.customMcp.probeOnSave"; + export const EXAMPLE_JSON = `{ "deepwiki": { "display_name": "DeepWiki", @@ -122,6 +126,8 @@ export function serversToCards(servers: CustomMcpServers): ServerCardState[] { enabled: spec.enabled !== false, defaultOpen: spec.default_open === true, collapsed: true, + oauthConfigured: spec.oauth?.configured === true, + oauthExpiresAt: spec.oauth?.expires_at, })); } @@ -171,6 +177,54 @@ export function cardsToServers(cards: ServerCardState[]): CustomMcpServers { return servers; } +/** Whether any server uses HTTP transport (probe / OAuth apply to these only). */ +export function hasHttpProbeTargets( + cards: ServerCardState[] | CustomMcpServers, +): boolean { + if (Array.isArray(cards)) { + return cards.some((card) => card.transport === "streamable_http"); + } + return Object.values(cards).some( + (spec) => + spec && + typeof spec === "object" && + (spec as CustomMcpServerSpec).transport !== "stdio", + ); +} + +export function mergeCustomMcpCards( + servers: CustomMcpServers, + prevCards: ServerCardState[], +): ServerCardState[] { + return serversToCards(servers).map((card) => { + const prev = prevCards.find((c) => c.name.trim() === card.name.trim()); + if (!prev) return card; + return { + ...card, + key: prev.key, + collapsed: prev.collapsed, + }; + }); +} + +export function oauthHintsFromServers( + servers: CustomMcpServers, + cards: ServerCardState[], +): Record { + const hints: Record = {}; + for (const card of cards) { + const spec = servers[card.name.trim()]; + if ( + card.transport === "streamable_http" && + spec?.oauth?.required && + spec.oauth?.configured !== true + ) { + hints[card.key] = true; + } + } + return hints; +} + export function newCard( transport: CustomMcpTransport, index: number, @@ -190,6 +244,7 @@ export function newCard( enabled: true, defaultOpen: false, collapsed: false, + oauthConfigured: false, }; } diff --git a/dashboard/src/pages/Agent/Connectors/index.module.less b/dashboard/src/pages/Agent/Connectors/index.module.less index 1487c918..c3d0a19a 100644 --- a/dashboard/src/pages/Agent/Connectors/index.module.less +++ b/dashboard/src/pages/Agent/Connectors/index.module.less @@ -697,6 +697,16 @@ justify-content: flex-start; } +.customMcpPanel { + display: flex; + flex-direction: column; + gap: 16px; + padding: 16px; + border: 1px solid var(--fn-border-secondary); + border-radius: var(--fn-radius-lg); + background: var(--fn-bg-primary); +} + .customMcpJsonEditor { display: flex; flex-direction: column; @@ -926,6 +936,47 @@ flex-shrink: 0; } +.customMcpEnableControl { + display: flex; + align-items: center; + gap: 6px; + margin-right: 4px; +} + +.customMcpEnableLabel { + font-size: 12px; + font-weight: 500; + color: var(--fn-text-secondary); + white-space: nowrap; +} + +.customMcpCardFooter { + display: flex; + align-items: center; + padding: 0 16px 14px; + margin-top: -4px; +} + +.customMcpConnectLink { + font-size: 12px; + color: var(--mcp-accent); + font-weight: 500; + background: none; + border: none; + padding: 0; + cursor: pointer; + text-align: left; + + &:hover:not(:disabled) { + text-decoration: underline; + } + + &:disabled { + opacity: 0.65; + cursor: wait; + } +} + .customMcpCardBody { display: flex; flex-direction: column; @@ -979,6 +1030,36 @@ .customMcpFooter { display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 16px; + padding: 16px; + border: 1px solid var(--fn-border-secondary); + border-radius: var(--fn-radius-lg); + background: var(--fn-bg-primary); +} + +.customMcpFooterMain { + display: flex; + flex-direction: column; + gap: 4px; + flex: 1; + min-width: 0; +} + +.customMcpFooterHint { + margin: 0; + padding-left: 24px; + font-size: 12px; + line-height: 1.45; + color: var(--fn-text-tertiary); +} + +.customMcpFooterSave { + flex-shrink: 0; + align-self: center; +} + +.customMcpFooterSaveOnly { justify-content: flex-end; - padding-top: 4px; } diff --git a/dashboard/src/pages/Agent/Connectors/index.tsx b/dashboard/src/pages/Agent/Connectors/index.tsx index abccf84e..a3b7e542 100644 --- a/dashboard/src/pages/Agent/Connectors/index.tsx +++ b/dashboard/src/pages/Agent/Connectors/index.tsx @@ -785,7 +785,7 @@ function ConnectorConfigDrawer({ try { const { authorize_url, state_id } = await connectorsApi.oauthStart( - entry.kind, + { type: "catalog", kind: entry.kind }, "/connectors", ); stateId = state_id; diff --git a/src/octop/api/routers/connectors.py b/src/octop/api/routers/connectors.py index 0624d749..8e43901a 100644 --- a/src/octop/api/routers/connectors.py +++ b/src/octop/api/routers/connectors.py @@ -11,7 +11,7 @@ from fastapi import APIRouter, Depends, Query, Request from fastapi.responses import HTMLResponse -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator from octop.api.common.public_base import resolve_public_base from octop.api.deps import current_user, get_server, require_permission @@ -23,13 +23,14 @@ from octop.infra.connectors.catalog import ( catalog_entry_to_dict, get_catalog_entry, - get_mcp_oauth_remote, list_catalog, ) from octop.infra.connectors.custom_mcp import ( CUSTOM_MCP_KIND, is_custom_mcp_kind, + oauth_configured, parse_synthetic_instance_id, + redact_servers_for_api, ) from octop.infra.connectors.default_open import ( build_instance_config_json, @@ -50,7 +51,11 @@ load_oauth_ctx, oauth_ready_for_kind, save_oauth_ctx, - start_oauth, + start_oauth_for_target, +) +from octop.infra.connectors.oauth.registry import ( + oauth_state_kind_for_target, + oauth_target_requires_https, ) from octop.infra.connectors.probe import ( prepare_probe_credentials, @@ -95,6 +100,15 @@ class PatchInstanceBody(BaseModel): class OAuthStartBody(BaseModel): redirect_after: str | None = None + target: dict[str, Any] | None = Field( + default=None, + description='OAuth target, e.g. {"type":"catalog","kind":"notion"} or ' + '{"type":"custom_mcp","server_name":"my-server"}', + ) + + +class OAuthStartLegacyBody(BaseModel): + redirect_after: str | None = None class ExchangeAuthCodeBody(BaseModel): @@ -127,7 +141,14 @@ class CustomMcpPutBody(BaseModel): class CustomMcpServerPatchBody(BaseModel): - enabled: bool + enabled: bool | None = None + default_open: bool | None = None + + @model_validator(mode="after") + def _require_one_field(self) -> CustomMcpServerPatchBody: + if self.enabled is None and self.default_open is None: + raise ValueError("provide enabled and/or default_open") + return self class CustomMcpTestBody(BaseModel): @@ -137,6 +158,94 @@ class CustomMcpTestBody(BaseModel): server: dict[str, Any] | None = None +def _resolve_custom_mcp_url(svc: ConnectorService, user_id: int, server_name: str) -> str: + saved = svc.get_custom_servers(user_id) + raw = saved.get(server_name) + if not isinstance(raw, dict): + raise OctopError( + ErrorCode.CONNECTOR_NOT_FOUND, + f"custom MCP server {server_name!r} not found; save configuration before OAuth", + ) + if str(raw.get("transport") or "") not in ("streamable_http", "http"): + raise OctopError( + ErrorCode.CONNECTOR_KIND_UNSUPPORTED, + "OAuth is only supported for streamable HTTP custom MCP servers", + ) + mcp_url = str(raw.get("url") or "").strip() + if not mcp_url: + raise OctopError(ErrorCode.CONNECTOR_INVALID_CREDENTIALS, "server url is required") + return mcp_url + + +async def _begin_oauth_flow( + *, + request: Request, + user: Any, + server: Any, + target: dict[str, Any], + redirect_after: str | None, +) -> dict[str, Any]: + target_type = str(target.get("type") or "").strip() + if target_type == "catalog": + kind = str(target.get("kind") or "").strip() + if not oauth_ready_for_kind(kind, server.services.settings_repo): + raise OctopError( + ErrorCode.CONNECTOR_INVALID_CREDENTIALS, + f"OAuth for {kind} is not available", + ) + elif target_type != "custom_mcp": + raise OctopError(ErrorCode.CONNECTOR_INVALID_CREDENTIALS, "unsupported oauth target") + + state = secrets.token_urlsafe(24) + state_id = new_ulid() + base = resolve_public_base(request) + redirect_uri = f"{base}/api/connectors/oauth/callback" + if oauth_target_requires_https(target) and _is_public_http_uri(redirect_uri): + label = str(target.get("kind") or target.get("server_name") or "MCP") + raise OctopError( + ErrorCode.CONNECTOR_OAUTH_HTTPS_REQUIRED, + f"{label} OAuth callbacks require HTTPS for non-loopback addresses", + ) + + mcp_url: str | None = None + if target_type == "custom_mcp": + server_name = str(target.get("server_name") or "").strip() + if not server_name: + raise OctopError( + ErrorCode.CONNECTOR_INVALID_CREDENTIALS, + "custom_mcp target requires server_name", + ) + mcp_url = _resolve_custom_mcp_url(_connector_service(server), user.id, server_name) + + try: + authorize_url, verifier, ctx = await start_oauth_for_target( + target=target, + redirect_uri=redirect_uri, + state=state, + settings_repo=server.services.settings_repo, + mcp_url=mcp_url, + ) + except ValueError as exc: + raise OctopError(ErrorCode.CONNECTOR_INVALID_CREDENTIALS, str(exc)) from exc + except Exception as exc: + logger.exception("oauth start failed for target %s", target) + raise OctopError( + ErrorCode.CONNECTOR_INVALID_CREDENTIALS, + f"无法启动 OAuth: {exc}", + ) from exc + + server.services.repos.connector_repo.create_oauth_state( + state_id=state_id, + state=state, + user_id=user.id, + kind=oauth_state_kind_for_target(target), + code_verifier=verifier, + redirect_after=redirect_after, + ) + save_oauth_ctx(server.services.settings_repo, state_id, ctx) + return {"authorize_url": authorize_url, "state_id": state_id} + + def _connector_service(server: Any) -> ConnectorService: return ConnectorService( repo=server.services.repos.connector_repo, @@ -322,7 +431,7 @@ async def get_custom_mcp( server: Any = Depends(get_server), ) -> dict[str, Any]: """Return the user's custom MCP server map (langchain-mcp-adapters shape).""" - servers = _connector_service(server).get_custom_servers(user.id) + servers = _connector_service(server).get_custom_servers_for_api(user.id) return {"servers": servers} @@ -345,12 +454,12 @@ async def put_custom_mcp( payload=str(len(servers)), ) _schedule_connector_reload(server, user.id) - return {"servers": servers} + return {"servers": redact_servers_for_api(servers)} @router.patch( "/connectors/custom-mcp/servers/{server_name}", - summary="Enable or disable one custom MCP server", + summary="Patch one custom MCP server", ) async def patch_custom_mcp_server( server_name: str, @@ -358,10 +467,15 @@ async def patch_custom_mcp_server( user: Any = Depends(current_user), server: Any = Depends(get_server), ) -> dict[str, Any]: - """Toggle ``enabled`` for a single custom MCP server without rewriting others.""" + """Update ``enabled`` and/or ``default_open`` for one custom MCP server.""" svc = _connector_service(server) try: - servers = svc.patch_custom_server_enabled(user.id, server_name, enabled=body.enabled) + servers = svc.patch_custom_server( + user.id, + server_name, + enabled=body.enabled, + default_open=body.default_open, + ) except KeyError as exc: raise OctopError( ErrorCode.CONNECTOR_NOT_FOUND, f"custom MCP server {server_name!r} not found" @@ -369,7 +483,7 @@ async def patch_custom_mcp_server( except ValueError as exc: raise OctopError(ErrorCode.CONNECTOR_INVALID_CREDENTIALS, str(exc)) from exc _schedule_connector_reload(server, user.id) - return {"servers": servers} + return {"servers": redact_servers_for_api(servers)} @router.post("/connectors/custom-mcp/test", summary="Probe a custom MCP server") @@ -396,7 +510,16 @@ async def test_custom_mcp( ErrorCode.CONNECTOR_INVALID_CREDENTIALS, "provide name or server spec to probe", ) - return await probe_custom_mcp_server(spec) + result = await probe_custom_mcp_server(spec) + if body.name: + try: + if result.get("oauth", {}).get("available") and not oauth_configured(spec): + svc.note_custom_server_oauth_required(user.id, body.name, required=True) + elif result.get("ok") or not result.get("oauth", {}).get("available"): + svc.note_custom_server_oauth_required(user.id, body.name, required=False) + except KeyError: + pass + return result @router.get("/connector-instances/{instance_id}", summary="Get connector instance") @@ -946,57 +1069,41 @@ async def auth_exchange_code( return {"credentials": tokens} -@router.post("/connectors/oauth/{kind}/start", summary="Start OAuth flow") -async def oauth_start( - kind: str, +@router.post("/connectors/oauth/start", summary="Start OAuth flow (unified)") +async def oauth_start_unified( body: OAuthStartBody, request: Request, user: Any = Depends(current_user), server: Any = Depends(get_server), ) -> dict[str, Any]: - """Begin browser OAuth: returns `authorize_url` and `state_id` to poll after redirect.""" - if not oauth_ready_for_kind(kind, server.services.settings_repo): - raise OctopError( - ErrorCode.CONNECTOR_INVALID_CREDENTIALS, - f"OAuth for {kind} is not available", - ) - - state = secrets.token_urlsafe(24) - state_id = new_ulid() - base = resolve_public_base(request) - redirect_uri = f"{base}/api/connectors/oauth/callback" - if get_mcp_oauth_remote(kind) is not None and _is_public_http_uri(redirect_uri): - raise OctopError( - ErrorCode.CONNECTOR_OAUTH_HTTPS_REQUIRED, - f"{kind} OAuth callbacks require HTTPS for non-loopback addresses", - ) + """Begin browser OAuth for a catalog connector or custom MCP server.""" + if not body.target or not isinstance(body.target, dict): + raise OctopError(ErrorCode.CONNECTOR_INVALID_CREDENTIALS, "oauth target is required") + return await _begin_oauth_flow( + request=request, + user=user, + server=server, + target=body.target, + redirect_after=body.redirect_after, + ) - try: - authorize_url, verifier, ctx = await start_oauth( - kind=kind, - redirect_uri=redirect_uri, - state=state, - settings_repo=server.services.settings_repo, - ) - except ValueError as exc: - raise OctopError(ErrorCode.CONNECTOR_INVALID_CREDENTIALS, str(exc)) from exc - except Exception as exc: - logger.exception("oauth start failed for %s", kind) - raise OctopError( - ErrorCode.CONNECTOR_INVALID_CREDENTIALS, - f"无法启动 OAuth: {exc}", - ) from exc - server.services.repos.connector_repo.create_oauth_state( - state_id=state_id, - state=state, - user_id=user.id, - kind=kind, - code_verifier=verifier, +@router.post("/connectors/oauth/{kind}/start", summary="Start OAuth flow (legacy)") +async def oauth_start_legacy( + kind: str, + body: OAuthStartLegacyBody, + request: Request, + user: Any = Depends(current_user), + server: Any = Depends(get_server), +) -> dict[str, Any]: + """Legacy catalog-only alias for :func:`oauth_start_unified`.""" + return await _begin_oauth_flow( + request=request, + user=user, + server=server, + target={"type": "catalog", "kind": kind}, redirect_after=body.redirect_after, ) - save_oauth_ctx(server.services.settings_repo, state_id, ctx) - return {"authorize_url": authorize_url, "state_id": state_id} @router.get("/connectors/oauth/callback", summary="OAuth callback") @@ -1040,11 +1147,39 @@ async def oauth_callback( logger.exception("oauth callback failed for %s", row.kind) return HTMLResponse(f"Token 交换失败: {exc}", status_code=400) + pending_payload: dict[str, Any] = { + "user_id": row.user_id, + "kind": row.kind, + "tokens": tokens, + } + if row.kind == CUSTOM_MCP_KIND: + server_name = str(ctx.get("server_name") or "") + issuer = str(ctx.get("issuer") or "") + resource = str(ctx.get("resource") or "").strip() or None + try: + svc = _connector_service(server) + svc.apply_custom_server_oauth( + row.user_id, + server_name, + tokens, + issuer=issuer, + resource=resource, + ) + _schedule_connector_reload(server, row.user_id) + pending_payload["server_name"] = server_name + pending_payload["applied"] = True + except Exception as exc: + logger.exception("custom MCP oauth apply failed for %s", server_name) + return HTMLResponse( + f"保存授权失败: {exc}", + status_code=400, + ) + # Store tokens in a short-lived settings key for frontend pickup, or auto-create instance. pending_key = f"connector.oauth.pending.{row.state_id}" server.services.settings_repo.set( pending_key, - json.dumps({"user_id": row.user_id, "kind": row.kind, "tokens": tokens}), + json.dumps(pending_payload), ) redirect = row.redirect_after or "/connectors" html = f""" @@ -1079,7 +1214,12 @@ async def oauth_pending( if int(data.get("user_id") or 0) != user.id: raise OctopError(ErrorCode.FORBIDDEN, "not your oauth session") server.services.settings_repo.delete(key) - return {"kind": data.get("kind"), "tokens": data.get("tokens") or {}} + return { + "kind": data.get("kind"), + "tokens": data.get("tokens") or {}, + "server_name": data.get("server_name"), + "applied": data.get("applied"), + } async def validate_chat_mcp_servers( diff --git a/src/octop/infra/connectors/custom_mcp.py b/src/octop/infra/connectors/custom_mcp.py index 95ea281d..90320e3c 100644 --- a/src/octop/infra/connectors/custom_mcp.py +++ b/src/octop/infra/connectors/custom_mcp.py @@ -13,6 +13,9 @@ _SERVER_NAME_RE = re.compile(r"^[A-Za-z0-9_-]+$") _META_KEYS = frozenset({"enabled", "display_name", "default_open"}) +_OAUTH_KEY = "oauth" +_SECRET_KEYS = frozenset({_OAUTH_KEY}) +_HARNESS_STRIP_KEYS = _META_KEYS | _SECRET_KEYS _DISPLAY_NAME_MAX = 64 _MCP_STREAMABLE_HTTP_ACCEPT = "application/json, text/event-stream" @@ -164,6 +167,10 @@ def normalize_server_spec(name: str, raw: Any) -> dict[str, Any]: if env: spec["env"] = env + oauth = raw.get(_OAUTH_KEY) + if isinstance(oauth, dict) and str(oauth.get("access_token") or "").strip(): + spec[_OAUTH_KEY] = dict(oauth) + return spec @@ -199,15 +206,127 @@ def validate_servers_map( return out +def oauth_tokens_from_spec(spec: dict[str, Any]) -> dict[str, Any]: + raw = spec.get(_OAUTH_KEY) + return dict(raw) if isinstance(raw, dict) else {} + + +def oauth_configured(spec: dict[str, Any]) -> bool: + oauth = oauth_tokens_from_spec(spec) + return bool(str(oauth.get("access_token") or "").strip()) + + +def oauth_required(spec: dict[str, Any]) -> bool: + if oauth_configured(spec): + return False + oauth = oauth_tokens_from_spec(spec) + return oauth.get("required") is True + + +def set_oauth_required_in_spec(spec: dict[str, Any], *, required: bool) -> dict[str, Any]: + """Persist or clear the dashboard hint that OAuth is needed (no tokens).""" + out = dict(spec) + if oauth_configured(out): + return out + oauth = dict(oauth_tokens_from_spec(out)) + if required: + oauth["required"] = True + oauth.pop("access_token", None) + oauth.pop("refresh_token", None) + out[_OAUTH_KEY] = oauth + return out + oauth.pop("required", None) + if str(oauth.get("access_token") or "").strip(): + out[_OAUTH_KEY] = oauth + else: + out.pop(_OAUTH_KEY, None) + return out + + +def build_oauth_storage( + tokens: dict[str, Any], + *, + issuer: str, + resource: str | None, +) -> dict[str, Any]: + out: dict[str, Any] = { + "access_token": str(tokens["access_token"]), + "oauth_issuer": issuer.rstrip("/"), + } + refresh = str(tokens.get("refresh_token") or "").strip() + if refresh: + out["refresh_token"] = refresh + if tokens.get("expires_at") is not None: + out["expires_at"] = int(tokens["expires_at"]) + if resource: + out["oauth_resource"] = resource + client_id = str(tokens.get("oauth_client_id") or "").strip() + if client_id: + out["oauth_client_id"] = client_id + client_secret = tokens.get("oauth_client_secret") + if client_secret: + out["oauth_client_secret"] = str(client_secret) + return out + + +def redact_server_for_api(spec: dict[str, Any]) -> dict[str, Any]: + """Remove secrets; expose oauth preview for the dashboard.""" + out = {k: v for k, v in spec.items() if k != _OAUTH_KEY} + if oauth_configured(spec): + oauth = oauth_tokens_from_spec(spec) + preview: dict[str, Any] = {"configured": True} + if oauth.get("expires_at") is not None: + preview["expires_at"] = oauth["expires_at"] + out[_OAUTH_KEY] = preview + elif oauth_required(spec): + out[_OAUTH_KEY] = {"configured": False, "required": True} + return out + + +def redact_servers_for_api(servers: dict[str, Any]) -> dict[str, Any]: + return { + name: redact_server_for_api(spec) + for name, spec in servers.items() + if isinstance(spec, dict) + } + + +def merge_preserved_oauth( + new_servers: dict[str, Any], + existing: dict[str, Any], +) -> dict[str, Any]: + """Drop client-supplied oauth blobs; keep stored tokens per server name.""" + out: dict[str, Any] = {} + for name, raw in new_servers.items(): + if not isinstance(raw, dict): + out[name] = raw + continue + spec = dict(raw) + spec.pop(_OAUTH_KEY, None) + old = existing.get(name) + if isinstance(old, dict): + old_oauth = old.get(_OAUTH_KEY) + if isinstance(old_oauth, dict) and str(old_oauth.get("access_token") or "").strip(): + spec[_OAUTH_KEY] = dict(old_oauth) + elif isinstance(old_oauth, dict) and old_oauth.get("required") is True: + spec[_OAUTH_KEY] = {"required": True} + out[name] = spec + return out + + def harness_spec_for_server(spec: dict[str, Any]) -> dict[str, Any]: """Strip Octop meta keys; keep langchain-mcp-adapters connection fields.""" - out = {k: v for k, v in spec.items() if k not in _META_KEYS} + out = {k: v for k, v in spec.items() if k not in _HARNESS_STRIP_KEYS} # Ensure stdio always has args list for adapters. if out.get("transport") == "stdio" and "args" not in out: out["args"] = [] # Streamable HTTP MCP requires both content types (same as built-in remote). if out.get("transport") == "streamable_http": headers = {str(k): str(v) for k, v in dict(out.get("headers") or {}).items()} + oauth = oauth_tokens_from_spec(spec) + token = str(oauth.get("access_token") or "").strip() + if token: + headers["Authorization"] = f"Bearer {token}" headers.setdefault("Accept", _MCP_STREAMABLE_HTTP_ACCEPT) out["headers"] = headers return out diff --git a/src/octop/infra/connectors/oauth/__init__.py b/src/octop/infra/connectors/oauth/__init__.py index 255c8311..c1937f6b 100644 --- a/src/octop/infra/connectors/oauth/__init__.py +++ b/src/octop/infra/connectors/oauth/__init__.py @@ -13,6 +13,7 @@ refresh_oauth_credentials, save_oauth_ctx, start_oauth, + start_oauth_for_target, ) __all__ = [ @@ -28,4 +29,5 @@ "refresh_oauth_credentials", "save_oauth_ctx", "start_oauth", + "start_oauth_for_target", ] diff --git a/src/octop/infra/connectors/oauth/discovery.py b/src/octop/infra/connectors/oauth/discovery.py new file mode 100644 index 00000000..b6ac0ff5 --- /dev/null +++ b/src/octop/infra/connectors/oauth/discovery.py @@ -0,0 +1,190 @@ +"""Discover MCP OAuth issuers from a remote MCP URL (RFC 9728 + MCP authorization).""" + +from __future__ import annotations + +import re +from typing import Any +from urllib.parse import urljoin, urlparse + +from octop.infra.connectors.oauth.mcp import fetch_authorization_metadata +from octop.infra.utils.ssrf_guard import UnsafeOutboundUrl, safe_request, validate_https_url + +_RESOURCE_METADATA_PARAM = re.compile( + r'resource_metadata\s*=\s*"([^"]+)"', + re.IGNORECASE, +) + + +def _mcp_url_host(url: str) -> str: + return (urlparse(url).hostname or "").lower().rstrip(".") + + +def _is_loopback_mcp_url(url: str) -> bool: + host = _mcp_url_host(url) + return host in {"localhost", "127.0.0.1", "::1"} + + +def build_protected_resource_metadata_urls( + mcp_url: str, + *, + www_auth_resource_metadata: str | None = None, +) -> list[str]: + """Ordered PRM discovery URLs per MCP authorization / RFC 9728.""" + urls: list[str] = [] + if www_auth_resource_metadata: + urls.append(www_auth_resource_metadata.strip()) + parsed = urlparse(mcp_url) + base = f"{parsed.scheme}://{parsed.netloc}" + path = parsed.path or "" + if path and path != "/": + urls.append(urljoin(base, f"/.well-known/oauth-protected-resource{path}")) + urls.append(urljoin(base, "/.well-known/oauth-protected-resource")) + # De-dupe while preserving order. + seen: set[str] = set() + out: list[str] = [] + for item in urls: + if item and item not in seen: + seen.add(item) + out.append(item) + return out + + +def parse_www_authenticate_resource_metadata(header_value: str | None) -> str | None: + if not header_value: + return None + match = _RESOURCE_METADATA_PARAM.search(header_value) + if not match: + return None + return match.group(1).strip() or None + + +async def _fetch_prm_document(url: str) -> dict[str, Any] | None: + try: + validate_https_url(url, field="resource_metadata") + except UnsafeOutboundUrl: + return None + try: + resp = await safe_request("GET", url, timeout=15.0) + except Exception: + return None + if resp.status_code >= 400: + return None + try: + data = resp.json() + except Exception: + return None + if not isinstance(data, dict): + return None + servers = data.get("authorization_servers") + if not isinstance(servers, list) or not servers: + return None + return data + + +async def _probe_401_resource_metadata(mcp_url: str) -> str | None: + """Unauthenticated MCP initialize may return WWW-Authenticate with PRM URL.""" + parsed = urlparse(mcp_url) + if parsed.scheme not in ("http", "https"): + return None + if _is_loopback_mcp_url(mcp_url): + return None + if parsed.scheme != "https": + return None + try: + validate_https_url(mcp_url, field="mcp_url") + except UnsafeOutboundUrl: + return None + try: + resp = await safe_request( + "POST", + mcp_url, + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "octop", "version": "0.1"}, + }, + }, + headers={"Accept": "application/json, text/event-stream"}, + timeout=15.0, + ) + except Exception: + return None + if resp.status_code not in (401, 403): + return None + return parse_www_authenticate_resource_metadata(resp.headers.get("WWW-Authenticate")) + + +def _normalize_issuer(raw: str, *, mcp_url: str) -> str: + text = raw.strip().rstrip("/") + if text.startswith("http://") or text.startswith("https://"): + return text + base = f"{urlparse(mcp_url).scheme}://{urlparse(mcp_url).netloc}" + return urljoin(base, text).rstrip("/") + + +async def discover_oauth_from_mcp_url(mcp_url: str) -> dict[str, Any]: + """Return OAuth discovery details for a remote MCP HTTP endpoint. + + Result shape:: + + { + "available": bool, + "issuer": str | None, + "resource": str | None, + "metadata": dict | None, # AS metadata when available + "scopes_supported": list[str] | None, + "error": str | None, + } + """ + url = mcp_url.strip() + if not url: + return {"available": False, "error": "empty mcp url"} + if _is_loopback_mcp_url(url): + return {"available": False, "error": "loopback MCP does not use remote OAuth"} + + www_auth_prm = await _probe_401_resource_metadata(url) + prm_doc: dict[str, Any] | None = None + for candidate in build_protected_resource_metadata_urls( + url, + www_auth_resource_metadata=www_auth_prm, + ): + prm_doc = await _fetch_prm_document(candidate) + if prm_doc is not None: + break + + if prm_doc is None: + return {"available": False, "error": "protected resource metadata not found"} + + resource = str(prm_doc.get("resource") or "").strip() or url + servers = prm_doc.get("authorization_servers") + if not isinstance(servers, list) or not servers: + return {"available": False, "error": "authorization_servers missing"} + + issuer = _normalize_issuer(str(servers[0]), mcp_url=url) + try: + metadata = await fetch_authorization_metadata(issuer) + except Exception as exc: + return {"available": False, "error": str(exc), "issuer": issuer, "resource": resource} + + if not metadata.get("registration_endpoint"): + return { + "available": False, + "error": "authorization server does not support dynamic client registration", + "issuer": issuer, + "resource": resource, + } + + scopes_raw = metadata.get("scopes_supported") + scopes = [str(s) for s in scopes_raw if s] if isinstance(scopes_raw, list) else None + return { + "available": True, + "issuer": issuer, + "resource": resource, + "metadata": metadata, + "scopes_supported": scopes, + "error": None, + } diff --git a/src/octop/infra/connectors/oauth/registry.py b/src/octop/infra/connectors/oauth/registry.py index 9a9f9fa3..969807d7 100644 --- a/src/octop/infra/connectors/oauth/registry.py +++ b/src/octop/infra/connectors/oauth/registry.py @@ -6,6 +6,8 @@ from typing import Any from octop.infra.connectors.catalog import get_catalog_entry, get_mcp_oauth_remote +from octop.infra.connectors.custom_mcp import CUSTOM_MCP_KIND +from octop.infra.connectors.oauth.discovery import discover_oauth_from_mcp_url from octop.infra.connectors.oauth.mcp import ( build_authorize_url, exchange_authorization_code, @@ -119,6 +121,49 @@ async def exchange_pasted_auth_code( raise ValueError(f"auth code exchange not supported for {kind}") +async def _start_mcp_oauth_from_discovery( + *, + flow: str, + kind: str, + redirect_uri: str, + state: str, + issuer: str, + resource: str | None, + metadata: dict[str, Any], + server_name: str | None = None, + mcp_url: str | None = None, + scope: str | None = None, +) -> tuple[str, str, dict[str, Any]]: + verifier, challenge = new_pkce_pair() + reg = await register_dynamic_client(metadata, issuer=issuer, redirect_uri=redirect_uri) + client_id = str(reg["client_id"]) + client_secret = str(reg.get("client_secret") or "") or None + url = build_authorize_url( + metadata, + client_id=client_id, + redirect_uri=redirect_uri, + state=state, + code_challenge=challenge, + scope=scope, + resource=resource, + ) + ctx: dict[str, Any] = { + "flow": flow, + "kind": kind, + "metadata": metadata, + "client_id": client_id, + "client_secret": client_secret, + "resource": resource, + "redirect_uri": redirect_uri, + "issuer": issuer, + } + if server_name: + ctx["server_name"] = server_name + if mcp_url: + ctx["mcp_url"] = mcp_url + return url, verifier, ctx + + async def start_oauth( *, kind: str, @@ -126,38 +171,136 @@ async def start_oauth( state: str, settings_repo: Any, ) -> tuple[str, str, dict[str, Any]]: - """Return authorize_url, code_verifier, ctx to persist for callback.""" + """Return authorize_url, code_verifier, ctx for a catalog connector kind.""" del settings_repo - verifier, challenge = new_pkce_pair() - - if kind in mcp_oauth_kinds(): - issuer = issuer_for_kind(kind) - metadata = await fetch_authorization_metadata(issuer) - reg = await register_dynamic_client(metadata, issuer=issuer, redirect_uri=redirect_uri) - client_id = str(reg["client_id"]) - client_secret = str(reg.get("client_secret") or "") or None - resource = resource_for_kind(kind) - url = build_authorize_url( - metadata, - client_id=client_id, + if kind not in mcp_oauth_kinds(): + raise ValueError(f"oauth not supported for {kind}") + issuer = issuer_for_kind(kind) + metadata = await fetch_authorization_metadata(issuer) + resource = resource_for_kind(kind) + return await _start_mcp_oauth_from_discovery( + flow="mcp", + kind=kind, + redirect_uri=redirect_uri, + state=state, + issuer=issuer, + resource=resource, + metadata=metadata, + scope=_scopes_for_kind(kind, metadata), + ) + + +async def start_oauth_for_target( + *, + target: dict[str, Any], + redirect_uri: str, + state: str, + settings_repo: Any, + mcp_url: str | None = None, +) -> tuple[str, str, dict[str, Any]]: + """Unified OAuth start for catalog connectors and custom MCP servers.""" + target_type = str(target.get("type") or "").strip() + if target_type == "catalog": + kind = str(target.get("kind") or "").strip() + if not kind: + raise ValueError("catalog target requires kind") + return await start_oauth( + kind=kind, redirect_uri=redirect_uri, state=state, - code_challenge=challenge, - scope=_scopes_for_kind(kind, metadata), - resource=resource, + settings_repo=settings_repo, ) - ctx = { - "flow": "mcp", - "kind": kind, - "metadata": metadata, - "client_id": client_id, - "client_secret": client_secret, - "resource": resource, - "redirect_uri": redirect_uri, - } - return url, verifier, ctx + if target_type == "custom_mcp": + server_name = str(target.get("server_name") or "").strip() + url = str(mcp_url or target.get("mcp_url") or "").strip() + if not server_name: + raise ValueError("custom_mcp target requires server_name") + if not url: + raise ValueError("custom MCP server url is required; save the server first") + return await start_custom_mcp_oauth( + server_name=server_name, + mcp_url=url, + redirect_uri=redirect_uri, + state=state, + ) + raise ValueError(f"unsupported oauth target type: {target_type!r}") + - raise ValueError(f"oauth not supported for {kind}") +def oauth_state_kind_for_target(target: dict[str, Any]) -> str: + target_type = str(target.get("type") or "").strip() + if target_type == "catalog": + return str(target.get("kind") or "").strip() + if target_type == "custom_mcp": + return CUSTOM_MCP_KIND + raise ValueError(f"unsupported oauth target type: {target_type!r}") + + +def oauth_target_requires_https(target: dict[str, Any]) -> bool: + target_type = str(target.get("type") or "").strip() + if target_type == "catalog": + kind = str(target.get("kind") or "").strip() + return get_mcp_oauth_remote(kind) is not None + return target_type == "custom_mcp" + + +async def start_custom_mcp_oauth( + *, + server_name: str, + mcp_url: str, + redirect_uri: str, + state: str, + discovery: dict[str, Any] | None = None, +) -> tuple[str, str, dict[str, Any]]: + """Start OAuth for a user-defined MCP server (issuer discovered from MCP URL).""" + found = discovery if discovery is not None else await discover_oauth_from_mcp_url(mcp_url) + if not found.get("available"): + raise ValueError(str(found.get("error") or "OAuth not available for this MCP URL")) + issuer = str(found["issuer"]) + resource = str(found.get("resource") or "").strip() or None + metadata = found.get("metadata") + if not isinstance(metadata, dict): + metadata = await fetch_authorization_metadata(issuer) + scopes_raw = metadata.get("scopes_supported") + scope = ( + " ".join(str(s) for s in scopes_raw if s) + if isinstance(scopes_raw, list) and scopes_raw + else None + ) + return await _start_mcp_oauth_from_discovery( + flow="custom_mcp", + kind=CUSTOM_MCP_KIND, + redirect_uri=redirect_uri, + state=state, + issuer=issuer, + resource=resource, + metadata=metadata, + server_name=server_name, + mcp_url=mcp_url, + scope=scope, + ) + + +async def refresh_custom_mcp_oauth(oauth: dict[str, Any]) -> dict[str, Any]: + refresh = str(oauth.get("refresh_token") or "").strip() + issuer = str(oauth.get("oauth_issuer") or "").strip() + client_id = str(oauth.get("oauth_client_id") or "").strip() + if not refresh or not issuer or not client_id: + return oauth + metadata = await fetch_authorization_metadata(issuer) + client_secret_raw = oauth.get("oauth_client_secret") + secret = str(client_secret_raw) if client_secret_raw else None + resource = str(oauth.get("oauth_resource") or "").strip() or None + refreshed = await refresh_access_token( + metadata, + issuer=issuer, + client_id=client_id, + client_secret=secret, + refresh_token=refresh, + resource=resource, + ) + merged = dict(oauth) + merged.update(refreshed) + return merged async def exchange_oauth_code( @@ -172,6 +315,32 @@ async def exchange_oauth_code( ctx = load_oauth_ctx(settings_repo, state_id) flow = ctx.get("flow") + if flow == "custom_mcp": + metadata = ctx.get("metadata") + if not isinstance(metadata, dict): + issuer = str(ctx.get("issuer") or "") + if not issuer: + raise ValueError("missing custom MCP oauth issuer") + metadata = await fetch_authorization_metadata(issuer) + client_id = str(ctx.get("client_id") or "") + client_secret_raw = ctx.get("client_secret") + secret = str(client_secret_raw) if client_secret_raw else None + if not client_id: + raise ValueError("missing custom MCP oauth client_id") + resource = str(ctx.get("resource") or "") or None + exchange_redirect = str(ctx.get("redirect_uri") or "").strip() or redirect_uri + issuer = str(ctx.get("issuer") or "") + return await exchange_authorization_code( + metadata, + issuer=issuer, + client_id=client_id, + client_secret=secret, + code=code, + redirect_uri=exchange_redirect, + code_verifier=code_verifier, + resource=resource, + ) + if flow == "mcp" or kind in mcp_oauth_kinds(): metadata = ctx.get("metadata") if not isinstance(metadata, dict): diff --git a/src/octop/infra/connectors/probe.py b/src/octop/infra/connectors/probe.py index 1ac5dd82..c5c05347 100644 --- a/src/octop/infra/connectors/probe.py +++ b/src/octop/infra/connectors/probe.py @@ -23,6 +23,7 @@ ) from octop.infra.connectors.gateway.protocol import handle_mcp_request from octop.infra.connectors.gateway.registry import probe_gateway_credentials +from octop.infra.connectors.oauth.discovery import discover_oauth_from_mcp_url from octop.infra.utils.ssrf_guard import UnsafeOutboundUrl, safe_request logger = logging.getLogger(__name__) @@ -124,6 +125,30 @@ def _probe_mcp_http_error(exc: httpx.HTTPStatusError, *, kind: str) -> dict[str, } +async def _maybe_attach_oauth_discovery( + result: dict[str, Any], + *, + url: str, + headers: dict[str, str], +) -> dict[str, Any]: + """When auth fails without a bearer token, try MCP OAuth discovery.""" + if result.get("ok") is not False or result.get("error_type") != "auth": + return result + auth_header = str(headers.get("Authorization") or "").strip() + if auth_header.lower().startswith("bearer ") and len(auth_header) > 7: + return result + discovery = await discover_oauth_from_mcp_url(url) + if discovery.get("available"): + result["oauth"] = { + "available": True, + "issuer": discovery.get("issuer"), + "resource": discovery.get("resource"), + } + else: + result["oauth"] = {"available": False} + return result + + def _probe_mcp_mcp_error(exc: McpError, *, kind: str) -> dict[str, Any]: """Map an MCP-level error (e.g. server closing the initialize stream). @@ -431,6 +456,16 @@ async def probe_connector( } +def format_probe_exception(exc: BaseException) -> str: + """Flatten TaskGroup / ExceptionGroup errors for API responses.""" + if isinstance(exc, BaseExceptionGroup): + parts = [format_probe_exception(sub) for sub in exc.exceptions] + joined = "; ".join(part for part in parts if part) + return joined or str(exc) + msg = str(exc).strip() + return msg or type(exc).__name__ + + async def probe_custom_mcp_server(spec: dict[str, Any]) -> dict[str, Any]: """Probe one user-defined MCP server (streamable_http or stdio).""" from octop.infra.connectors.custom_mcp import harness_spec_for_server, normalize_server_spec @@ -448,7 +483,8 @@ async def probe_custom_mcp_server(spec: dict[str, Any]) -> dict[str, Any]: headers = {str(k): str(v) for k, v in dict(connection.get("headers") or {}).items()} # Ensure streamable Accept if caller omitted it. headers.setdefault("Accept", "application/json, text/event-stream") - return await probe_streamable_http_mcp(url, headers, kind="custom-mcp") + result = await probe_streamable_http_mcp(url, headers, kind="custom-mcp") + return await _maybe_attach_oauth_discovery(result, url=url, headers=headers) if transport == "stdio": return await _probe_stdio_mcp(connection) @@ -485,4 +521,4 @@ async def _probe_stdio_mcp(connection: dict[str, Any]) -> dict[str, Any]: return {"ok": False, "error": "stdio MCP probe timed out"} except Exception as exc: logger.exception("stdio MCP probe failed") - return {"ok": False, "error": str(exc)} + return {"ok": False, "error": format_probe_exception(exc)} diff --git a/src/octop/infra/connectors/service.py b/src/octop/infra/connectors/service.py index c4c1eae2..b0128ce7 100644 --- a/src/octop/infra/connectors/service.py +++ b/src/octop/infra/connectors/service.py @@ -14,11 +14,17 @@ from octop.infra.connectors.custom_mcp import ( CUSTOM_MCP_DISPLAY_NAME, CUSTOM_MCP_KIND, + build_oauth_storage, enabled_harness_configs, expand_custom_instances, extract_servers, is_custom_mcp_kind, + merge_preserved_oauth, + oauth_configured, + oauth_tokens_from_spec, + redact_servers_for_api, server_enabled, + set_oauth_required_in_spec, validate_servers_map, wrap_servers, ) @@ -29,6 +35,7 @@ start_user_device_login, ) from octop.infra.connectors.oauth import refresh_oauth_credentials +from octop.infra.connectors.oauth.registry import refresh_custom_mcp_oauth from octop.infra.db.repos.connectors import ConnectorRepo, ConnectorRow from octop.infra.db.repos.secrets import SecretRepo from octop.infra.errors import ErrorCode, OctopError @@ -140,7 +147,15 @@ def get_custom_servers(self, user_id: int) -> dict[str, Any]: return {} return extract_servers(self.decrypt(row.instance_id)) + def get_custom_servers_for_api(self, user_id: int) -> dict[str, Any]: + return redact_servers_for_api(self.get_custom_servers(user_id)) + def put_custom_servers(self, user_id: int, servers: dict[str, Any]) -> dict[str, Any]: + existing = self.get_custom_servers(user_id) + merged = merge_preserved_oauth(servers, existing) + return self._save_custom_servers(user_id, merged) + + def _save_custom_servers(self, user_id: int, servers: dict[str, Any]) -> dict[str, Any]: normalized = validate_servers_map( servers, reserved_names=self.reserved_builtin_mcp_names(user_id), @@ -167,38 +182,110 @@ def put_custom_servers(self, user_id: int, servers: dict[str, Any]) -> dict[str, ) return normalized + def apply_custom_server_oauth( + self, + user_id: int, + server_name: str, + tokens: dict[str, Any], + *, + issuer: str, + resource: str | None, + ) -> dict[str, Any]: + servers = dict(self.get_custom_servers(user_id)) + if server_name not in servers: + raise KeyError(server_name) + access = str(tokens.get("access_token") or "").strip() + if not access: + raise ValueError("missing access_token") + spec = dict(servers[server_name]) + spec["oauth"] = build_oauth_storage(tokens, issuer=issuer, resource=resource) + spec = set_oauth_required_in_spec(spec, required=False) + servers[server_name] = spec + return self._save_custom_servers(user_id, servers) + + async def ensure_fresh_custom_servers(self, user_id: int) -> dict[str, Any]: + servers = dict(self.get_custom_servers(user_id)) + changed = False + now = int(time.time()) + for name, raw in list(servers.items()): + if not isinstance(raw, dict): + continue + oauth = oauth_tokens_from_spec(raw) + if not oauth_configured(raw): + continue + expires_at = oauth.get("expires_at") + refresh = str(oauth.get("refresh_token") or "").strip() + if not refresh: + continue + if expires_at and int(expires_at) > now + 120: + continue + try: + refreshed = await refresh_custom_mcp_oauth(oauth) + except Exception: + continue + spec = dict(raw) + spec["oauth"] = refreshed + servers[name] = spec + changed = True + if not changed: + return servers + return self._save_custom_servers(user_id, servers) + def patch_custom_server_enabled( self, user_id: int, server_name: str, *, enabled: bool, + ) -> dict[str, Any]: + return self.patch_custom_server(user_id, server_name, enabled=enabled) + + def patch_custom_server( + self, + user_id: int, + server_name: str, + *, + enabled: bool | None = None, + default_open: bool | None = None, ) -> dict[str, Any]: servers = dict(self.get_custom_servers(user_id)) if server_name not in servers: raise KeyError(server_name) spec = dict(servers[server_name]) - spec["enabled"] = enabled + if enabled is not None: + spec["enabled"] = enabled + if not enabled: + spec.pop("default_open", None) + if default_open is not None: + if default_open: + spec["default_open"] = True + else: + spec.pop("default_open", None) servers[server_name] = spec return self.put_custom_servers(user_id, servers) - def patch_custom_server_default_open( + def note_custom_server_oauth_required( self, user_id: int, server_name: str, *, - default_open: bool, + required: bool, ) -> dict[str, Any]: servers = dict(self.get_custom_servers(user_id)) if server_name not in servers: raise KeyError(server_name) spec = dict(servers[server_name]) - if default_open: - spec["default_open"] = True - else: - spec.pop("default_open", None) - servers[server_name] = spec - return self.put_custom_servers(user_id, servers) + servers[server_name] = set_oauth_required_in_spec(spec, required=required) + return self._save_custom_servers(user_id, servers) + + def patch_custom_server_default_open( + self, + user_id: int, + server_name: str, + *, + default_open: bool, + ) -> dict[str, Any]: + return self.patch_custom_server(user_id, server_name, default_open=default_open) def list_instances_for_api(self, user_id: int) -> list[dict[str, Any]]: """Built-in rows + expanded custom servers (hide parent custom-mcp row).""" @@ -307,6 +394,7 @@ async def mcp_configs_for_user(self, user_id: int) -> dict[str, Any]: creds=creds, config=self._config, ) + await self.ensure_fresh_custom_servers(user_id) configs.update(self.custom_harness_configs(user_id)) return configs diff --git a/tests/integration/test_connectors_api.py b/tests/integration/test_connectors_api.py index 1f377958..4fcb61e2 100644 --- a/tests/integration/test_connectors_api.py +++ b/tests/integration/test_connectors_api.py @@ -206,7 +206,7 @@ async def test_oauth_start_public_http_notion_error_is_actionable(tmp_octop_home await bootstrap_admin(c, tmp_octop_home) auth = await auth_header(c) mocked_start = AsyncMock() - with patch("octop.api.routers.connectors.start_oauth", mocked_start): + with patch("octop.api.routers.connectors.start_oauth_for_target", mocked_start): r = await c.post( "/api/connectors/oauth/notion/start", headers={**auth, "host": "58.87.70.170"}, @@ -305,3 +305,39 @@ async def test_cli_status_available_to_non_admin(env, monkeypatch: pytest.Monkey r = await c.get("/api/connectors/feishu-cli/cli-status", headers=user_auth) assert r.status_code == 200 assert r.json()["installed"] is False + + +async def test_patch_custom_mcp_server_default_open_only(env): + c, _, auth, _ = env + put = await c.put( + "/api/connectors/custom-mcp", + headers=auth, + json={ + "servers": { + "linear": { + "transport": "streamable_http", + "url": "https://mcp.linear.app/mcp", + "enabled": True, + } + } + }, + ) + assert put.status_code == 200 + + patch = await c.patch( + "/api/connectors/custom-mcp/servers/linear", + headers=auth, + json={"default_open": True}, + ) + assert patch.status_code == 200 + servers = patch.json()["servers"] + assert servers["linear"]["default_open"] is True + assert servers["linear"]["enabled"] is True + + patch_off = await c.patch( + "/api/connectors/custom-mcp/servers/linear", + headers=auth, + json={"default_open": False}, + ) + assert patch_off.status_code == 200 + assert "default_open" not in patch_off.json()["servers"]["linear"] diff --git a/tests/unit/connectors/test_oauth_discovery.py b/tests/unit/connectors/test_oauth_discovery.py new file mode 100644 index 00000000..5658ed5b --- /dev/null +++ b/tests/unit/connectors/test_oauth_discovery.py @@ -0,0 +1,106 @@ +"""Unit tests for MCP OAuth discovery and unified oauth targets.""" + +from __future__ import annotations + +import pytest + +from octop.infra.connectors.custom_mcp import ( + build_oauth_storage, + harness_spec_for_server, + merge_preserved_oauth, + redact_server_for_api, +) +from octop.infra.connectors.oauth.discovery import ( + build_protected_resource_metadata_urls, + parse_www_authenticate_resource_metadata, +) +from octop.infra.connectors.oauth.registry import ( + oauth_state_kind_for_target, + oauth_target_requires_https, +) + + +def test_build_protected_resource_metadata_urls_path_and_root(): + urls = build_protected_resource_metadata_urls("https://example.com/mcp") + assert urls == [ + "https://example.com/.well-known/oauth-protected-resource/mcp", + "https://example.com/.well-known/oauth-protected-resource", + ] + + +def test_build_protected_resource_metadata_urls_prefers_www_auth_header(): + header_url = "https://example.com/.well-known/oauth-protected-resource/custom" + urls = build_protected_resource_metadata_urls( + "https://example.com/mcp", + www_auth_resource_metadata=header_url, + ) + assert urls[0] == header_url + + +def test_parse_www_authenticate_resource_metadata(): + raw = 'Bearer error="invalid_token", resource_metadata="https://mcp.example.com/prm"' + assert parse_www_authenticate_resource_metadata(raw) == "https://mcp.example.com/prm" + + +def test_oauth_target_helpers(): + assert oauth_state_kind_for_target({"type": "catalog", "kind": "notion"}) == "notion" + assert oauth_state_kind_for_target({"type": "custom_mcp", "server_name": "x"}) == ("custom-mcp") + assert oauth_target_requires_https({"type": "catalog", "kind": "notion"}) is True + assert oauth_target_requires_https({"type": "custom_mcp", "server_name": "x"}) is True + assert oauth_target_requires_https({"type": "catalog", "kind": "tencent-docs"}) is False + + +def test_merge_preserved_oauth_and_redaction(): + existing = { + "srv": { + "transport": "streamable_http", + "url": "https://example.com/mcp", + "oauth": {"access_token": "secret", "expires_at": 123}, + } + } + incoming = { + "srv": { + "transport": "streamable_http", + "url": "https://example.com/mcp", + "oauth": {"access_token": "client-sent"}, + } + } + merged = merge_preserved_oauth(incoming, existing) + assert merged["srv"]["oauth"]["access_token"] == "secret" + preview = redact_server_for_api(merged["srv"]) + assert preview["oauth"] == {"configured": True, "expires_at": 123} + assert "access_token" not in preview["oauth"] + + +def test_oauth_required_hint_redaction_and_merge(): + existing = { + "srv": { + "transport": "streamable_http", + "url": "https://example.com/mcp", + "oauth": {"required": True}, + } + } + incoming = {"srv": {"transport": "streamable_http", "url": "https://example.com/mcp"}} + merged = merge_preserved_oauth(incoming, existing) + assert merged["srv"]["oauth"] == {"required": True} + preview = redact_server_for_api(merged["srv"]) + assert preview["oauth"] == {"configured": False, "required": True} + + +def test_harness_spec_injects_oauth_bearer(): + spec = { + "transport": "streamable_http", + "url": "https://example.com/mcp", + "oauth": build_oauth_storage( + {"access_token": "tok123", "refresh_token": "r", "expires_at": 1}, + issuer="https://example.com", + resource="https://example.com/mcp", + ), + } + harness = harness_spec_for_server(spec) + assert harness["headers"]["Authorization"] == "Bearer tok123" + + +def test_oauth_state_kind_invalid(): + with pytest.raises(ValueError, match="unsupported oauth target"): + oauth_state_kind_for_target({"type": "unknown"})