""" MCP 프록시 엔드포인트 + 공통 리스크 평가 유틸리티 이슈 4 - request_mcp_risk_assessment(): auto/day 스케줄러 양쪽에서 재사용 가능한 공통 RISK_ASSESSMENT 함수. 호출부는 trade_plan / portfolio / signals 세 dict를 넘긴다. 이슈 6 - httpx.AsyncClient 싱글톤: 요청마다 클라이언트를 생성/소멸하지 않는다. lifespan에서 init_http_client() → 사용 → close_http_client() 순서로 호출. """ from fastapi import APIRouter, HTTPException from typing import Dict, Any, Optional, Tuple from pydantic import BaseModel import httpx import logging from app.core.config import settings logger = logging.getLogger(__name__) router = APIRouter() # ─── httpx 싱글톤 클라이언트 ────────────────────────────────────────── _http_client: Optional[httpx.AsyncClient] = None def _mcp_default_headers() -> dict: """MCP 서버 인증용 기본 헤더. MCP_API_KEY가 설정되어 있으면 X-API-Key 포함.""" headers: dict = {} if settings.MCP_API_KEY: headers["X-API-Key"] = settings.MCP_API_KEY return headers async def init_http_client() -> None: """lifespan 시작 시 한 번만 호출. keepalive 연결 풀 설정.""" global _http_client _http_client = httpx.AsyncClient( timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0), limits=httpx.Limits(max_keepalive_connections=5, max_connections=10), headers=_mcp_default_headers(), ) logger.info("MCP httpx.AsyncClient initialized (api_key=%s)", "set" if settings.MCP_API_KEY else "none") async def close_http_client() -> None: """lifespan 종료 시 한 번만 호출.""" global _http_client if _http_client is not None: await _http_client.aclose() _http_client = None logger.info("🔒 MCP httpx.AsyncClient closed") def get_http_client() -> httpx.AsyncClient: """공유 클라이언트 반환. 초기화 전 접근 시 RuntimeError.""" if _http_client is None: raise RuntimeError( "http_client not initialized — call init_http_client() in lifespan" ) return _http_client # ─── fail-open 카운터 ───────────────────────────────────────────────── _fail_open_count: int = 0 _FAIL_OPEN_WARN_THRESHOLD: int = 5 # 연속 N회 초과 시 경고 로그 # ─── 공통 리스크 평가 함수 ──────────────────────────────────────────── async def request_mcp_risk_assessment( trade_plan: Dict[str, Any], portfolio: Dict[str, Any], signals: Dict[str, Any], ) -> Tuple[bool, str]: """MCP RISK_ASSESSMENT 호출 — auto/day 스케줄러 공통 진입점. Args: trade_plan: 주문 계획 (action, strategy, stock_code, quantity, price 등) portfolio: 현재 포트폴리오 요약 (보유 종목 목록, 총 평가금액 등) signals: 진입 신호 정보 (strategy, rules, indicators 등) Returns: (approve: bool, reason: str) - approve=True → 주문 실행 - approve=False → 주문 스킵 - MCP 장애 / 타임아웃 → fail-open (approve=True, 매매 중단 방지) """ global _fail_open_count if not settings.MCP_VALIDATION_ENABLED: return True, "mcp_disabled" try: client = get_http_client() except RuntimeError: _fail_open_count += 1 logger.warning( "MCP http_client 미초기화 — fail-open (누적 %d회)", _fail_open_count ) if _fail_open_count > _FAIL_OPEN_WARN_THRESHOLD: logger.error( "MCP fail-open 연속 %d회 초과: MCP 서버 연결 상태를 점검하세요", _fail_open_count, ) return True, "client_not_initialized_fail_open" try: resp = await client.post( f"{settings.MCP_SERVER_URL}/analyze", json={ "analysis_type": "RISK_ASSESSMENT", "data": { "trade_plan": trade_plan, "portfolio": portfolio, "signals": signals, }, }, timeout=15.0, ) if resp.status_code == 200: result = resp.json() # 응답: {"result": {"decision": "approve"|"deny", "reason": "..."}} payload = result.get("result", result) decision = str(payload.get("decision", "approve")).lower() reason = str(payload.get("reason", "")) if "deny" in decision or "reject" in decision: _fail_open_count = 0 # 정상 응답 수신 시 카운터 리셋 return False, reason _fail_open_count = 0 # 정상 응답 수신 시 카운터 리셋 return True, reason else: _fail_open_count += 1 logger.warning( "MCP RISK_ASSESSMENT HTTP %d — fail-open (누적 %d회)", resp.status_code, _fail_open_count, ) if _fail_open_count > _FAIL_OPEN_WARN_THRESHOLD: logger.error( "MCP fail-open 연속 %d회 초과: MCP 서버 응답 오류가 지속됩니다", _fail_open_count, ) return True, f"mcp_http_{resp.status_code}_fail_open" except httpx.TimeoutException: _fail_open_count += 1 logger.warning( "MCP RISK_ASSESSMENT 타임아웃 — fail-open (누적 %d회)", _fail_open_count ) if _fail_open_count > _FAIL_OPEN_WARN_THRESHOLD: logger.error( "MCP fail-open 연속 %d회 초과: MCP 서버 타임아웃이 지속됩니다", _fail_open_count, ) return True, "mcp_timeout_fail_open" except Exception as e: _fail_open_count += 1 logger.warning( "MCP RISK_ASSESSMENT 오류: %s — fail-open (누적 %d회)", e, _fail_open_count ) if _fail_open_count > _FAIL_OPEN_WARN_THRESHOLD: logger.error( "MCP fail-open 연속 %d회 초과: 오류=%s", _fail_open_count, e ) return True, f"mcp_error_fail_open: {e}" # ─── 모델 ────────────────────────────────────────────────────────────── class MCPToolRequest(BaseModel): tool_name: str parameters: Dict[str, Any] class MCPAnalysisRequest(BaseModel): analysis_type: str # "RULE_VALIDATION", "MARKET_ANALYSIS", "RISK_ASSESSMENT" data: Dict[str, Any] # ─── 엔드포인트 ──────────────────────────────────────────────────────── @router.get("/status") async def get_mcp_status(): """MCP 서버 상태 확인""" try: client = get_http_client() except RuntimeError: return {"status": "disconnected", "message": "http_client not initialized"} try: response = await client.get(f"{settings.MCP_SERVER_URL}/health", timeout=2.0) if response.status_code == 200: health_data = response.json() return { "status": "connected", "mcp_server_url": settings.MCP_SERVER_URL, "claude_configured": health_data.get("claude_configured", False), "ai_provider": health_data.get("ai_provider", "unknown"), "details": health_data, } return {"status": "error", "message": "MCP server responded with error"} except Exception as e: return {"status": "disconnected", "message": str(e)} @router.post("/tools/call") async def call_mcp_tool(request: MCPToolRequest): """MCP 도구 호출""" try: client = get_http_client() except RuntimeError as e: raise HTTPException(status_code=503, detail=str(e)) try: response = await client.post( f"{settings.MCP_SERVER_URL}/tools/call", json={"tool_name": request.tool_name, "parameters": request.parameters}, timeout=30.0, ) if response.status_code == 200: return response.json() raise HTTPException( status_code=response.status_code, detail=f"MCP tool call failed: {response.text}", ) except httpx.TimeoutException: raise HTTPException(status_code=504, detail="MCP server timeout") except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to call MCP tool: {str(e)}") @router.post("/analyze") async def analyze_with_claude(request: MCPAnalysisRequest): """Claude AI를 통한 분석 요청""" try: client = get_http_client() except RuntimeError as e: raise HTTPException(status_code=503, detail=str(e)) try: response = await client.post( f"{settings.MCP_SERVER_URL}/analyze", json={"analysis_type": request.analysis_type, "data": request.data}, timeout=60.0, ) if response.status_code == 200: return response.json() raise HTTPException( status_code=response.status_code, detail=f"Claude analysis failed: {response.text}", ) except httpx.TimeoutException: raise HTTPException(status_code=504, detail="Analysis timeout") except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to analyze: {str(e)}") @router.get("/tools") async def list_mcp_tools(): """사용 가능한 MCP 도구 목록""" try: client = get_http_client() except RuntimeError as e: raise HTTPException(status_code=503, detail=str(e)) try: response = await client.get(f"{settings.MCP_SERVER_URL}/tools", timeout=10.0) if response.status_code == 200: return response.json() raise HTTPException( status_code=response.status_code, detail="Failed to fetch tools list" ) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to list tools: {str(e)}")