FastMCP是什麼?
專門用來快速開發 MCP 伺服器的工具
FastMCP 特點:
Swagger3.0/OpenAPI是什麼
用 YAML 或 JSON 描述 API 的規格,AI 可以直接讀這個檔案,就知道有哪些 API、參數格式、回傳資料型態
實作時間
先到中央氣象署的網站申請API金鑰 https://opendata.cwa.gov.tw
在.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):
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是否能正常運行