A股实时行情MCP服务器
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": {}
}详细使用说明请参考 MCP使用指南
🛠️ 命令行工具使用
安装后可以使用命令行工具:
# 查询股票价格
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
Latest Blog Posts
- Who's Calling? MCP Hosts Are an Identity Blind Spot (And the Spec Knows It)By Om-Shree-0709 on .mcpAgent IdentityOAuth 2.1
- Your AI Chatbot Just Exposed Your CEO's Salary to an InternBy Om-Shree-0709 on .Agent IdentityMCP SecurityOAuth Delegation
- Why MCP Servers Need Execution Sandboxing (And Why Your Current Stack Isn't Enough)By Om-Shree-0709 on .Agentic AiPrompt InjectionWebAssembly
MCP directory API
We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/Llldmiao/a-stock-mcp-server'
If you have feedback or need assistance with the MCP directory API, please join our Discord server