server.py5.12 kB
#!/usr/bin/env python3
"""
股票分析SSE服务器 - 官方推荐方案最终版
基于Starlette和mcp.py库内置的SseServerTransport，确保协议兼容性。
"""
import json
import logging
import akshare as ak
import uvicorn
from datetime import datetime
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Mount, Route
# --- 配置 ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# --- MCP服务器初始化 ---
mcp = FastMCP("StockAnalysisServer")
# --- 工具实现 ---
@mcp.tool()
async def get_stock_realtime_data(symbol: str) -> str:
"""
获取指定股票代码的实时行情数据。
:param symbol: 股票代码, 例如 '000001'
:return: JSON格式的股票数据字符串
"""
logger.info(f"执行工具 get_stock_realtime_data，参数: {symbol}")
try:
df = ak.stock_zh_a_spot_em()
stock_data = df[df['代码'] == symbol]
if stock_data.empty:
result = {"error": f"未找到股票代码 {symbol} 的数据"}
else:
data = stock_data.iloc[0]
result = {
"股票代码": symbol,
"股票名称": data['名称'],
"最新价": float(data['最新价']),
"涨跌幅": float(data['涨跌幅']),
"更新时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
return json.dumps(result, ensure_ascii=False)
except Exception as e:
logger.error(f"获取实时数据失败: {e}")
error_result = {"error": f"获取实时数据失败: {str(e)}"}
return json.dumps(error_result, ensure_ascii=False)
@mcp.tool()
async def get_market_sentiment() -> str:
"""
获取当前市场的整体情绪分析。
:return: JSON格式的市场情绪数据字符串
"""
logger.info("执行工具 get_market_sentiment")
try:
df_market = ak.stock_zh_a_spot_em()
up_count = len(df_market[df_market['涨跌幅'] > 0])
down_count = len(df_market[df_market['涨跌幅'] < 0])
total_count = len(df_market)
sentiment = "乐观" if up_count > down_count else "悲观" if down_count > up_count else "中性"
result = {
"上涨股票数": up_count,
"下跌股票数": down_count,
"市场情绪": sentiment,
"分析时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
return json.dumps(result, ensure_ascii=False)
except Exception as e:
logger.error(f"获取市场情绪失败: {e}")
error_result = {"error": f"获取市场情绪失败: {str(e)}"}
return json.dumps(error_result, ensure_ascii=False)
# --- SSE传输和Starlette应用设置 ---
# 此部分完全遵循参考实现，能正确处理MCP-over-SSE协议。
sse_transport = SseServerTransport('/messages/')
async def sse_handler(request):
"""处理初始SSE连接并运行MCP服务器会话。"""
async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams:
await mcp._mcp_server.run(streams[0], streams[1], mcp._mcp_server.create_initialization_options())
# 定义Starlette应用的路由。
app = Starlette(
debug=True,
routes=[
Route('/sse', endpoint=sse_handler),
Mount('/messages/', app=sse_transport.handle_post_message)
]
)
# --- 服务器执行 ---
if __name__ == "__main__":
port = 8001 # 修改为8001端口
# 打印FastMCP logo和启动信息
print("""
███████╗ █████╗ ███████╗████████╗███╗ ███╗ ██████╗██████╗
██╔════╝██╔══██╗██╔════╝╚══██╔══╝████╗ ████║██╔════╝██╔══██╗
█████╗ ███████║███████╗ ██║ ██╔████╔██║██║ ██████╔╝
██╔══╝ ██╔══██║╚════██║ ██║ ██║╚██╔╝██║██║ ██╔═══╝
██║ ██║ ██║███████║ ██║ ██║ ╚═╝ ██║╚██████╗██║
╚═╝ ╚═╝ ╚═╝╚══════╝ ╚═╝ ╚═╝ ╚═╝ ╚═════╝╚═╝
""")
logger.info("启动基于官方推荐方案的股票分析SSE服务器...")
logger.info(f"服务器名称: 股票分析MCP服务器")
logger.info(f"服务器版本: v1.0.0")
logger.info(f"描述: 基于AKShare API的股票分析工具，支持SSE协议")
logger.info(f"MCP主端点 (请在CherryStudio中配置此URL): http://127.0.0.1:{port}/sse")
logger.info("所有依赖包检查通过")
logger.info("正在启动MCP服务器...")
uvicorn.run(app, host="127.0.0.1", port=port)