iT邦幫忙

2025 iThome 鐵人賽

DAY 18
0

看到n8n中的ai agent可以用mcp server,來測試是否能行,可以的話就不用使用claude desktop了

首先在右上角搜尋並使用ai agent,並在tool那選擇mcp server
https://ithelp.ithome.com.tw/upload/images/20251002/201779202Vuuf8G3FV.png
https://ithelp.ithome.com.tw/upload/images/20251002/20177920l1pd8ytyxN.png
更改之前的FASTMCP

#!/usr/bin/env python3
"""
n8n MCP Server - 專為 n8n 工作流程最佳化的 MCP 伺服器
基於 mcp_from_openapi.py 並針對 n8n 的 HTTP Streamable 傳輸進行最佳化
"""

import httpx
import yaml
import os
import sys
import uvicorn
import logging
from fastmcp import FastMCP

# 設定日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def load_openapi_spec():
    """載入並配置 OpenAPI 規範"""
    script_dir = os.path.dirname(os.path.abspath(__file__))
    openapi_path = os.path.join(script_dir, "openapi.yaml")
    
    if not os.path.exists(openapi_path):
        logger.error(f"找不到 OpenAPI 檔案: {openapi_path}")
        sys.exit(1)
    
    try:
        with open(openapi_path, "r", encoding="utf-8") as f:
            openapi_spec = yaml.safe_load(f)
        
        # 更新 server URL 指向 REST API
        openapi_spec["servers"] = [
            {"url": "http://localhost:8001", "description": "REST API 伺服器"}
        ]
        
        logger.info("✅ OpenAPI 規範載入並配置成功")
        return openapi_spec
        
    except Exception as e:
        logger.error(f"載入 OpenAPI 規範失敗: {e}")
        sys.exit(1)

def create_http_client():
    """建立 HTTP 客戶端連接到 REST API"""
    try:
        client = httpx.AsyncClient(
            base_url="http://localhost:8001",
            timeout=30.0,
            verify=False  # 本地開發環境
        )
        logger.info("✅ HTTP 客戶端建立成功")
        return client
        
    except Exception as e:
        logger.error(f"建立 HTTP 客戶端失敗: {e}")
        sys.exit(1)

def create_mcp_server(openapi_spec, client):
    """建立 MCP 伺服器實例"""
    try:
        mcp = FastMCP.from_openapi(
            openapi_spec=openapi_spec,
            client=client,
            name="n8n Weather & Earthquake MCP Server"
        )
        logger.info("✅ MCP 伺服器建立成功")
        return mcp
        
    except Exception as e:
        logger.error(f"建立 MCP 伺服器失敗: {e}")
        sys.exit(1)

def check_port_available(port):
    """檢查端口是否可用"""
    import socket
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        result = s.connect_ex(('127.0.0.1', port))
        return result != 0

def find_available_port(start_port=9001, max_attempts=10):
    """尋找可用的端口"""
    for i in range(max_attempts):
        port = start_port + i
        if check_port_available(port):
            return port
    
    logger.error(f"無法找到可用端口 (嘗試範圍: {start_port}-{start_port + max_attempts - 1})")
    sys.exit(1)

def main():
    
    # 載入 OpenAPI 規範
    openapi_spec = load_openapi_spec()
    
    # 建立 HTTP 客戶端
    client = create_http_client()
    
    # 建立 MCP 伺服器
    mcp = create_mcp_server(openapi_spec, client)
    
    # 尋找可用端口
    port = find_available_port(9001)
    
    try:
        # 使用 streamable_http_app 方法支援 HTTP Streamable 傳輸
        app = mcp.streamable_http_app()
    
        # 啟動 HTTP 伺服器
        uvicorn.run(
            app,
            host="127.0.0.1",
            port=port,
            log_level="info",
            access_log=True
        )
        
    except KeyboardInterrupt:
        logger.info("伺服器已停止")
        
    except Exception as e:
        logger.error(f"伺服器執行錯誤: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()
openapi: 3.0.3
info:
  title: My MCP APIs
  version: 1.0.0
  description: APIopenapi: 3.0.3
info:
  title: My MCP APIs
  version: 1.0.0
  description: API
servers:
  - url: http://localhost:8000
    description: 本地開發伺服器
components:
  securitySchemes:
    ApiKeyAuth:
      type: apiKey
      in: header
      name: Authorization
      description: "CWB API Key for weather data access"
    GeminiApiKeyAuth:
      type: apiKey
      in: header
      name: x-goog-api-key
      description: "Google Gemini API Key for AI chat functionality"
paths:
  /weather:
    post:
      operationId: getWeather
      summary: 查詢天氣
      description: 根據縣市名稱查詢天氣資訊
      security:
        - ApiKeyAuth: []
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required:
                - LocationName
              properties:
                LocationName:
                  type: string
                  description: "臺灣各縣市名稱,例如:臺中市、台北市、高雄市"
                  example: "臺中市"
                ElementName:
                  type: string
                  description: "天氣預報因子,多個要素用逗號分隔"
                  example: "最高溫度,最低溫度,天氣預報綜合描述"
                  default: "最高溫度,最低溫度,天氣預報綜合描述"
                limit:
                  type: integer
                  description: "限制回傳筆數"
                  example: 10
                  default: 10
                offset:
                  type: integer
                  description: "資料偏移量"
                  example: 0
                  default: 0
      responses:
        '200':
          description: 天氣結果
          content:
            application/json:
              schema:
                type: object
                properties:
                  city:
                    type: string
                    description: "城市名稱"
                  temperature:
                    type: string
                    description: "溫度資訊"
                  description:
                    type: string
                    description: "天氣描述"
  /earthquake:
    post:
      operationId: getEarthquakes
      summary: 取得地震資料
      description: 可指定縣市或取得全部最新地震
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              properties:
                target_areas:
                  type: string
                  description: "目標縣市名稱,例如:臺中市、台北市"
                  example: "臺中市"
                limit:
                  type: string
                  description: "限制回傳地震筆數"
                  example: "5"
                  default: "5"
      responses:
        '200':
          description: 地震查詢結果
          content:
            application/json:
              schema:
                type: object
                properties:
                  success:
                    type: boolean
                  count:
                    type: integer
                  earthquakes:
                    type: array
                    items:
                      type: object
                      properties:
                        earthquake_no:
                          type: string
                        origin_time:
                          type: string
                        location:
                          type: string
                        magnitude:
                          type: string
                        depth:
                          type: string
                        latitude:
                          type: string
                        longitude:
                          type: string
                  target_areas:
                    type: array
                    items:
                      type: string
                    description: "如果指定了特定縣市"
        '400':
          description: 請求錯誤
          content:
            application/json:
              schema:
                type: object
                properties:
                  error:
                    type: string

servers:
  - url: http://localhost:8000
    description: 本地開發伺服器
components:
  securitySchemes:
    ApiKeyAuth:
      type: apiKey
      in: header
      name: Authorization
      description: "CWB API Key for weather data access"
    GeminiApiKeyAuth:
      type: apiKey
      in: header
      name: x-goog-api-key
      description: "Google Gemini API Key for AI chat functionality"
paths:
  /weather:
    post:
      operationId: getWeather
      summary: 查詢天氣
      description: 根據縣市名稱查詢天氣資訊
      security:
        - ApiKeyAuth: []
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required:
                - LocationName
              properties:
                LocationName:
                  type: string
                  description: "臺灣各縣市名稱,例如:臺中市、台北市、高雄市"
                  example: "臺中市"
                ElementName:
                  type: string
                  description: "天氣預報因子,多個要素用逗號分隔"
                  example: "最高溫度,最低溫度,天氣預報綜合描述"
                  default: "最高溫度,最低溫度,天氣預報綜合描述"
                limit:
                  type: integer
                  description: "限制回傳筆數"
                  example: 10
                  default: 10
                offset:
                  type: integer
                  description: "資料偏移量"
                  example: 0
                  default: 0
      responses:
        '200':
          description: 天氣結果
          content:
            application/json:
              schema:
                type: object
                properties:
                  city:
                    type: string
                    description: "城市名稱"
                  temperature:
                    type: string
                    description: "溫度資訊"
                  description:
                    type: string
                    description: "天氣描述"
  /earthquake:
    post:
      operationId: getEarthquakes
      summary: 取得地震資料
      description: 可指定縣市或取得全部最新地震
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              properties:
                target_areas:
                  type: string
                  description: "目標縣市名稱,例如:臺中市、台北市"
                  example: "臺中市"
                limit:
                  type: string
                  description: "限制回傳地震筆數"
                  example: "5"
                  default: "5"
      responses:
        '200':
          description: 地震查詢結果
          content:
            application/json:
              schema:
                type: object
                properties:
                  success:
                    type: boolean
                  count:
                    type: integer
                  earthquakes:
                    type: array
                    items:
                      type: object
                      properties:
                        earthquake_no:
                          type: string
                        origin_time:
                          type: string
                        location:
                          type: string
                        magnitude:
                          type: string
                        depth:
                          type: string
                        latitude:
                          type: string
                        longitude:
                          type: string
                  target_areas:
                    type: array
                    items:
                      type: string
                    description: "如果指定了特定縣市"
        '400':
          description: 請求錯誤
          content:
            application/json:
              schema:
                type: object
                properties:
                  error:
                    type: string

修改N8N上的MCP SERVER
https://ithelp.ithome.com.tw/upload/images/20251002/20177920bfm1zlkzuZ.png
確定有東西出來之後CHAT MODEL選擇Google Gemini Chat Model
今天先確認mcp server能否使用
https://ithelp.ithome.com.tw/upload/images/20251002/20177920qtCvoVCruq.png


上一篇
D17
系列文
這是一個一個一個 Python API 與 Gemini 整合、n8n入門指南18
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言