auto-stock/mcp-server/claude_code_client.py

"""Claude Code CLI 기반 AI 클라이언트 (테스트/개발용).

Anthropic API 크레딧 대신 Claude Code 구독(`claude -p` 헤드리스 모드)으로 분석을 실행한다.
ClaudeClient와 동일한 인터페이스 — `analyze()`만 서브프로세스 호출로 교체하고
나머지 도메인 메서드(assess_risk, analyze_market 등)는 그대로 상속.

사용:
  AI_PROVIDER=claude_code
  CLAUDE_CODE_OAUTH_TOKEN=<`claude setup-token` 출력>   # 컨테이너에서 실행 시 필수

제약:
  - API tool_use(함수 호출) 미지원 → tools 인자는 무시하고 텍스트(JSON) 응답만 사용.
  - 프롬프트 캐시 usage 는 CLI가 보고하는 값을 그대로 전달.
"""

import asyncio
import json
import logging
import os
import shutil
import tempfile
from typing import Any, Dict, List

from config import settings
from claude_client import ClaudeClient, _CACHED_CONTEXT

logger = logging.getLogger(__name__)

_CLI_TIMEOUT = float(os.getenv("CLAUDE_CODE_TIMEOUT", "120"))
_CLI_BIN = os.getenv("CLAUDE_CODE_BIN", "claude")
_MAX_CONCURRENT = int(os.getenv("CLAUDE_CODE_MAX_CONCURRENT", "2"))
# 프로젝트 .claude/ 설정(훅·CLAUDE.md)이 끼어들지 않도록 중립 디렉터리에서 실행
_CLI_CWD = os.getenv("CLAUDE_CODE_CWD") or os.path.join(tempfile.gettempdir(), "autostock-claude-code")


class ClaudeCodeClient(ClaudeClient):
    """`claude -p` 서브프로세스로 ClaudeClient.analyze() 를 대체."""

    def __init__(self):
        # 부모 __init__ 은 AsyncAnthropic(api_key) 를 만들지만 여기서는 SDK 미사용
        self.client = None
        self.model = settings.CLAUDE_MODEL
        self.haiku_model = settings.CLAUDE_HAIKU_MODEL
        self._sem = asyncio.Semaphore(_MAX_CONCURRENT)
        os.makedirs(_CLI_CWD, exist_ok=True)
        if shutil.which(_CLI_BIN) is None:
            logger.error("[ClaudeCode] '%s' 실행 파일을 찾을 수 없음 — 분석 호출은 모두 실패합니다", _CLI_BIN)
        if not os.getenv("CLAUDE_CODE_OAUTH_TOKEN") and not os.getenv("ANTHROPIC_API_KEY"):
            logger.warning("[ClaudeCode] CLAUDE_CODE_OAUTH_TOKEN 미설정 — 호스트 로그인 세션이 없으면 인증 실패")

    async def analyze(
        self,
        system_prompt: str,
        user_message: str,
        tools: List[Dict[str, Any]] = None,
        max_tokens: int = None,
        model_override: str = None,
    ) -> Dict[str, Any]:
        use_model = model_override or self.model
        if tools:
            logger.debug("[ClaudeCode] tools %d개 무시 (CLI 모드는 tool_use 미지원)", len(tools))

        full_system = f"{_CACHED_CONTEXT}\n\n{system_prompt}"
        cmd = [
            _CLI_BIN, "-p", user_message,
            "--output-format", "json",
            "--model", use_model,
            "--system-prompt", full_system,
            "--tools", "",          # 파일/셸 도구 비활성 — 순수 텍스트 분석
            "--max-turns", "1",
        ]

        env = os.environ.copy()
        env.pop("ANTHROPIC_API_KEY", None)  # 크레딧 없는 API 키가 구독 인증을 가리지 않도록

        try:
            async with self._sem:
                proc = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                    env=env,
                    cwd=_CLI_CWD,
                )
                try:
                    out, err = await asyncio.wait_for(proc.communicate(), timeout=_CLI_TIMEOUT)
                except asyncio.TimeoutError:
                    proc.kill()
                    raise RuntimeError(f"claude CLI timeout ({_CLI_TIMEOUT}s)")

            if proc.returncode != 0:
                raise RuntimeError(
                    f"claude CLI exit={proc.returncode}: {err.decode(errors='ignore')[-500:]}"
                )

            data = json.loads(out.decode())
            if data.get("is_error"):
                raise RuntimeError(f"claude CLI error: {data.get('result')}")

            usage = data.get("usage") or {}
            result = {
                "content": [{"type": "text", "text": data.get("result", "")}],
                "stop_reason": data.get("stop_reason", "end_turn"),
                "model_used": use_model,
                "usage": {
                    "input_tokens": usage.get("input_tokens", 0),
                    "output_tokens": usage.get("output_tokens", 0),
                    "cache_read_tokens": usage.get("cache_read_input_tokens", 0),
                    "cache_write_tokens": usage.get("cache_creation_input_tokens", 0),
                },
            }
            logger.debug(
                "[ClaudeCode] model=%s in=%s out=%s cost=$%s",
                use_model, usage.get("input_tokens"), usage.get("output_tokens"),
                data.get("total_cost_usd"),
            )
            return result

        except Exception as e:
            logger.error(f"Claude Code analysis error: {e}")
            raise


claude_code_client = ClaudeCodeClient()