资讯动态

基于MCP协议封装股票数据SDK:AI Agent行情的标准化接入实践

发布时间:2026/9/9 21:15:20 来源:尧图企业网站定制
最近在折腾 AI Agent 和股票行情数据对接的时候把stock-sdk-mcp这个项目完整跑通了一遍。这名字看着像三个概念的拼凑其实核心就一句话把股票数据 SDK 的能力通过 MCP 协议暴露给 AI 客户端让 Claude、Cursor 或者自研 Agent 能直接用自然语言查行情、拉K线、搜标的。这篇是实操之后的整理从方案选型到代码实现再到踩坑记录一次性讲清楚。先说明一下适用人群正在做智能投顾助手、量化研究辅助工具的开发者或者单纯想给自己的 AI 工作流里加一个“能查股票数据”技能的玩家这篇都应该能帮上忙。即使你不做股票方向里面 MCP Server 的设计思路和排错方法对任何“把第三方 SDK 接进 MCP”的场景都通用。1. 项目概述stock-sdk-mcp 到底解决什么问题1.1 从一次“AI 炒股助手”的需求说起我一开始接到的需求很朴素做一个能让 AI 帮忙查股票的小工具。用户直接在对话框里问“贵州茅台今天什么情况”“最近新能源板块哪些票涨得多”AI 要能给出真实、及时的数据而不是凭训练语料瞎编。第一反应是直接把行情 SDK 封装成函数通过大模型的 Function Calling 来调用。做了一版 demo 之后问题陆续冒出来每个函数要手写 JSON Schema不同客户端Claude、Cursor、自研 Agent接入方式各不相同模型经常把参数名理解错鉴权逻辑散落在各个函数里。改起来很痛苦加一个新工具要动好几处代码。后来我把服务改成了 MCP Server 架构用stock-sdk-mcp作为项目代号做了一次重构效果立竿见影。1.2 MCP 在这里扮演的角色MCP 全称 Model Context Protocol是 Anthropic 在 2024 年底开源的一套标准化协议用来统一 AI 应用和外部工具/数据源的通信方式。通俗点说它给 AI 客户端定义了一个“标准插座”任何符合这个协议的服务都能即插即用。在stock-sdk-mcp这个项目里MCP 起的就是中间翻译层的作用左侧是股票数据 SDK负责和真实的数据服务通信处理连接、鉴权、协议解析中间是 MCP Server把 SDK 的能力封装成一个个工具Tool右侧是任意支持 MCP 的客户端通过 JSON-RPC 调用这些工具拿到结构化数据。这意味着我只需要写一套 Server就能同时服务 Claude Desktop、Cursor、VS Code 里的 AI 插件、自研 Agent 等所有支持 MCP 的客户端。而如果用传统 SDK 接入每个场景都要分别适配成本完全不在一个量级。2. 方案设计为什么选择 MCP Server 而非其他接入方式2.1 接入方式横向对比我实际对比了三种主流方案这里直接给结论。对比维度直接调 SDK Function Calling自研工具函数注入MCP Server 标准化客户端适配成本每种客户端单独开发仅自研系统可用一次开发多端复用工具描述管理手写 JSON Schema易出错散落在代码中统一注册自动生成鉴权与会话管理每个函数都要处理全局但要自行设计框架级支持生态兼容性无法跨平台不适用Claude/Cursor/自研全兼容调试成本中等高有生态工具方便我之前那个 Function Calling 版本最大的问题是工具定义和业务逻辑纠缠在一起模型偶尔会把股票代码和日期参数结合起来乱填。比如用户问“昨天涨得最多的银行股”模型会自己编一个日期参数传进去导致 SDK 返回空数据。MCP 模式下工具参数同样由模型决定但协议层的严格校验帮我挡掉了一部分脏数据同时错误信息能标准地返回给客户端而不是在 SDK 内部抛一个被吞掉的异常。2.2 MCP Server 的边界设计设计 Server 结构前我给自己定了三条边界这也是我强烈建议每个实践者想清楚的第一条只读优先。行情查询、历史数据、标的搜索这些是纯只读操作放进 MCP Server 没有问题。涉及委托下单、撤单、资金划转这类写操作我明确不在 Server 里暴露。即便你的 SDK 支持实盘交易也建议把 MCP Server 限定在模拟盘或只读行情范围这是安全和合规的底线。第二条工具粒度要粗不细。一个 MCP 工具尽量对应一个完整业务动作而不是一个底层函数。比如我要查“个股资金流向”就封装成get_capital_flow一个工具内部完成从 SDK 取数、清洗、格式化的全过程不让 AI 自己去拼多个原子调用。工具细了模型编排链路过长出错的概率成倍上升。第三条失败要说得清楚。MCP 工具返回给模型的不只是数据还有错误语义。如果 SDK 返回“连接超时”你不能直接抛出异常让模型猜而要返回一个结构化错误信息比如“行情服务暂时不可用请稍后重试”或“股票代码不存在请检查输入”。模型能理解错误原因才能向用户给出合理的答复。3. 环境准备与 SDK 接入细节3.1 SDK 选型与初始化做这个项目前我评估了几种拿行情数据的路子包括免费的数据接口、开源 Python 库、还有几个商业行情 SDK。选择标准就三条文档是否完整、是否稳定维护、连接方式是否符合服务端场景。我只用了半天就把一个商用的 Python 行情 SDK 跑通了。这类 SDK 通常提供 WebSocket 或 TCP 长连接支持订阅实时行情也支持一次性 HTTP 拉取历史数据。我们的项目基于stock-sdk这个库来做封装它的初始化方式是创建 Client 实例然后 connect下面是我实际用过的初始化代码from stock_sdk import StockClient, StockConfig config StockConfig( token这里填你的访问令牌, endpointwss://api.example.com/market, timeout10, max_retries3, ) client StockClient(config) client.connect() print(SDK 连接状态:, client.is_connected)这里有个容易忽略的细节connect 不是一次性动作行情长连接可能随时断开。我第一版代码只连了一次结果跑了半小时后行情就静止了排查半天发现是连接 socket 被服务端关闭了。正确做法是为 SDK 挂上断线重连的回调或者用监听线程定时检查连接状态断开就自动重连。3.2 项目结构设计stock-sdk-mcp是一个典型的 Python 项目我建议的结构是这样的stock-sdk-mcp/ ├── pyproject.toml ├── requirements.txt ├── stock_server.py # MCP Server 入口 ├── stock_sdk/ │ ├── __init__.py │ ├── client.py # SDK 封装与连接管理 │ ├── tools.py # 工具函数定义 │ └── formatter.py # 数据格式化与清洗 └── config/ └── settings.toml # 配置文件token、端口等依赖管理我走了最稳妥的路子pip install mcp stock-sdk。这里提示一下很多同学会忽略虚拟环境的问题直接把包装到全局 Python 里后面客户端配置绝对路径时特别容易路径错乱。建议用 venv 建一个干净环境锁定依赖版本比如我当前用的mcp1.0版本不同API 差异比较大网上很多旧教程写的是mcp.tool()装饰器但新版本推荐用mcp.add_tool()或 FastMCP 的mcp.tool()动手前先确认版本。4. 核心实现从 SDK 方法到 MCP 工具的完整封装4.1 工具注册与参数 SchemaMCP Server 的核心工作是“注册工具”。我用的是mcp官方 Python SDK 的 FastMCP 接口写起来很简洁。每个工具就是一个函数函数名、参数、描述都会自动映射为模型可见的工具定义。from mcp.server.fastmcp import FastMCP mcp FastMCP(stock-sdk-mcp) mcp.tool( namesearch_stock, description根据关键词搜索 A 股股票代码和名称关键词可以是代码或名称的一部分, ) def search_stock(keyword: str) - list[dict]: 调用 SDK 的搜索接口返回匹配的股票列表。 records client.search_stock(keyword) return [ {code: r.code, name: r.name, exchange: r.exchange} for r in records ]这里最关键的是description字段。写 tool 描述不是给程序员看的是给大模型看的。描述写得好不好直接决定模型能不能在正确的场景下调用这个工具。我见过有人写“查询股票”结果模型完全不知道什么时候该用这个工具。我的写法是“根据关键词搜索 A 股股票代码和名称关键词可以是代码或名称的一部分”——把触发条件、输入规则、返回内容都讲清楚。参数方面MCP 支持标准的 JSON SchemaFastMCP 会根据函数签名自动推导但复杂参数最好手动补充说明。比如查询 K 线需要起始日期和终止日期模型可能不理解格式mcp.tool( nameget_kline_history, description获取股票的历史 K 线数据支持日线/周线/月线, ) def get_kline_history( code: str, start_date: str , end_date: str , period: str day, ) - list[dict]: 参数说明: code: 6 位股票代码如 600519 start_date: 起始日期格式 YYYY-MM-DD默认空表示最近一年 end_date: 结束日期格式 YYYY-MM-DD默认空表示今天 period: K 线周期可选 day/week/month ...看到没有docstring 里也写了参数说明这在 FastMCP 里很有用模型会把这些信息作为 tool 描述的一部分来理解。4.2 典型工具的实现我封装了几个高频工具逐个说一下实现要点。第一个是get_realtime_quote实时行情查询。这个工具背后调用 SDK 的查询接口从多个数据源聚合出价格、涨跌、成交量、成交额等字段。返回前我会做一层格式化把数字改成人类可读形式比如成交量从“股”转为“万手”避免模型对单位产生错误推断。第二个是get_market_overview大盘概况用于回答“今天大盘怎么样”这类问题。这个工具拿到上证指数、深证成指、创业板指的涨跌数据再配上成交额统计返回一个紧凑摘要。第三个是search_stock刚才已经演示过了。还有个get_stock_list_by_sector按行业板块筛股票支撑“银行板块今天涨得怎么样”“新能源车概念股有哪些”这类问题。每个工具内部都包了 try-excepttry: data client.get_quote(code) if not data: return {status: no_data, message: f股票 {code} 暂无行情数据} return {status: ok, data: data} except Exception as e: return {status: error, message: f行情获取失败: {str(e)}}这个写法有个好处无论发生什么MCP 都能拿到一个可读的 JSON 响应而不是抛出异常把整个对话打断。模型拿到status: error后会尝试用更口语化的方式告诉用户“我自己查不到可能是代码写错了”体验好很多。4.3 错误处理与状态管理MCP Server 的运行状态直接依赖 SDK 的连接状态所以我把“重连”做成全局管理器而不是每个工具里各自处理import threading import time def _keepalive(): while True: if not client.is_connected: try: client.connect() except Exception: pass time.sleep(3) threading.Thread(target_keepalive, daemonTrue).start()这个后台线程每 3 秒检查一次连接断线就自动重连。另外网络请求要设置超时默认 SDK 的超时可能很长AI 客户端等不了那么久。我在配置里把超时控制在 10 秒以内避免模型长时间空等。这些细节看起来不值得写实际对用户体验影响很大。5. 客户端接入与验证5.1 Claude Desktop/Cursor 的配置方法Server 写好后接入客户端就完全是配置工作了。以 Claude Desktop 为例配置文件在claude_desktop_config.json里加一段mcpServers{ mcpServers: { stock-sdk-mcp: { command: python, args: [ /absolute/path/to/stock_server.py ], env: { STOCK_TOKEN: xxx } } } }注意几个点。第一command要写成虚拟环境里的 python 绝对路径或者你确保系统 PATH 能找到的 python不然后面一直报“spawn python ENOENT”你都不知道原因。第二args里的路径也要用绝对路径相对路径在部分客户端里解析会出问题。第三环境变量可以在env里传递不要把 token 硬编码在代码里。Cursor 的配置入口在 Settings → MCP也可以通过项目根目录的.cursor/mcp.json文件管理格式类似{ mcpServers: { stock-sdk-mcp: { command: python, args: [/absolute/path/to/stock_server.py] } } }配置完成后重启客户端在 MCP 面板里看到 server 状态为 connected 就成功了。5.2 实测调用示例连接成功后我用自然语言试了这样一串问题“贵州茅台目前的价格和历史最高价差多少”——模型自动调用get_realtime_quote和get_kline_history计算出差值。“帮我把今天的银行板块涨幅前 5 列出来。”——模型调用search_stock拿到银行板块名单再逐个get_realtime_quote排序后输出。“最近一个月的平安银行日 K 线数据帮我分析趋势。”——模型拉取 K 线数据结合上下文做技术分析。这些场景下MCP 工具的输出准确且结构化模型的分析也有据可依。对比之前用 Function Calling 的版本稳定性提升很明显因为工具返回的字段都是约定好的 Schema模型理解成本降低编造数据的概率大大下降。6. 常见问题与排查技巧实录6.1 配置类问题我在实践过程中踩了不少坑整理成排查速查表分享给大家。现象可能原因排查方法客户端提示spawn python ENOENTcommand 路径找不到检查 python 绝对路径Windows 上要用 venv/Scripts/python.exe不能写pythonServer 启动报ConfigurationErrorMCP 配置格式不对用 JSON 校验工具检查配置文件注意末位逗号这些细节客户端 MCP 面板显示timeoutServer 卡在启动过程先在终端手动运行 server 脚本确认能正常启动且不报错工具调用返回“400 error”SDK 的 token 没传或传错检查环境变量 STOCK_TOKEN 是否设置server 进程有没有继承该变量行情数据一直为空SDK 没正常连接打印client.is_connected状态确认数据类型和订阅设置是否正确6.2 运行类问题最让人头疼的是stdio transport报错。MCP 默认走 stdio客户端通过标准输入输出和 Server 通信。这个模式要求 Server 的全部日志打印都必须走 stderr 或文件不能往 stdout 写任何调试信息。我第一次跑的时候在代码里加了一堆print([DEBUG] got data)结果客户端直接报JSON-RPC parse error。排查方法是在终端跑 server如果看见 stdout 有杂七杂八的输出全部改成logging输出到 stderr。还有连接池打满的问题。SDK 和行情服务的连接数有限当模型的多个工具调用并发发生时SDK 内部连接池可能被耗尽新请求阻塞。解决办法是给 SDK 合理设置并发上限或者在 Server 层做线程池限制控制同时操作 SDK 的请求数量。时区问题也值得提醒。行情服务返回的时间可能是 UTC也可能带时区信息如果不处理K 线图上日期会整体偏移 8 个小时特别是在日线数据上可能出现“今天的数据看不到明天数据出现”的诡异现象。我写了一个formatter.py统一把所有时间字段转成本地时区格式再返回给 MCP 客户端。这个小函数回购项目里极其重要。6.3 数据准确性问题最后说一个隐蔽且容易被忽视的问题SDK 返回的数字精度。有些历史行情数据的字段是浮点SDK 底层可能直接序列化了精度丢失后的值例如 12.9999999999 这种。我在格式化层统一用 Decimal 做处理和四舍五入保留两位小数。别小看这个问题模型拿到的数据如果带着一长串小数它可能会“一本正经”地分析出毫无意义的小差异。7. 后续还可以这样扩展stock-sdk-mcp目前完成的是行情查询这一层但 MCP 的想象空间远不止于此。我个人的下一步计划是把两个方向补上一是把财务报表数据接进来模型可以直接回答“某公司过去三年的营收增速”这类问题现在查询链路太长二是把技术指标计算服务也做成 MCP 工具让模型能基于解析后的数据直接算 MACD、RSI 这些指标而不是自己在 Prompt 里写计算逻辑。另外一个实用的小技巧如果不想用 Python 的 FastMCP也可以用 TypeScript 的modelcontextprotocol/sdk重写一遍 Server逻辑完全一样只是语言不同。选择哪种语言取决于你的项目技术栈和部署环境。最后一点心得。做这类 AI 数据接入项目我发现最花时间的其实不是写代码而是理解数据如何被模型消费。MCP 只解决了传输协议的问题数据长什么样、单位是什么、边界在哪里、失败怎么表达这些决定了大模型能不能正确使用工具。没有想清楚这层关系就算协议接得再完美模型回答的质量也上不去。做stock-sdk-mcp这轮实践我在这个层面的收获比拿到一份能跑的代码要大得多。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价