a-stock-mcp-server by Llldmiao - MCP Server

A股实时行情MCP服务器 v2.0

这是一个基于Model Context Protocol (MCP) 的A股实时行情查询服务器，采用多数据源架构，支持查询A股实时价格、历史数据、财务信息等。专为AI助手和金融分析工具设计。

🚀 新版本特性 (v2.0)

多数据源架构: 支持故障转移和负载均衡
模块化设计: 工具和数据源完全解耦
智能缓存: 提高响应速度和减少API调用
增强错误处理: 完善的异常处理和重试机制
完整测试覆盖: 单元测试和集成测试

功能特性

🔥 核心功能

实时价格查询: 获取A股实时价格、涨跌幅、成交量等
股票基本信息: 查询股票的基本信息和公司概况
市场概况: 获取上证、深证等主要指数信息
历史数据: 查询股票的历史K线数据
财务数据: 获取利润表、资产负债表、现金流量表

🛠️ 支持的工具

get_realtime_price - 获取实时价格
get_stock_info - 获取股票基本信息
get_market_summary - 获取市场概况
get_stock_history - 获取历史数据
get_financial_data - 获取财务数据

安装和使用

🚀 PyPI安装（推荐）

pip install a-stock-mcp-server

📦 从源码安装

# 克隆仓库
git clone https://github.com/Llldmiao/a-stock-mcp-server.git
cd a-stock-mcp-server

# 使用Poetry安装（推荐）
poetry install

# 或者使用pip安装
pip install -r requirements.txt
pip install -e .

🤖 MCP服务器使用

快速配置

在你的MCP客户端配置文件中添加：

{
  "mcpServers": {
    "a-stock-realtime": {
      "command": "python",
      "args": ["-m", "a_stock_mcp_server"]
    }
  }
}

可用工具

get_realtime_price - 获取实时价格
- 参数: symbol (股票代码，如"000001")
get_stock_info - 获取股票基本信息
- 参数: symbol (股票代码)
get_market_summary - 获取市场概况
- 参数: 无
get_stock_history - 获取历史数据
- 参数: symbol, period(daily/weekly/monthly), start_date, end_date
get_financial_data - 获取财务数据
- 参数: symbol, report_type(income/balance/cashflow)

使用示例

// 查询平安银行实时价格
{
  "name": "get_realtime_price",
  "arguments": {
    "symbol": "000001"
  }
}

// 查询市场概况
{
  "name": "get_market_summary",
  "arguments": {}
}

详细使用说明请参考

🛠️ 命令行工具使用

安装后可以使用命令行工具：

# 查询股票价格
a-stock-cli price -s 000001

# 查询股票信息
a-stock-cli info -s 000001

# 查询市场概况
a-stock-cli market

🧪 本地测试

# 运行完整测试
python3 -m pytest tests/ -v

# 运行测试客户端
python3 -m a_stock_mcp_server.test_client

# 运行旧版测试（兼容性）
python3 local_test.py

📚 Python代码使用

新版本 (推荐)

import asyncio
from a_stock_mcp_server.test_client import AStockTestClient

async def main():
    client = AStockTestClient()
    
    # 查询平安银行实时价格
    result = await client.call_tool("get_realtime_price", {"symbol": "000001"})
    print(result)
    
    # 健康检查
    health = await client.health_check()
    print("数据源状态:", health)

asyncio.run(main())

旧版本 (兼容性)

import asyncio
from a_stock_mcp_server.local_test import AStockLocalTest

async def main():
    server = AStockLocalTest()
    
    # 查询平安银行实时价格
    result = await server.call_tool("get_realtime_price", {"symbol": "000001"})
    print(result)

asyncio.run(main())

输出示例：

股票代码: 000001
股票名称: 平安银行
当前价格: ¥11.45
涨跌额: +0.04
涨跌幅: +0.35%
成交量: 834,651.0
成交额: ¥955,004,096.91
最高价: ¥11.51
最低价: ¥11.37
开盘价: ¥11.41
昨收价: ¥11.41
换手率: 0.43%
市盈率: 4.47
市净率: 0.5

使用示例

查询实时价格

{
  "tool": "get_realtime_price",
  "arguments": {
    "symbol": "000001"
  }
}

查询历史数据

{
  "tool": "get_stock_history", 
  "arguments": {
    "symbol": "000001",
    "period": "daily",
    "start_date": "20240101",
    "end_date": "20241231"
  }
}

查询财务数据

{
  "tool": "get_financial_data",
  "arguments": {
    "symbol": "000001",
    "report_type": "income"
  }
}

🔧 常用股票代码

股票名称	代码	市场
平安银行	000001	深市
万科A	000002	深市
中国平安	601318	沪市
招商银行	600036	沪市
工商银行	601398	沪市
建设银行	601939	沪市
农业银行	601288	沪市

架构设计

多数据源架构

数据源层 (Data Sources)
├── BaseDataSource (抽象基类)
├── AKShareDataSource (AKShare实现)
├── DataSourceManager (数据源管理器)
└── CacheManager (缓存管理器)

工具层 (Tools)
├── BaseTool (工具基类)
├── RealtimePriceTool (实时价格)
├── StockInfoTool (股票信息)
├── MarketSummaryTool (市场概况)
├── StockHistoryTool (历史数据)
└── FinancialDataTool (财务数据)

服务器层 (Server)
├── AStockMCPServer (主服务器)
├── AStockMCPServerWithAKShare (兼容包装)
└── AStockTestClient (测试客户端)

核心特性

故障转移: 自动切换到备用数据源
智能缓存: 减少API调用，提高响应速度
模块化设计: 易于扩展和维护
完整测试: 单元测试和集成测试覆盖

数据源

本MCP服务器使用 AKShare 作为主要数据源：

免费、开源
数据更新及时
支持多种金融数据
社区活跃

扩展数据源

架构支持添加新的数据源，如：

新浪财经API
腾讯财经API
其他金融数据提供商

详细扩展指南请参考

扩展建议

1. 多数据源支持

集成新浪财经API
集成腾讯财经API
添加数据源故障转移

2. 缓存机制

添加Redis缓存
减少API调用频率
提高响应速度

3. 数据验证

添加数据有效性检查
异常数据处理
错误重试机制

4. 更多功能

技术指标计算
股票筛选器
实时推送
历史回测

注意事项

数据延迟: AKShare数据可能有15-20分钟延迟
API限制: 注意API调用频率限制
数据准确性: 仅供参考，投资决策请谨慎
网络依赖: 需要稳定的网络连接

故障排除

常见问题

问题1: 导入错误

ModuleNotFoundError: No module named 'akshare'

解决方案: 安装AKShare

pip3 install akshare

问题2: 网络超时

获取数据失败: timeout

解决方案: 检查网络连接，可能需要代理

问题3: 股票代码不存在

未找到股票代码: XXXXXX

解决方案: 检查股票代码格式，确保是6位数字

问题4: 数据为空

解决方案: 检查股票代码格式是否正确

日志调试

import logging
logging.basicConfig(level=logging.DEBUG)

📝 完整使用示例

#!/usr/bin/env python3
import asyncio
from a_stock_mcp_server.local_test import AStockLocalTest

async def main():
    # 创建服务器实例
    server = AStockLocalTest()
    
    # 查询平安银行实时价格
    print("=== 平安银行实时价格 ===")
    price = await server.call_tool("get_realtime_price", {"symbol": "000001"})
    print(price)
    
    # 查询股票基本信息
    print("\n=== 平安银行基本信息 ===")
    info = await server.call_tool("get_stock_info", {"symbol": "000001"})
    print(info)
    
    # 查询市场概况（只显示前10个）
    print("\n=== 市场概况（前10个）===")
    market = await server.call_tool("get_market_summary", {})
    lines = market.split('\n')
    for line in lines[:12]:  # 标题 + 前10个指数
        print(line)

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

贡献

欢迎提交Issue和Pull Request来改进这个项目！

许可证

MIT License