GET /apod
:回傳每日天文圖(Astronomy Picture of the Day)GET /mars-photo
:回傳火星照片GET /neo
:回傳近地天體(Near-Earth Objects)資訊此為簡單的 FastAPI 專案,正式的作業環境應該把 route 分離到不同的檔案中。此專案結構僅作demo用:
nasa-app-mcp-demo/
├── .env
├── README.md
├── main.py
├── test_basic.py
├── venv/
│ └── ... (virtual environment files)
fastapi>=0.104.0
uvicorn[standard]>=0.24.0
httpx>=0.25.0
python-dotenv>=1.0.0
fastapi-mcp>=0.4.0
# 建資料夾
mkdir -p nasa-fastapi-mcp-demo/app && cd nasa-fastapi-mcp-demo
# 建立虛擬環境
python3.12 -m venv .venv && source .venv/bin/activate
# 建 requirements.txt(貼上上面內容)
# 然後安裝
pip install -r requirements.txt
建立 .env.example
:
NASA_API_KEY=your_nasa_api_key_here
複製一份:
cp .env.example .env
# 然後把 your_nasa_api_key_here 改成你自己的 Key
main.py
import os
import httpx
from typing import Optional, List, Dict, Any
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
app = FastAPI(
title="Simple NASA MCP Server",
description="A simple FastAPI server providing NASA data through REST endpoints and MCP",
version="1.0.0"
)
# NASA API configuration
NASA_API_KEY = os.getenv("NASA_API_KEY", "DEMO_KEY")
NASA_BASE_URL = "https://api.nasa.gov"
# HTTP client for NASA API calls
client = httpx.AsyncClient(timeout=30.0)
# Response models
class APODResponse(BaseModel):
title: str
explanation: str
url: str
date: str
media_type: str
hdurl: Optional[str] = None
copyright: Optional[str] = None
class MarsPhotoResponse(BaseModel):
id: int
sol: int
camera: Dict[str, Any]
img_src: str
earth_date: str
rover: Dict[str, Any]
class MarsPhotosResponse(BaseModel):
photos: List[MarsPhotoResponse]
class NEOResponse(BaseModel):
id: str
name: str
estimated_diameter: Dict[str, Any]
is_potentially_hazardous_asteroid: bool
close_approach_data: List[Dict[str, Any]]
class NEOFeedResponse(BaseModel):
element_count: int
near_earth_objects: Dict[str, List[NEOResponse]]
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {"status": "healthy", "service": "simple-nasa-mcp"}
@app.get("/apod", response_model=APODResponse)
async def get_astronomy_picture_of_day(date: Optional[str] = Query(None, description="Date in YYYY-MM-DD format")):
"""Get NASA's Astronomy Picture of the Day"""
try:
params = {"api_key": NASA_API_KEY}
if date:
params["date"] = date
response = await client.get(f"{NASA_BASE_URL}/planetary/apod", params=params)
if response.status_code == 200:
data = response.json()
return APODResponse(**data)
else:
raise HTTPException(
status_code=response.status_code,
detail=f"NASA API error: {response.text}"
)
except httpx.RequestError as e:
raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@app.get("/mars-photos", response_model=MarsPhotosResponse)
async def get_mars_rover_photos(
rover: str = Query(..., description="Rover name (curiosity, perseverance, opportunity, spirit)"),
sol: int = Query(..., description="Martian sol (day) number"),
camera: Optional[str] = Query(None, description="Camera name (FHAZ, RHAZ, MAST, NAVCAM, etc.)")
):
"""Get Mars rover photos"""
try:
params = {
"api_key": NASA_API_KEY,
"sol": sol
}
if camera:
params["camera"] = camera
response = await client.get(f"{NASA_BASE_URL}/mars-photos/api/v1/rovers/{rover}/photos", params=params)
if response.status_code == 200:
data = response.json()
return MarsPhotosResponse(**data)
else:
raise HTTPException(
status_code=response.status_code,
detail=f"NASA API error: {response.text}"
)
except httpx.RequestError as e:
raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@app.get("/neo", response_model=NEOFeedResponse)
async def get_near_earth_objects(
start_date: str = Query(..., description="Start date in YYYY-MM-DD format"),
end_date: str = Query(..., description="End date in YYYY-MM-DD format")
):
"""Get Near Earth Objects (asteroids and comets)"""
try:
params = {
"api_key": NASA_API_KEY,
"start_date": start_date,
"end_date": end_date
}
response = await client.get(f"{NASA_BASE_URL}/neo/rest/v1/feed", params=params)
if response.status_code == 200:
data = response.json()
return NEOFeedResponse(**data)
else:
raise HTTPException(
status_code=response.status_code,
detail=f"NASA API error: {response.text}"
)
except httpx.RequestError as e:
raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@app.on_event("shutdown")
async def shutdown_event():
"""Clean up HTTP client on shutdown"""
await client.aclose()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
uvicorn main:app --reload
--reload
參數會在代碼變更時自動重啟伺服器python main.py
這會啟動 FastAPI 伺服器,並在 http://127.0.0.1:8000
提供 API。
http://127.0.0.1:8000/docs
# 健康檢查
curl "http://127.0.0.1:8000/health"
# 今日 APOD
curl "http://127.0.0.1:8000/apod"
# 指定日期 APOD
curl "http://127.0.0.1:8000/apod?date=2024-07-20&hd=true"
# 搜尋圖片
curl "http://127.0.0.1:8000/mars-photos?rover=curiosity&sol=1000&camera=FHAZ"
# 近地天體
curl "http://127.0.0.1:8000/neo?start_date=2024-07-01&end_date=2024-07-31"
.env
YYYY-MM-DD
下一篇我們會用 fastapi-mcp 直接將這些endpoint轉為MCP可發現的tools,讓 VS Code 可以直接 discover 與呼叫。