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"


@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) 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())