"""MCP 서버 메인 모듈 (main.py) Model Context Protocol(MCP) 서버 — AI 기반 주식 매매를 위한 REST API 제공. 백엔드(FastAPI)에서 이 서버를 호출하여 AI 분석 및 도구 실행을 요청한다. 엔드포인트: GET / — 서버 상태 확인 GET /health — 헬스 체크 (AI 프로바이더 설정 확인) GET /tools — 사용 가능한 MCP 도구 목록 POST /tools/call — MCP 도구 호출 (시장 데이터, 지표 계산, 뉴스 조회 등) POST /analyze — AI 분석 실행 (규칙 검증, 시장 분석, 리스크 평가 등) 보안: - MCP_API_KEY가 설정된 경우 X-API-Key 헤더 검증 (constant-time 비교) - CORS: MCP_CORS_ORIGINS로 origin 제한 AI 프로바이더: - AI_PROVIDER=gemini 이면 GeminiClient, 아니면 ClaudeClient 사용 """ from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel from typing import Dict, Any, List import hmac import logging import os from config import settings from claude_client import claude_client from mcp_tools import mcp_tools # AI_PROVIDER에 따라 클라이언트 동적 선택 if settings.AI_PROVIDER == "gemini": from gemini_client import gemini_client ai_client = gemini_client else: ai_client = claude_client # 로깅 설정 logging.basicConfig( level=getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) app = FastAPI( title="Auto Stock Trading MCP Server", description="Model Context Protocol Server for Stock Trading with Claude AI", version="1.0.0" ) # CORS 설정 (이슈 5) # 문제: allow_origins=["*"] + allow_credentials=True 조합은 CORS 스펙 위반 (브라우저 차단) # 해결: 환경변수 MCP_CORS_ORIGINS로 origin 제한 + allow_credentials=False (서버-서버 통신 불필요) _raw_cors = settings.MCP_CORS_ORIGINS _allowed_origins = ( [o.strip() for o in _raw_cors.split(",") if o.strip()] if _raw_cors else ["http://localhost:8000", "http://127.0.0.1:8000", "http://backend:8000"] ) logger.info(f"MCP CORS allowed origins: {_allowed_origins}") app.add_middleware( CORSMiddleware, allow_origins=_allowed_origins, allow_credentials=False, # 서버-서버 통신; 브라우저 세션 쿠키 불필요 allow_methods=["GET", "POST"], allow_headers=["Content-Type", "Authorization", "X-API-Key"], ) # ─── API Key 인증 미들웨어 ────────────────────────────────────────────── # MCP_API_KEY가 설정되어 있으면 모든 요청에 X-API-Key 헤더 검증 (health/root 제외) _MCP_API_KEY = settings.MCP_API_KEY _PUBLIC_PATHS = {"/", "/health"} @app.middleware("http") async def verify_api_key(request: Request, call_next): """X-API-Key 헤더를 constant-time 비교로 검증. MCP_API_KEY가 빈 문자열이면 인증을 비활성화 (개발 환경 호환). """ if _MCP_API_KEY and request.url.path not in _PUBLIC_PATHS: api_key = request.headers.get("X-API-Key", "") if not api_key or not hmac.compare_digest(api_key.encode(), _MCP_API_KEY.encode()): logger.warning(f"[MCP Security] API 키 검증 실패 — path={request.url.path}") return JSONResponse( status_code=401, content={"detail": "유효한 X-API-Key 헤더가 필요합니다."}, ) return await call_next(request) class ToolCallRequest(BaseModel): tool_name: str parameters: Dict[str, Any] class AnalysisRequest(BaseModel): analysis_type: str # RULE_VALIDATION, MARKET_ANALYSIS, RISK_ASSESSMENT data: Dict[str, Any] @app.get("/") async def root(): return { "message": "MCP Server for Auto Stock Trading", "version": "1.0.0", "status": "running" } @app.get("/health") async def health_check(): return { "status": "healthy", "ai_provider": settings.AI_PROVIDER, "hybrid_ai_enabled": settings.HYBRID_AI_ENABLED if settings.AI_PROVIDER == "claude" else False, "claude_model": settings.CLAUDE_MODEL, "claude_haiku_model": settings.CLAUDE_HAIKU_MODEL, "claude_configured": bool(settings.ANTHROPIC_API_KEY), "gemini_configured": bool(settings.GEMINI_API_KEY), } @app.get("/tools") async def list_tools(): """사용 가능한 도구 목록""" return { "tools": mcp_tools.get_tool_definitions() } @app.post("/tools/call") async def call_tool(request: ToolCallRequest): """MCP 도구 호출""" try: result = await mcp_tools.call_tool(request.tool_name, request.parameters) return result except Exception as e: logger.error(f"Tool call error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/analyze") async def analyze(request: AnalysisRequest): """Claude를 통한 분석""" try: analysis_type = request.analysis_type data = request.data if analysis_type == "RULE_VALIDATION": rule = data.get("rule") market_data = data.get("market_data") if not rule or not market_data: raise HTTPException( status_code=400, detail="rule and market_data are required" ) stock_code = market_data.get("stock_code") # TOKEN_SAVE_MODE: OHLCV 배열 제거, 핵심 지표만 유지 if settings.TOKEN_SAVE_MODE: _keep = {"stock_code", "current_price", "change_rate", "volume", "rsi", "atr", "adx", "bollinger_upper", "bollinger_lower", "bollinger_middle", "ma5", "ma20", "ma60", "signals", "score"} market_data = {k: v for k, v in market_data.items() if k in _keep} news_data = [] if stock_code: try: news_result = await mcp_tools.get_news(stock_code, limit=settings.NEWS_LIMIT) if "news" in news_result: news_data = news_result["news"] except Exception as e: logger.warning(f"Failed to fetch news for validation: {e}") result = await ai_client.validate_trading_rule(rule, market_data, news_data) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "MARKET_ANALYSIS": stock_code = data.get("stock_code") market_data = data.get("market_data") news = data.get("news", []) if not stock_code or not market_data: raise HTTPException( status_code=400, detail="stock_code and market_data are required" ) # TOKEN_SAVE_MODE: OHLCV 배열 제거 if settings.TOKEN_SAVE_MODE and isinstance(market_data, dict): market_data = {k: v for k, v in market_data.items() if k != "data"} # 뉴스 제목만 유지 if settings.TOKEN_SAVE_MODE and news: news = [{"title": n.get("title", "")} for n in news[:settings.NEWS_LIMIT]] result = await ai_client.analyze_market(stock_code, market_data, news) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "RISK_ASSESSMENT": # 이슈 4: request_mcp_risk_assessment()가 trade_plan/portfolio/signals 전달 portfolio = data.get("portfolio") or {} trade_plan = data.get("trade_plan") signals = data.get("signals") or {} if not trade_plan: raise HTTPException( status_code=400, detail="trade_plan is required" ) # signals를 trade_plan에 병합하여 AI에 전달 if signals: trade_plan["signals"] = signals result = await ai_client.assess_risk(portfolio, trade_plan) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "FAIR_PRICE": stock_code = data.get("stock_code") market_data = data.get("market_data") financial_data = data.get("financial_data") if not stock_code or not market_data: raise HTTPException( status_code=400, detail="stock_code and market_data are required" ) result = await ai_client.calculate_fair_price( stock_code, market_data, financial_data ) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "RULE_OPTIMIZATION": stock_code = data.get("stock_code") market_data = data.get("market_data") if not stock_code or not market_data: raise HTTPException( status_code=400, detail="stock_code and market_data are required" ) news_data = [] try: news_result = await mcp_tools.get_news(stock_code, limit=settings.NEWS_LIMIT) if "news" in news_result: news_data = news_result["news"] except Exception as e: logger.warning(f"Failed to fetch news for optimization: {e}") result = await ai_client.optimize_trading_rules(stock_code, market_data, news_data) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "GENERATE_JOURNAL": trade_info = data.get("trade_info") market_data = data.get("market_data") if not trade_info or not market_data: raise HTTPException( status_code=400, detail="trade_info and market_data are required" ) result = await ai_client.generate_trade_journal(trade_info, market_data) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "BATCH_STOCK_ANALYSIS": stocks = data.get("stocks") if not stocks or not isinstance(stocks, list): raise HTTPException( status_code=400, detail="stocks (list) is required" ) result = await ai_client.analyze_batch_stocks(stocks) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "NEWS_SENTIMENT": stock_code = data.get("stock_code") stock_name = data.get("stock_name") news_titles = data.get("news_titles") if not stock_code or not stock_name or not news_titles: raise HTTPException( status_code=400, detail="stock_code, stock_name, and news_titles are required" ) result = await ai_client.check_news_sentiment(stock_code, stock_name, news_titles) return {"analysis_type": analysis_type, "result": result} elif analysis_type == "DAILY_REVIEW": trades_summary = data.get("trades_summary") market_summary = data.get("market_summary") if not trades_summary or not market_summary: raise HTTPException( status_code=400, detail="trades_summary and market_summary are required" ) result = await ai_client.review_daily_trades(trades_summary, market_summary) return {"analysis_type": analysis_type, "result": result} else: raise HTTPException( status_code=400, detail=f"Unknown analysis type: {analysis_type}" ) except HTTPException: raise except Exception as e: logger.error(f"Analysis error: {e}") raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": import uvicorn # [수정 5] reload=True는 운영 환경에서 성능 저하 및 다중 프로세스 부작용을 유발한다. # 개발 환경에서는 MCP_DEV_RELOAD=true 환경변수로 활성화할 수 있다. _dev_reload = os.getenv("MCP_DEV_RELOAD", "false").lower() == "true" uvicorn.run( "main:app", host=settings.MCP_SERVER_HOST, port=settings.MCP_SERVER_PORT, reload=_dev_reload, )