315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""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,
|
|
)
|