auto-stock/mcp-server/main.py

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Dict, Any, List
import logging

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 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


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,
        "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")
            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"
                )
            
            result = await ai_client.analyze_market(stock_code, market_data, news)
            return {
                "analysis_type": analysis_type,
                "result": result
            }
        
        elif analysis_type == "RISK_ASSESSMENT":
            # 리스크 평가
            portfolio = data.get("portfolio")
            trade_plan = data.get("trade_plan")
            
            if not portfolio or not trade_plan:
                raise HTTPException(
                    status_code=400,
                    detail="portfolio and trade_plan are required"
                )
            
            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
            }
        
        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
    uvicorn.run(
        "main:app",
        host=settings.MCP_SERVER_HOST,
        port=settings.MCP_SERVER_PORT,
        reload=True
    )