auto-stock/mcp-server/stdio_server.py

#!/usr/bin/env python3
"""
MCP Server with stdio transport for Claude CLI

This server provides stock trading tools via Model Context Protocol
using stdio transport for direct integration with Claude CLI.
"""

import asyncio
import sys
from typing import Any
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# Import configuration
from config import settings
from utils import _dumps

# Initialize MCP Server
app = Server("auto-stock-mcp")

# Backend API base URL
BACKEND_URL = "http://localhost:8000/api/v1"

# [2026-10-02 H-1] 백엔드 조회 API(portfolio·trades·recommendations·market-data)는 X-API-Key 필수.
# 백엔드 security._get_api_key() 와 같은 소스(API_KEY 우선, 없으면 SECRET_KEY 폴백).
# .env 는 위의 `from config import settings` 가 import 시점에 load_dotenv(override=False) 로
# 이미 읽는다(python-dotenv, mcp-server/requirements.txt) — 그래서 이 블록은 반드시 그 import 뒤에 둔다.
# 단, mcp-server/.env 에는 API_KEY/SECRET_KEY 가 없을 수 있다(2026-10-02 로컬 확인) → 백엔드가 쓰는
# 저장소 루트 .env 를 추가로 읽는다(override=False: 이미 설정된 환경변수가 우선).
import os as _os
from pathlib import Path as _Path
if not (_os.getenv("API_KEY") or _os.getenv("SECRET_KEY")):
    try:
        from dotenv import load_dotenv as _load_dotenv
        _load_dotenv(_Path(__file__).resolve().parents[1] / ".env", override=False)
    except ImportError:
        print("[stdio_server] python-dotenv 없음 — API_KEY 환경변수를 직접 설정하세요", file=sys.stderr)
_BACKEND_API_KEY = _os.getenv("API_KEY") or _os.getenv("SECRET_KEY", "")
if not _BACKEND_API_KEY:
    print("[stdio_server] API_KEY/SECRET_KEY 미설정 — 백엔드 조회가 401 로 실패합니다", file=sys.stderr)
_BACKEND_HEADERS = {"X-API-Key": _BACKEND_API_KEY} if _BACKEND_API_KEY else {}


@app.list_tools()
async def list_tools() -> list[Tool]:
    """List available tools"""
    return [
        Tool(
            name="get_portfolio",
            description="Get current portfolio information including holdings and balance",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="get_market_data",
            description="Get real-time market data for a specific stock",
            inputSchema={
                "type": "object",
                "properties": {
                    "stock_code": {
                        "type": "string",
                        "description": "Stock code (e.g., '005930' for Samsung)"
                    }
                },
                "required": ["stock_code"]
            }
        ),
        Tool(
            name="get_recommendations",
            description="Get AI-generated stock recommendations based on technical analysis",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="get_trade_history",
            description="Get recent trade history",
            inputSchema={
                "type": "object",
                "properties": {
                    "limit": {
                        "type": "number",
                        "description": "Number of trades to return (default: 20)"
                    }
                },
                "required": []
            }
        ),
        Tool(
            name="analyze_stock",
            description="Perform detailed technical analysis on a stock",
            inputSchema={
                "type": "object",
                "properties": {
                    "stock_code": {
                        "type": "string",
                        "description": "Stock code to analyze"
                    }
                },
                "required": ["stock_code"]
            }
        )
    ]


@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Handle tool calls"""
    
    try:
        async with httpx.AsyncClient(timeout=30.0, headers=_BACKEND_HEADERS) as client:
            if name == "get_portfolio":
                response = await client.get(f"{BACKEND_URL}/portfolio")
                data = response.json()
                return [TextContent(
                    type="text",
                    text=_dumps(data)
                )]
            
            elif name == "get_market_data":
                stock_code = arguments.get("stock_code")
                response = await client.get(
                    f"{BACKEND_URL}/market-data/prices",
                    params={"codes": stock_code}
                )
                data = response.json()
                return [TextContent(
                    type="text",
                    text=_dumps(data)
                )]
            
            elif name == "get_recommendations":
                response = await client.get(f"{BACKEND_URL}/recommendations/rule-based")
                data = response.json()
                return [TextContent(
                    type="text",
                    text=_dumps(data)
                )]
            
            elif name == "get_trade_history":
                limit = arguments.get("limit", 10 if settings.TOKEN_SAVE_MODE else 20)
                response = await client.get(
                    f"{BACKEND_URL}/trades",
                    params={"limit": limit}
                )
                data = response.json()
                return [TextContent(
                    type="text",
                    text=_dumps(data)
                )]
            
            elif name == "analyze_stock":
                stock_code = arguments.get("stock_code")
                # Get chart data
                response = await client.get(
                    f"{BACKEND_URL}/market-data/chart/{stock_code}",
                    params={"period": "D", "realtime": "true"}
                )
                data = response.json()
                return [TextContent(
                    type="text",
                    text=_dumps(data)
                )]
            
            else:
                return [TextContent(
                    type="text",
                    text=f"Unknown tool: {name}"
                )]
    
    except Exception as e:
        return [TextContent(
            type="text",
            text=f"Error calling tool {name}: {str(e)}"
        )]


async def main():
    """Run the MCP server with stdio transport"""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )


if __name__ == "__main__":
    asyncio.run(main())