2025-06-09 一個 MCP server 同時連兩種 AI（ChatGPT / OpenAI 與 DeepSeek）」的整體設計

發表於 2025-10-01 分類於 OneDayOneNote 閱讀次數：

好 — 我幫你把「一個 MCP server 同時連兩種 AI（ChatGPT / OpenAI 與 DeepSeek）」的整體設計、實作重點、風險與範例程式碼一次講清楚，讓你能立刻把它接進去。

核心概念（一句話）

把 MCP server 當成 多供應商 AI Gateway：接收前端請求 → 決定要用哪個 AI（或同時多個） → 呼叫對應供應商 API → 將回應標準化並回到前端或 MCP client。

（DeepSeek 提供類 OpenAI 風格的 chat API；多供應商 gateway 是常見做法，可用於路由、降級、成本控制與安全審查）。(api-docs.deepseek.com)

設計要點（必讀）

抽象化 Provider 層：為每個 AI 寫一個「adapter/driver」，讓上層呼叫統一介面（例如 getChatResponse({messages, opts})），內部負責 provider-specific 的轉換與呼叫。這樣新增/換模型時只需新增 adapter。(GitHub)
路由策略（如何決定用哪個模型）：
- 依使用者選擇（UI 提供 switch）
- 根據功能（例如：長篇寫作給 ChatGPT；快速搜尋型回覆給 DeepSeek）
- 成本/速率路由（若超出費用上限就改用較便宜模型）
一致化輸入/輸出：不同廠商回傳格式不同（choices、message、content 等），要在 gateway 標準化成你 MCP client 要的格式。(kgateway.dev)
串流（stream）支援：若有即時需求，確認兩邊是否支援 SSE/WS 或 streaming flag，且在 adapter 處理逐塊轉發。DeepSeek 與 OpenAI 都提供非同步 streaming 選項（但實作細節不同）。(api-docs.deepseek.com)
錯誤 / 降級 & 超時：一個 provider 失敗時能自動 failover（例如轉到另一個模型或回傳友善錯誤）。
安全與合規：把 API keys 放在 server 環境變數；若某供應商在特定議題上有行為差異（如新聞報導指出 DeepSeek 在敏感議題上可能表現不一），在路由與審核邏輯加入白/黑名單或額外的 moderation。(The Washington Post)
計費/日誌/速率限制：記 tokens、cost、latency，必要時實作速率限制、quota。可用中介層（cache）減少重複請求。(Medium)

簡單的 Node.js 範例（概念性、可直接貼用）

下面示範一個最小 Gateway：/api/chat 接收 provider（”openai” 或 “deepseek”）與 messages，再呼叫對應 adapter。範例用 fetch（Node 18+），真實環境要做好錯誤處理、驗證、logging、速率限制、以及 stream 支援（這裡只示單次回應）。

.env

1 2	OPENAI_API_KEY=sk-xxxxx DEEPSEEK_API_KEY=ds-xxxxx

server.mjs

import express from 'express';
import dotenv from 'dotenv';
dotenv.config();
const app = express();
app.use(express.json());

// provider adapter registry
const providers = {
  openai: async ({messages, opts}) => {
    const resp = await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        model: opts?.model || 'gpt-4o-mini',
        messages
      })
    });
    const data = await resp.json();
    // standardize: return text and raw
    return {
      text: data?.choices?.[0]?.message?.content ?? '',
      raw: data
    };
  },

  deepseek: async ({messages, opts}) => {
    // DeepSeek 的 API 與 OpenAI 相似（多數情況下互換），但確認官方 doc 欄位
    const resp = await fetch('https://api.deepseek.com/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${process.env.DEEPSEEK_API_KEY}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        model: opts?.model || 'deepseek-v3.1',
        messages
      })
    });
    const data = await resp.json();
    return {
      text: data?.choices?.[0]?.message?.content ?? '',
      raw: data
    };
  }
};

app.post('/api/chat', async (req, res) => {
  try {
    const { provider = 'openai', messages, opts } = req.body;
    if (!messages) return res.status(400).json({ error: 'missing messages' });
    const adapter = providers[provider];
    if (!adapter) return res.status(400).json({ error: 'unknown provider' });

    // basic failover example: 如果第一個 provider 失敗，嘗試 fallback
    try {
      const { text, raw } = await adapter({ messages, opts });
      return res.json({ provider, text, raw });
    } catch (err) {
      console.error(`provider ${provider} failed:`, err);
      // example fallback to other provider
      const fallback = provider === 'openai' ? providers['deepseek'] : providers['openai'];
      try {
        const { text, raw } = await fallback({ messages, opts });
        return res.json({ provider: provider + '_fallback', text, raw });
      } catch (err2) {
        console.error('fallback failed:', err2);
        return res.status(500).json({ error: 'all providers failed' });
      }
    }
  } catch (err) {
    console.error(err);
    res.status(500).json({ error: 'server error' });
  }
});

app.listen(3000, ()=> console.log('AI gateway listening on 3000'));

注意：DeepSeek 與 OpenAI 的 endpoint 路徑、參數或 header 可能會隨時間有差異。請以官方文件為主，DeepSeek 官方文件可以參考其 API docs。(api-docs.deepseek.com)

進階功能建議（部署時再加）

路由規則管理 UI：可在後台設定哪些 prompt / user 用哪個 provider（方便 AB 測試）。
多路並發 / 條件聚合：同一個 prompt 同時問兩個 provider，將答案合併或做可信度比對後再回給使用者（多模型投票/融合）。
安全內容審查 pipeline：先把 user prompt 經過 moderation（可使用 OpenAI /自身規則），必要時拒絕或轉人工審核。
token & cost 控制：每次呼叫前估算 tokens，超出閾值則轉到 cheaper model 或拒絕。
監控 & 告警：監控 latency、error rate 與花費；異常時自動切流。(AWS 解決方案庫範本)

風險提醒（特別重要）

內容與安全性：最近報導指出 DeepSeek 在某些敏感題材或地緣政治議題上回答品質或傾向可能與其它國家供應商不同（在某些情況下產生不安全或不穩定的程式碼建議）。若你的應用會處理敏感議題或生成可執行程式碼，務必在 pipeline 加強審查與測試。(The Washington Post)
法規/資料主權：若有嚴格資料主權要求（例如不能把特定資料送到境外），必須在路由規則上控管哪些請求可以被哪些 provider 接收。
可靠性差異：不同 provider 的可用性/延遲/計費差異會影響 UX，要在 UX 層處理好等待/降級行為。

我可以幫你做什麼（選一個）

幫你把你現有的 MCP server 程式碼改寫為上面那種 provider-adapter 架構（請貼 server 主要檔案）。
提供「支援 stream 的 adapter」範例（如果你要進行即時串流、SSE 或 websocket）。
幫你把路由策略與成本/安全檢查寫成可配置的中介層（config + UI）。

你偏好哪一項？或直接把你目前的 server 程式碼貼上（告訴我 Node 或 Python），我就直接改給你可貼入的程式。