224 lines
8.4 KiB
Python
224 lines
8.4 KiB
Python
"""
|
|
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
|
|
|
|
|
|
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),
|
|
)
|
|
logger.info("✅ MCP httpx.AsyncClient initialized")
|
|
|
|
|
|
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
|
|
|
|
|
|
# ─── 공통 리스크 평가 함수 ────────────────────────────────────────────
|
|
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, 매매 중단 방지)
|
|
"""
|
|
if not settings.MCP_VALIDATION_ENABLED:
|
|
return True, "mcp_disabled"
|
|
|
|
try:
|
|
client = get_http_client()
|
|
except RuntimeError:
|
|
logger.warning("⚠️ http_client 미초기화 — fail-open")
|
|
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:
|
|
return False, reason
|
|
return True, reason
|
|
else:
|
|
logger.warning(f"⚠️ MCP RISK_ASSESSMENT HTTP {resp.status_code} — fail-open")
|
|
return True, f"mcp_http_{resp.status_code}_fail_open"
|
|
except httpx.TimeoutException:
|
|
logger.warning("⚠️ MCP RISK_ASSESSMENT 타임아웃 — fail-open")
|
|
return True, "mcp_timeout_fail_open"
|
|
except Exception as e:
|
|
logger.warning(f"⚠️ MCP RISK_ASSESSMENT 오류: {e} — fail-open")
|
|
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)}")
|