Upload 3 files

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --upgrade pip && pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8000
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8001"]

main.py

@@ -0,0 +1,337 @@
+import os
+import requests
+import yfinance as yf
+import pandas as pd
+from datetime import datetime, timedelta
+from fastapi import FastAPI, HTTPException
+from fastapi_mcp import FastApiMCP
+from typing import Dict, Any, Optional
+from dotenv import load_dotenv
+
+load_dotenv()
+app = FastAPI(title="Weather & Stock MCP Server")
+
+# OpenWeather API 설정
+OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY", "your_api_key_here")
+OPENWEATHER_BASE_URL = "http://api.openweathermap.org/data/2.5/weather"
+
+@app.get("/weather", operation_id="get_weather")
+def get_weather(city: str, country: str = None, units: str = "metric") -> Dict[str, Any]:
+    """
+    OpenWeather API를 사용해서 지정된 도시의 현재 날씨 정보를 가져옵니다.
+    
+    Args:
+        city: 도시 이름 (예: "Seoul", "Tokyo")
+        country: 국가 코드 (선택사항, 예: "KR", "JP")
+        units: 온도 단위 ("metric"=섭씨, "imperial"=화씨, "kelvin"=켈빈)
+    
+    Returns:
+        날씨 정보 딕셔너리
+    """
+    try:
+        # API 키 확인
+        if OPENWEATHER_API_KEY == "your_api_key_here":
+            raise HTTPException(
+                status_code=400, 
+                detail="OpenWeather API 키가 설정되지 않았습니다. 환경 변수 OPENWEATHER_API_KEY를 설정하세요."
+            )
+        
+        # 도시 이름 구성
+        location = city
+        if country:
+            location = f"{city},{country}"
+        
+        # API 요청 매개변수
+        params = {
+            "q": location,
+            "appid": OPENWEATHER_API_KEY,
+            "units": units,
+            "lang": "kr"  # 한국어 설명
+        }
+        
+        # OpenWeather API 호출
+        response = requests.get(OPENWEATHER_BASE_URL, params=params, timeout=10)
+        
+        if response.status_code == 404:
+            raise HTTPException(status_code=404, detail=f"도시 '{city}'를 찾을 수 없습니다.")
+        elif response.status_code == 401:
+            raise HTTPException(status_code=401, detail="잘못된 API 키입니다.")
+        elif response.status_code != 200:
+            raise HTTPException(status_code=response.status_code, detail="날씨 정보를 가져오는데 실패했습니다.")
+        
+        weather_data = response.json()
+        
+        # 응답 데이터 정리
+        result = {
+            "city": weather_data["name"],
+            "country": weather_data["sys"]["country"],
+            "temperature": weather_data["main"]["temp"],
+            "feels_like": weather_data["main"]["feels_like"],
+            "humidity": weather_data["main"]["humidity"],
+            "pressure": weather_data["main"]["pressure"],
+            "description": weather_data["weather"][0]["description"],
+            "wind_speed": weather_data.get("wind", {}).get("speed", 0),
+            "visibility": weather_data.get("visibility", 0) / 1000,  # km 단위
+            "units": units,
+            "timestamp": weather_data["dt"]
+        }
+        
+        return result
+        
+    except requests.exceptions.RequestException as e:
+        raise HTTPException(status_code=500, detail=f"API 요청 실패: {str(e)}")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"오류 발생: {str(e)}")
+
+@app.get("/weather/forecast", operation_id="get_weather_forecast")
+def get_weather_forecast(city: str, country: str = None, units: str = "metric") -> Dict[str, Any]:
+    """
+    OpenWeather API를 사용해서 5일 날씨 예보를 가져옵니다.
+    """
+    try:
+        if OPENWEATHER_API_KEY == "your_api_key_here":
+            raise HTTPException(
+                status_code=400,
+                detail="OpenWeather API 키가 설정되지 않았습니다."
+            )
+        
+        location = city
+        if country:
+            location = f"{city},{country}"
+        
+        params = {
+            "q": location,
+            "appid": OPENWEATHER_API_KEY,
+            "units": units,
+            "lang": "kr"
+        }
+        
+        forecast_url = "http://api.openweathermap.org/data/2.5/forecast"
+        response = requests.get(forecast_url, params=params, timeout=10)
+        
+        if response.status_code != 200:
+            raise HTTPException(status_code=response.status_code, detail="예보 정보를 가져오는데 실패했습니다.")
+        
+        forecast_data = response.json()
+        
+        # 하루에 하나씩만 선택 (정오 12시 기준)
+        daily_forecasts = []
+        for item in forecast_data["list"][::8]:  # 3시간 간격이므로 8개마다 선택
+            daily_forecasts.append({
+                "date": item["dt_txt"],
+                "temperature": item["main"]["temp"],
+                "description": item["weather"][0]["description"],
+                "humidity": item["main"]["humidity"]
+            })
+        
+        return {
+            "city": forecast_data["city"]["name"],
+            "country": forecast_data["city"]["country"],
+            "forecasts": daily_forecasts[:5],  # 5일치만
+            "units": units
+        }
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"오류 발생: {str(e)}")
+
+@app.get("/stock", operation_id="get_stock_data")
+def get_stock_data(
+    symbol: str, 
+    period: str = "1mo", 
+    interval: str = "1d"
+) -> Dict[str, Any]:
+    """
+    yfinance를 사용해서 주식 데이터를 가져옵니다.
+    
+    Args:
+        symbol: 주식 심볼 (예: "AAPL", "005930.KS" (삼성전자), "TSLA")
+        period: 기간 ("1d", "5d", "1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max")
+        interval: 간격 ("1m", "2m", "5m", "15m", "30m", "60m", "90m", "1h", "1d", "5d", "1wk", "1mo", "3mo")
+    
+    Returns:
+        주식 데이터와 기본 정보
+    """
+    try:
+        # Ticker 객체 생성
+        ticker = yf.Ticker(symbol)
+        
+        # 주식 정보 가져오기
+        info = ticker.info
+        
+        # 히스토리 데이터 가져오기
+        hist = ticker.history(period=period, interval=interval)
+        
+        if hist.empty:
+            raise HTTPException(status_code=404, detail=f"심볼 '{symbol}'의 데이터를 찾을 수 없습니다.")
+        
+        # DataFrame을 딕셔너리로 변환
+        hist_dict = {}
+        for date, row in hist.iterrows():
+            date_str = date.strftime('%Y-%m-%d %H:%M:%S') if hasattr(date, 'strftime') else str(date)
+            hist_dict[date_str] = {
+                "open": float(row['Open']) if pd.notna(row['Open']) else None,
+                "high": float(row['High']) if pd.notna(row['High']) else None,
+                "low": float(row['Low']) if pd.notna(row['Low']) else None,
+                "close": float(row['Close']) if pd.notna(row['Close']) else None,
+                "volume": int(row['Volume']) if pd.notna(row['Volume']) else None
+            }
+        
+        # 최신 가격 정보
+        latest_data = hist.iloc[-1]
+        current_price = float(latest_data['Close'])
+        
+        # 가격 변화 계산
+        if len(hist) > 1:
+            prev_close = float(hist.iloc[-2]['Close'])
+            price_change = current_price - prev_close
+            price_change_percent = (price_change / prev_close) * 100
+        else:
+            price_change = 0
+            price_change_percent = 0
+        
+        # 기본 정보 추출
+        company_info = {
+            "symbol": symbol,
+            "company_name": info.get("longName", info.get("shortName", symbol)),
+            "sector": info.get("sector", "N/A"),
+            "industry": info.get("industry", "N/A"),
+            "market_cap": info.get("marketCap", 0),
+            "currency": info.get("currency", "USD")
+        }
+        
+        result = {
+            "company_info": company_info,
+            "current_price": current_price,
+            "price_change": price_change,
+            "price_change_percent": round(price_change_percent, 2),
+            "period": period,
+            "interval": interval,
+            "data_points": len(hist),
+            "historical_data": hist_dict
+        }
+        
+        return result
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"주식 데이터 조회 실패: {str(e)}")
+
+@app.get("/stock/info", operation_id="get_stock_info")
+def get_stock_info(symbol: str) -> Dict[str, Any]:
+    """
+    특정 주식의 상세 정보를 가져옵니다.
+    """
+    try:
+        ticker = yf.Ticker(symbol)
+        info = ticker.info
+        
+        if not info or "symbol" not in info:
+            raise HTTPException(status_code=404, detail=f"심볼 '{symbol}'의 정보를 찾을 수 없습니다.")
+        
+        # 주요 정보 추출
+        stock_info = {
+            "symbol": symbol,
+            "company_name": info.get("longName", info.get("shortName", symbol)),
+            "sector": info.get("sector", "N/A"),
+            "industry": info.get("industry", "N/A"),
+            "market_cap": info.get("marketCap", 0),
+            "enterprise_value": info.get("enterpriseValue", 0),
+            "trailing_pe": info.get("trailingPE", 0),
+            "forward_pe": info.get("forwardPE", 0),
+            "peg_ratio": info.get("pegRatio", 0),
+            "price_to_book": info.get("priceToBook", 0),
+            "debt_to_equity": info.get("debtToEquity", 0),
+            "return_on_equity": info.get("returnOnEquity", 0),
+            "revenue_growth": info.get("revenueGrowth", 0),
+            "earnings_growth": info.get("earningsGrowth", 0),
+            "current_price": info.get("currentPrice", info.get("regularMarketPrice", 0)),
+            "target_high_price": info.get("targetHighPrice", 0),
+            "target_low_price": info.get("targetLowPrice", 0),
+            "target_mean_price": info.get("targetMeanPrice", 0),
+            "recommendation": info.get("recommendationKey", "N/A"),
+            "52_week_high": info.get("fiftyTwoWeekHigh", 0),
+            "52_week_low": info.get("fiftyTwoWeekLow", 0),
+            "dividend_yield": info.get("dividendYield", 0),
+            "ex_dividend_date": info.get("exDividendDate", None),
+            "currency": info.get("currency", "USD"),
+            "exchange": info.get("exchange", "N/A"),
+            "website": info.get("website", "N/A"),
+            "business_summary": info.get("longBusinessSummary", "N/A")
+        }
+        
+        return stock_info
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"주식 정보 조회 실패: {str(e)}")
+
+@app.get("/stock/multiple", operation_id="get_multiple_stocks")
+def get_multiple_stocks(symbols: str, period: str = "1mo") -> Dict[str, Any]:
+    """
+    여러 주식의 데이터를 한 번에 가져옵니다.
+    
+    Args:
+        symbols: 쉼표로 구분된 주식 심볼들 (예: "AAPL,GOOGL,MSFT")
+        period: 기간
+    """
+    try:
+        symbol_list = [s.strip() for s in symbols.split(",")]
+        
+        if len(symbol_list) > 10:
+            raise HTTPException(status_code=400, detail="한 번에 최대 10개의 종목만 조회할 수 있습니다.")
+        
+        results = {}
+        
+        for symbol in symbol_list:
+            try:
+                ticker = yf.Ticker(symbol)
+                hist = ticker.history(period=period)
+                info = ticker.info
+                
+                if not hist.empty:
+                    latest_data = hist.iloc[-1]
+                    current_price = float(latest_data['Close'])
+                    
+                    # 가격 변화 계산
+                    if len(hist) > 1:
+                        prev_close = float(hist.iloc[-2]['Close'])
+                        price_change = current_price - prev_close
+                        price_change_percent = (price_change / prev_close) * 100
+                    else:
+                        price_change = 0
+                        price_change_percent = 0
+                    
+                    results[symbol] = {
+                        "company_name": info.get("longName", info.get("shortName", symbol)),
+                        "current_price": current_price,
+                        "price_change": price_change,
+                        "price_change_percent": round(price_change_percent, 2),
+                        "volume": int(latest_data['Volume']) if pd.notna(latest_data['Volume']) else 0,
+                        "market_cap": info.get("marketCap", 0),
+                        "currency": info.get("currency", "USD")
+                    }
+                else:
+                    results[symbol] = {"error": "데이터를 찾을 수 없습니다."}
+                    
+            except Exception as e:
+                results[symbol] = {"error": str(e)}
+        
+        return {
+            "period": period,
+            "stocks": results,
+            "requested_symbols": symbol_list,
+            "successful_count": len([r for r in results.values() if "error" not in r])
+        }
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"다중 주식 조회 실패: {str(e)}")
+
+mcp = FastApiMCP(
+    app,
+    name="Weather & Stock API MCP"
+  
+)
+
+# /mcp 경로에 MCP 서버를 마운트합니다.
+mcp.mount_http(mount_path="/mcp") 
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8001)

requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+fastapi-mcp
+uvicorn
+requests
+python-dotenv
+yfinance
+pandas