iT邦幫忙

2025 iThome 鐵人賽

DAY 8
0

FastMCP是什麼?
專門用來快速開發 MCP 伺服器的工具

FastMCP 特點:

  • 可自動讀取 OpenAPI 規格檔案
  • 支援非同步操作
  • 內建工具函式註冊機制(使用 @app.tool 裝飾器)

Swagger3.0/OpenAPI是什麼
用 YAML 或 JSON 描述 API 的規格,AI 可以直接讀這個檔案,就知道有哪些 API、參數格式、回傳資料型態

實作時間
先到中央氣象署的網站申請API金鑰 https://opendata.cwa.gov.tw
https://ithelp.ithome.com.tw/upload/images/20250922/20177920hi1xGdxcCY.png

在.env檔中放入你的金鑰

CWB_API_KEY="你的金鑰"

記得先進虛擬環境

安裝FASTMCP及httpx(支援同步和非同步的request套件)還有查看openapi格式是否正確的套件

pip install fastmcp
pip install httpx
pip install openapi-spec-validator

建立一個"自訂名稱".yaml檔案,在裡面輸入OpenAPI格式的接收要求

oopenapi: 3.0.3 #swagger版本
info:
  title: My MCP APIs #這個文件是什麼
  version: 1.0.0
  description: 提供天氣查詢的 API #作用
servers:
  - url: http://localhost:8000 #伺服器會開在哪裡
    description: 本地開發伺服器
components:
  securitySchemes: #放安全性相關的東西,向api_key
    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:
      summary: 查詢天氣
      security:
        - ApiKeyAuth: [] #需要金鑰
      requestBody:
        required: true #一定要輸入
        content:
          application/json:
            schema:
              type: object
              properties:
                LocationName:
                  type: array
                  items:
                    type: string
                  description: "臺灣各縣市名稱列表"
                  example: ["臺中市"]
                ElementName:
                  type: array
                  items:
                    type: string
                  description: "天氣預報因子,多個要素用逗號分隔"
                  example: "最高溫度,最低溫度,天氣預報綜合描述"
                limit:
                  type: integer
                  example: 10
                offset:
                  type: integer
                  example: 0
      responses:
        '200':
          description: 天氣結果
          content:
            application/json:
              schema:
                type: object
                properties:
                  city:
                    type: string
                  temperature:
                    type: string
                  description:
                    type: string
  /gemini:
    post:
      summary: 呼叫 Gemini AI 聊天
      security:
        - GeminiApiKeyAuth: []
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              properties:
                message:
                  type: string
                  description: "要發送給 Gemini 的訊息"
                  example: "你好,請介紹一下台灣的天氣特色"
                model:
                  type: string
                  description: "Gemini 模型名稱"
                  example: "gemini-2.5-flash"
                  default: "gemini-2.5-flash"
                temperature:
                  type: number
                  description: "回應的創造性程度 (0-1)"
                  example: 0.7
                  default: 0.7
                max_tokens:
                  type: integer
                  description: "最大回應長度"
                  example: 1000
                  default: 1000
              required:
                - message
      responses:
        '200':
          description: Gemini AI 回應
          content:
            application/json:
              schema:
                type: object
                properties:
                  success:
                    type: boolean
                  response:
                    type: string
                    description: "Gemini 的回應內容"
                  model_used:
                    type: string
                    description: "使用的模型名稱"
                  usage:
                    type: object
                    properties:
                      prompt_tokens:
                        type: integer
                      completion_tokens:
                        type: integer
                      total_tokens:
                        type: integer
        '400':
          description: 請求錯誤
          content:
            application/json:
              schema:
                type: object
                properties:
                  error:
                    type: string

好了之後退出虛擬環境後在終端機上輸入

openapi-spec-validator "自訂名稱".yaml

如果顯示:

openapi.yaml: OK

代表沒有做錯
回到虛擬環境後再來在python檔案中輸入

import httpx
import os
import asyncio
import urllib3
from typing import Optional
from dotenv import load_dotenv

# 載入環境變數
load_dotenv()

# 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# 設定環境變數來禁用 SSL 驗證
os.environ['PYTHONHTTPSVERIFY'] = '0'
os.environ['CURL_CA_BUNDLE'] = ''

# 從環境變數取得 API 金鑰
CWA_API_KEY = os.getenv("CWB_API_KEY")

VALID_LOCATIONS = [
    "宜蘭縣", "花蓮縣", "臺東縣", "澎湖縣", "金門縣", "連江縣",
    "臺北市", "新北市", "桃園市", "臺中市", "臺南市", "高雄市",
    "基隆市", "新竹縣", "新竹市", "苗栗縣", "彰化縣", "南投縣",
    "雲林縣", "嘉義縣", "嘉義市", "屏東縣"
]

# 支援的天氣要素
VALID_ELEMENTS = [
    "最高溫度", "天氣預報綜合描述", "平均相對濕度", "最高體感溫度",
    "12小時降雨機率", "風向", "平均露點溫度", "最低體感溫度",
    "平均溫度", "最大舒適度指數", "最小舒適度指數", "風速",
    "紫外線指數", "天氣現象", "最低溫度"
]

class WeatherAPI:
    def __init__(self):![](http://)
        self.api_key = CWA_API_KEY
        self.base_url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-D0047-091"
    
    async def get_weather_forecast(
        self,
        LocationName: Optional[str] = None,
        ElementName: Optional[str] = None,
        limit: Optional[int] = None,
        offset: Optional[int] = None,
        format: str = "JSON",
        sort: Optional[str] = None,
        StartTime: Optional[str] = None,
        timeFrom: Optional[str] = None,
        timeTo: Optional[str] = None
    ):
        """
        取得台灣氣象開放資料平台天氣預報
        
        Args:
            LocationName: 臺灣各縣市名稱,多個縣市用逗號分隔
            ElementName: 天氣預報因子,多個要素用逗號分隔
            limit: 限制最多回傳的資料筆數
            offset: 指定從第幾筆後開始回傳
            format: 回傳資料格式 (預設為 JSON)
            sort: 對時間做升冪排序
            StartTime: 時間因子,多個時間用逗號分隔,格式為 yyyy-MM-ddThh:mm:ss
            timeFrom: 時間區段開始時間
            timeTo: 時間區段結束時間
            
        Returns:
            氣象預報資料或錯誤訊息
        """
        if not self.api_key:
            return {"error": "未設定 CWB_API_KEY 環境變數"}
        
        # 設定請求標頭
        headers = {
            "Authorization": self.api_key
        }
        
        # 建立查詢參數
        params = {
            "format": format
        }
        
        # 處理 LocationName 參數
        if LocationName:
            if isinstance(LocationName, str):
                location_list = [loc.strip() for loc in LocationName.split(",")]
            else:
                location_list = LocationName
            
            # 驗證縣市名稱
            invalid_locations = [loc for loc in location_list if loc not in VALID_LOCATIONS]
            if invalid_locations:
                return {"error": f"無效的縣市名稱: {invalid_locations}"}
            params["LocationName"] = ",".join(location_list)
        
        # 處理 ElementName 參數
        if ElementName:
            if isinstance(ElementName, str):
                element_list = [elem.strip() for elem in ElementName.split(",")]
            else:
                element_list = ElementName
            
            # 驗證天氣要素
            invalid_elements = [elem for elem in element_list if elem not in VALID_ELEMENTS]
            if invalid_elements:
                return {"error": f"無效的天氣要素: {invalid_elements}"}
            params["ElementName"] = ",".join(element_list)
        
        if limit is not None:
            params["limit"] = str(limit)
        
        if offset is not None:
            params["offset"] = str(offset)
        
        if sort:
            params["sort"] = sort
        
        if StartTime:
            if isinstance(StartTime, str):
                params["StartTime"] = StartTime
            else:
                params["StartTime"] = ",".join(StartTime)
        
        if timeFrom:
            params["timeFrom"] = timeFrom
        
        if timeTo:
            params["timeTo"] = timeTo
        
        try:
            async with httpx.AsyncClient(timeout=30.0, verify=False) as client:
                response = await client.get(self.base_url, headers=headers, params=params)
                response.raise_for_status()
                
                data = response.json()
                
                # 檢查 API 回應是否成功
                if data.get("success") != "true":
                    return {"error": f"API 回應錯誤: {data.get('result', {}).get('message', '未知錯誤')}"}
                
                return {
                    "success": True,
                    "data": data.get("records", {}),
                    "request_params": params
                }
                
        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP 錯誤 {e.response.status_code}"}
        except httpx.RequestError as e:
            return {"error": f"連線錯誤: {str(e)}"}
        except Exception as e:
            return {"error": f"未預期的錯誤: {str(e)}"}

    def format_weather_data(self, data):
        """格式化天氣資料以便顯示"""
        if not data.get("success"):
            return f"錯誤: {data.get('error', '未知錯誤')}"
        
        records = data.get("data", {})
        locations = records.get("Locations", [])
        
        if not locations:
            return "未找到天氣資料"
        
        result = []
        for location_group in locations:
            for location in location_group.get("Location", []):
                location_name = location.get("LocationName", "未知地點")
                result.append(f"\n📍 {location_name} 天氣預報")
                result.append("=" * 30)
                
                for element in location.get("WeatherElement", []):
                    element_name = element.get("ElementName", "未知要素")
                    result.append(f"\n🌤️ {element_name}:")
                    
                    for time_info in element.get("Time", []):
                        start_time = time_info.get("StartTime", "")
                        end_time = time_info.get("EndTime", "")
                        
                        # 格式化時間顯示
                        if start_time and end_time:
                            start_display = start_time.split("T")[0] + " " + start_time.split("T")[1][:5]
                            end_display = end_time.split("T")[0] + " " + end_time.split("T")[1][:5]
                            result.append(f"  ⏰ {start_display} ~ {end_display}")
                        
                        # 顯示預報資料
                        for value in time_info.get("ElementValue", []):
                            for key, val in value.items():
                                if val and val != "-":
                                    result.append(f"     {key}: {val}")
                        result.append("")
        
        return "\n".join(result)

async def main():
    """主程式:查詢天氣預報"""
    weather_api = WeatherAPI()
    
    print("🌤️ 台灣天氣預報查詢系統")
    print("=" * 40)
    
    # 詢問使用者要查詢的縣市
    print("\n可查詢的縣市:")
    for i, city in enumerate(VALID_LOCATIONS, 1):
        print(f"{i:2d}. {city}", end="  ")
        if i % 4 == 0:  # 每4個換行
            print()
    print("\n")
    
    location = input("請輸入要查詢的縣市名稱 (例: 臺中市): ").strip()
    if not location:
        location = "臺中市"  # 預設值
    
    # 詢問要查詢的天氣要素
    print("\n可查詢的天氣要素:")
    for i, element in enumerate(VALID_ELEMENTS, 1):
        print(f"{i:2d}. {element}", end="  ")
        if i % 3 == 0:  # 每3個換行
            print()
    print("\n")
    
    element_choice = input("請輸入要查詢的天氣要素 (直接按Enter查詢所有要素): ").strip()
    
    # 執行查詢
    print(f"\n🔍 正在查詢 {location} 的天氣預報...")
    
    try:
        result = await weather_api.get_weather_forecast(
            LocationName=location,
            ElementName=element_choice if element_choice else None
        )
        
        # 格式化並顯示結果
        formatted_result = weather_api.format_weather_data(result)
        print(formatted_result)
        
    except Exception as e:
        print(f"❌ 查詢失敗: {str(e)}")

if __name__ == "__main__":
    asyncio.run(main())

啟動之後在終端機上看到這個畫面就代表做對了

╭─ FastMCP 2.0 ──────────────────────────────────────────────────────────────╮
│                                                                            │
│        _ __ ___ ______           __  __  _____________    ____    ____     │
│       _ __ ___ / ____/___ ______/ /_/  |/  / ____/ __ \  |___ \  / __ \    │
│      _ __ ___ / /_  / __ `/ ___/ __/ /|_/ / /   / /_/ /  ___/ / / / / /    │
│     _ __ ___ / __/ / /_/ (__  ) /_/ /  / / /___/ ____/  /  __/_/ /_/ /     │
│    _ __ ___ /_/    \__,_/____/\__/_/  /_/\____/_/      /_____(_)____/      │
│                                                                            │
│                                                                            │
│                                                                            │
│    🖥️  Server name:     FastMCP                                             │
│    📦 Transport:       STDIO                                               │
│                                                                            │
│    📚 Docs:            https://gofastmcp.com                               │
│    🚀 Deploy:          https://fastmcp.cloud                               │
│                                                                            │
│    🏎️  FastMCP version: 2.11.2                                              │
│    🤝 MCP version:     1.12.4                                              │
│                                                                            │
╰────────────────────────────────────────────────────────────────────────────

如果沒有就直接執行weather.py來偵錯

明天來使用CLAUDE DESKTOP來確認FASTMCP是否能正常運行


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

尚未有邦友留言

立即登入留言