今天讓我們來把他分解成函式，restful伺服器並且把它用fastmcp包裝成mcp server吧，這樣之後才能方便使用

實作時間

先把之前的程式分成兩個部分，首先是天氣的部分

首先開一個新的檔案叫做weather_api.py，再來把之前的天氣的部分切出來

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())

接著是rest api server，這東西可以讓我們在不使用mcp server的情況下查看有哪些功能及他的網址，讓我們之後給ai及自己讀蠻有幫助的，首先先安裝需要用的套件

pip install fastapi uvicorn pydantic pyyaml

再來把之前測試用的fastmcp的東西改個名放進去

from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
from weather_api import WeatherAPI
import asyncio
import os
import google.generativeai as genai
from dotenv import load_dotenv

# 載入環境變數
load_dotenv()

app = FastAPI(
    title="My MCP APIs",
    version="1.0.0",
    description="提供天氣查詢的 API"
)

# 初始化 API 實例
weather_api = WeatherAPI()

#要輸入的
class WeatherRequest(BaseModel):
    LocationName: Optional[List[str]] = None
    ElementName: Optional[List[str]] = None
    limit: Optional[int] = None
    offset: Optional[int] = None

#要輸出的
class WeatherResponse(BaseModel):
    city: str
    temperature: str
    description: str

#網址後面+/weather會導向哪個程式
@app.post("/weather", response_model=dict)
async def get_weather(
    request: WeatherRequest,
    authorization: Optional[str] = Header(None, description="CWB API Key")
):
    """查詢天氣"""
    try:
        print(f"收到天氣查詢請求: {request}")
        
        # 使用 weather_api 實例來處理請求
        result = await weather_api.get_weather_forecast(
            LocationName=",".join(request.LocationName) if request.LocationName else None,
            ElementName=",".join(request.ElementName) if request.ElementName else None,
            limit=request.limit,
            offset=request.offset
        )
        
        return result
        
    except Exception as e:
        print(f"天氣查詢錯誤: {str(e)}")
        raise HTTPException(status_code=400, detail=str(e))
#根會長怎樣
@app.get("/")
async def root():
    """根路徑"""
    return {
        "message": "My MCP APIs",
        "version": "1.0.0",
        "endpoints": ["/weather", "/gemini/art-description"],
        "docs": "/docs"
    }

if __name__ == "__main__":
    uvicorn.run(
        app, 
        host="0.0.0.0", 
        port=8001,
        log_level="info"
    )

這樣rest api會出現在你的localhost:8001/docs中
再來是fastmcp的部分

import httpx
import yaml
from fastmcp import FastMCP

def main():
    # 載入 OpenAPI 規範
    with open("openapi.yaml", "r", encoding="utf-8") as f:
        openapi_spec = yaml.safe_load(f)
    
    # 更新 OpenAPI 規範中的 server URL，指向我們的 REST API
    openapi_spec["servers"] = [
        {"url": "http://localhost:8001", "description": "REST API 伺服器"}
    ]
    
    print("📋 載入的 OpenAPI 規範:")
    print(f"   標題: {openapi_spec['info']['title']}")
    print(f"   版本: {openapi_spec['info']['version']}")
    print(f"   伺服器: {openapi_spec['servers'][0]['url']}")
    print(f"   路徑: {list(openapi_spec['paths'].keys())}")
    
    # 建立 HTTP client，指向 REST API 伺服器
    client = httpx.AsyncClient(
        base_url="http://localhost:8001",
        timeout=30.0
    )
    
    # 建立 MCP 伺服器
    mcp = FastMCP.from_openapi(
        openapi_spec=openapi_spec,
        client=client,
        name="Weather Gemini MCP Server"
    )
    
    # 啟動 MCP 伺服器
    mcp.run(transport="http", port=8000)

if __name__ == "__main__":
    main()

fastmcp的位置是在localhost:8000/mcp
明天開始要用n8n相關的東西了，但筆者發現n8n官方平台只有14天的免費試用期，所以會找更便宜的方法