# 环境要求：Python 3.8+# 依赖包：pip install mcp requests
import asyncio
import json
from typing import Any, Sequence
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
    CallToolRequest,
    CallToolResult,
    ListToolsRequest,
    ListToolsResult,
    Tool,
    TextContent,
)

# 创建MCP服务器
server = Server("stock-analysis-tools")

@server.list_tools()
async def handle_list_tools() -> ListToolsResult:
    """列出所有可用工具"""
    return ListToolsResult(
        tools=[
            Tool(
                name="get_stock_price",
                description="获取指定股票的当前价格",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "symbol": {
                            "type": "string",
                            "description": "股票代码，如AAPL、GOOGL"
                        }
                    },
                    "required": ["symbol"]
                }
            ),
            Tool(
                name="get_stock_news",
                description="获取指定股票的最新新闻",
                inputSchema={
                    "type": "object", 
                    "properties": {
                        "symbol": {
                            "type": "string",
                            "description": "股票代码"
                        },
                        "limit": {
                            "type": "integer",
                            "description": "新闻数量限制",
                            "default": 5
                        }
                    },
                    "required": ["symbol"]
                }
            ),
            Tool(
                name="calculate_portfolio_value",
                description="计算投资组合价值",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "investments": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "symbol": {"type": "string"},
                                    "shares": {"type": "number"}
                                }
                            },
                            "description": "投资组合详情"
                        }
                    },
                    "required": ["investments"]
                }
            )
        ]
    )

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any]) -> CallToolResult:
    """处理工具调用"""
    if name == "get_stock_price":
        symbol = arguments["symbol"]
        # 模拟股票价格查询
        price = await get_mock_stock_price(symbol)
        return CallToolResult(
            content=[TextContent(type="text", text=f"{symbol}当前价格: ${price}")]
        )
    
    elif name == "get_stock_news":
        symbol = arguments["symbol"]
        limit = arguments.get("limit", 5)
        news = await get_mock_stock_news(symbol, limit)
        return CallToolResult(
            content=[TextContent(type="text", text=news)]
        )
    
    elif name == "calculate_portfolio_value":
        investments = arguments["investments"]
        total_value = await calculate_portfolio(investments)
        return CallToolResult(
            content=[TextContent(type="text", text=f"投资组合总价值: ${total_value:,.2f}")]
        )
    
    else:
        raise ValueError(f"未知工具: {name}")

async def get_mock_stock_price(symbol: str) -> float:
    """模拟获取股票价格"""# 实际项目中这里会调用真实的股票API
    prices = {"AAPL": 150.0, "GOOGL": 2800.0, "MSFT": 300.0}
    return prices.get(symbol, 100.0)

async def get_mock_stock_news(symbol: str, limit: int) -> str:
    """模拟获取股票新闻"""
    news = [
        f"{symbol}发布最新财报，营收增长超预期",
        f"分析师上调{symbol}目标价格",
        f"{symbol}推出新产品线，市场反应积极"
    ]
    return "\n".join(news[:limit])

async def calculate_portfolio(investments: list) -> float:
    """计算投资组合价值"""
    total = 0.0for investment in investments:
        price = await get_mock_stock_price(investment["symbol"])
        total += price * investment["shares"]
    return total

async def main():
    """启动MCP服务器"""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="stock-analysis-tools",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=None,
                    experimental_capabilities=None,
                ),
            ),
        )

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