最近在整理一个挺实在的需求:把股票数据能力接进 AI。场景其实很朴素——我在用 AI 工具查行情的时候,它总是回同一句“我无法实时获取股票数据,请自行查询”。模型本身没说谎,它确实没有实时数据源。于是我把手头的行情服务包了一层,做成一个叫 stock-sdk-mcp 的 MCP 服务。跑通之后,AI 就能直接调用工具拿实时报价、查 K 线、搜股票代码,整个过程对用户来说就像在对话框里多了一个“看盘插件”。
这篇文章会把整个实践过程完整复盘一遍,从需求拆解、架构设计、代码实现到问题排查都有覆盖。适合手里有股票数据接口、想把行情能力接进 AI 产品的开发者参考;即使暂时不写代码,也可以把它当作理解 MCP 协议的一次完整实例来读。我尽量把每一步为什么这么做、踩过哪些坑写明白,方便你直接照着落地。
1. 项目背景与需求拆解:AI 为什么需要股票数据工具
1.1 直观痛点:模型看不见实时行情
大模型本质上是“离线”的。它真正擅长的是把文字整理成结构化的结论,而不是实时监听外部世界。训练时的数据截止日期决定了它天然不知道今天的价格、当下的涨跌。你在对话里问“现在某某股票多少钱”,它只能靠训练知识里的旧数据给你一个模糊答案,或者直接告诉你没有实时能力。
我最初的诉求很简单:让 AI 能回答行情类问题,并且答案来自真实数据,而不是模型瞎编。这就必须给 AI 一个“取数工具”。传统的做法是把行情信息复制粘贴到对话框里,让模型做解读。这种做法对单次问答还行,但如果你有十个股票要盯、有复盘模板要填、有例行日报要生成,人工复制粘贴的工作量会迅速失控。真正该做的是让 AI 自己去调取数据。
1.2 方案选型:为什么不能只靠直接调用 HTTP API
有人可能会问:直接让 AI 调用行情服务的 HTTP 接口不行吗?当然可以,你的代码里可以写一个 fetch 函数,然后把行情数据塞进提示词。但这样做的代价是:每接一个 AI 应用,都要重新写一遍调用逻辑、重新处理鉴权、重新定义参数格式。折腾一次还行,反复折腾就烦了。
更重要的是,现在的 AI 客户端生态已经形成了“工具即插即用”的交互习惯。与其自己维护一套私有协议,不如直接使用业界已经统一的标准接入方式。MCP(Model Context Protocol)解决的就是这个问题:它把“数据能力”抽象成标准化的工具/资源,任何支持 MCP 的 AI 客户端都能直接发现和调用。
另外还有一个现实因素:模型厂商在快速迭代自己的工具调用方案,但 MCP 作为开放协议是跨厂商、跨产品的。我写完一个服务端,桌面端 AI、命令行工具、甚至团队内部自研的客户端,只要实现了 MCP 客户端规范,就能直接复用,不需要为每个平台单独适配。这一点是最终让我下决心选它的原因。
1.3 三种接入方式的对比
先放一张表,把这个选型逻辑说透:
| 方案 | 实现成本 | 复用性 | 维护难度 | 适用场景 |
|---|---|---|---|---|
| 直接在 AI 应用代码里调 HTTP API | 低,几行请求代码 | 差,每个应用都要重接 | 中,参数和鉴权散落各处 | 只服务一个固定应用 |
| 使用模型厂商的 Function Calling | 中,要按厂商格式写函数描述 | 一般,绑定特定模型生态 | 中,模型升级可能要改格式 | 单模型单平台 |
| 通过 MCP 服务封装数据能力 | 中高,要先理解协议 | 高,一次实现多端共用 | 低,能力和协议解耦 | 多个 AI 客户端共用同一套数据 |
我当时的需求恰好是“同一个行情能力要接到多个地方”:自己调试用的桌面助手、团队协作用的聊天机器人、还有不定期写的小工具脚本。如果用 Function Calling,等于每个地方都维护一套函数描述;用 MCP 只需要维护一个服务端,其余全程统一。而且 MCP 的生态在快速成熟,调试面板、客户端支持、SDK 文档都已经比较完善了,上手门槛比前两年低很多。
一句话总结选型逻辑:数据能力只需要实现一次,协议是免费的通用语言。
2. 架构设计与工具集定义:stock-sdk-mcp 的核心设计
2.1 三层架构:接入层、协议层、运维层
动手写代码前,先把整个服务端拆成三层,后面实现就会很清晰:
- 数据接入层:负责对接行情服务商,把市场的 HTTP 接口或者已有 SDK 包一层统一的 Python 接口。无论底层数据源长什么样,上层看到的都是“给我一个股票代码,返回行情对象”。
- MCP 协议层:基于 FastMCP 实现服务端,把业务能力注册成 AI 可见的“工具”。这一层对应的是“能查什么、参数怎么传、返回什么”。
- 运维与调度层:负责缓存、限流、日志、进程守护等和业务无关但决定服务能否长期跑下去的琐碎事。
把三层分开最大的好处是:将来如果你想换一个行情数据源,只需要改接入层,协议层和工具定义都不用动。同样的道理反过来也成立——如果 MCP 协议升级、你要改动工具描述,也不用去碰数据源适配逻辑。
2.2 工具集设计:不贪多,先围绕高频场景
工具集是 AI 能看到的能力清单,设计它的核心原则是:命名要直白、参数要少、返回要结构化。模型在对话里做工具调用时,是根据工具名称和描述来“猜”应该调哪一个的。如果你把一个工具描述写得学术味太重,模型大概率会选错,然后你说它笨——其实是描述没写清楚。
我最初只做了四个工具,覆盖了最常被问到的行情场景:
| 工具名 | 输入参数 | 返回内容 | 典型用户提问 |
|---|---|---|---|
| search_stock_code | 关键词(股票名称/拼音) | 匹配的股票代码列表 | “帮我找一下某白酒股的代码” |
| get_stock_quote | 股票代码(如 SH600XXX) | 最新价、涨跌幅、成交额、更新时间 | “XX 现在多少钱” |
| get_kline | 代码、周期、条数 | K 线序列(时间、开高低收、成交量) | “帮我拉最近 30 天日 K” |
| get_market_snapshot | 市场类型 | 指定市场的指数/板块概览 | “今天大盘整体怎么样” |
先把这四个跑通,稳定之后再加新工具。原因是 AI 工具调用有一个信噪比问题:工具数量越多,模型选错的可能性越大。十个工具里九个不常用,只会增加每次调用的判断成本和出错的概率。我见过有人一口气注册了二十个工具,结果模型频繁把“分红数据”和“股东数据”搞混。少而精,永远是第一原则。
2.3 缓存与限流:两个最容易被忽略的“隐形设计”
很多新手在接入行情数据时只关心“能不能取到数”,完全忽略缓存和限流。结果上线第一天就被数据源服务商限流,甚至连客户端都跟着卡。行情数据的特性决定了缓存策略可以很激进:实时报价在 3 秒内基本不会有肉眼可感知的变化,K 线数据在分钟级别内更是恒定不变。
我当时的方案是:报价接口做 2 秒 TTL 的本地缓存,多个模型请求在 2 秒内命中直接返回缓存,不重复打数据源;K 线接口按“代码+周期”做 30 秒缓存;市场总览做 10 秒缓存。这个设置的逻辑是:宁可让 AI 拿到的数据“老了两秒”,也不要因为一秒钟十几个并发请求而被数据源封掉。对对话场景来说,2 秒延迟根本察觉不到,但请求量能减少 80% 以上。
限流再做一层保险:使用 Python 的信号量把对外部的请求频率限制在数据源允许的 QPS 以内。比如某服务商允许每秒 5 次请求,我就在适配层加一个每秒最多 4 次的节流阀,留出余量。这就像一个水龙头:上游水管再粗,你家水龙头出水慢,你硬开最大只会让水管爆掉。稳住节流,反而能用得更久。
2.4 只读原则与安全边界
接股票数据这件事,天然有个安全红线:永远只做只读操作。行情报价、K 线、市场总览这些都是公开市场数据,可以放心接;但任何需要登录态的账户数据、持仓数据、交易接口,一概不碰。MCP 服务端一旦接了账户能力,等于把用户的敏感凭证暴露给了 AI 调用链,这风险太大,没必要。
API Key 的管理也有讲究。代码里不能写死密钥,必须通过环境变量的方式注入。客户端的 MCP 配置里单独配置 env 字段,服务启动时从 os.environ 读取。这样做的原因是:你的配置文件将来很可能要提交到代码仓库、分享给同事,如果里面有明文密钥,就等于公开了数据源凭证。把密钥放环境变量,既能保证服务在部署时不泄露敏感信息,也不影响本地调试。
3. 实操落地:从空目录到接入 AI 客户端
3.1 工程初始化与依赖准备
先把环境准备好。我用的是 Python 3.10+,MCP 官方 SDK 对 Python 的支持已经很成熟,asyncio 的兼容性也稳定。创建虚拟环境并安装依赖:
python -m venv .venv source .venv/bin/activate pip install mcp这里只装一个mcp就够了,HTTP 请求用标准库的urllib或者requests都行,看你的习惯。我的源码结构很简单:
stock-sdk-mcp/ server.py # MCP 服务端入口,定义工具 datasource_adapter.py # 行情数据源适配层 config.py # 配置读取 requirements.txt为什么要单独拆一个datasource_adapter.py?因为 MCP 工具函数里不应该出现任何具体的 HTTP 请求代码。工具层只负责把模型传来的参数转发给适配层,把适配层返回的结果格式化成字符串。这样将来你想从服务商 A 换成服务商 B,只改动适配层一个文件,协议层完全不用动。
3.2 写一个能直接跑通的 MCP 服务端(Python 代码)
下面是我整理的核心代码,去掉了具体数据源细节,替换成通用的适配层接口。你把自己的行情服务商 SDK 放进适配层即可。
# config.py import os STOCK_API_KEY = os.environ.get("STOCK_API_KEY", "") STOCK_API_BASE = os.environ.get("STOCK_API_BASE", "https://api.example-stock-service.com")# datasource_adapter.py import json import threading import time from urllib import request class StockDataSource: """行情数据源适配层:统一封装底层 HTTP 接口""" def __init__(self, api_key: str, base_url: str, max_qps: int = 4): self.api_key = api_key self.base_url = base_url self.max_qps = max_qps self._lock = threading.Lock() self._last_call_ts = 0.0 self._cache = {} self._cache_time = {} def _throttle(self): """简单的每秒限流""" with self._lock: now = time.time() min_interval = 1.0 / self.max_qps wait_time = min_interval - (now - self._last_call_ts) if wait_time > 0: time.sleep(wait_time) self._last_call_ts = time.time() def _get_cached(self, key: str, ttl: float): """读取缓存,超过 TTL 返回 None""" if key in self._cache and time.time() - self._cache_time[key] < ttl: return self._cache[key] return None def _set_cache(self, key: str, value, ttl: float): self._cache[key] = value self._cache_time[key] = time.time() def get_quote(self, code: str) -> dict: """实时报价,TTL 2 秒""" cache_key = f"quote:{code}" cached = self._get_cached(cache_key, ttl=2) if cached: return cached self._throttle() url = f"{self.base_url}/quote?code={code}&api_key={self.api_key}" with request.urlopen(url, timeout=5) as resp: data = json.loads(resp.read()) self._set_cache(cache_key, data, ttl=2) return data def get_kline(self, code: str, period: str = "1d", limit: int = 20) -> dict: """K 线数据,按代码+周期做 30 秒缓存""" cache_key = f"kline:{code}:{period}:{limit}" cached = self._get_cached(cache_key, ttl=30) if cached: return cached self._throttle() url = f"{self.base_url}/kline?code={code}&period={period}&limit={limit}&api_key={self.api_key}" with request.urlopen(url, timeout=10) as resp: data = json.loads(resp.read()) self._set_cache(cache_key, data, ttl=30) return data# server.py import json import os from mcp.server.fastmcp import FastMCP from datasource_adapter import StockDataSource from config import STOCK_API_KEY, STOCK_API_BASE mcp = FastMCP("stock-sdk-mcp") datasource = StockDataSource( api_key=STOCK_API_KEY, base_url=STOCK_API_BASE, max_qps=4, ) @mcp.tool() def search_stock_code(keyword: str) -> str: """按股票名称或代码片段搜索,返回匹配列表。例如输入“白酒”或“600”,返回相关股票代码和名称。""" # 实际接数据源的搜索接口;这里返回示例结构 result = {"items": [{"code": "SH600XXX", "name": "示例股票"}]} return json.dumps(result, ensure_ascii=False) @mcp.tool() def get_stock_quote(code: str) -> str: """获取股票实时行情快照,包含最新价、涨跌幅、成交量、更新时间。code 形如 SH600XXX。""" data = datasource.get_quote(code) return json.dumps(data, ensure_ascii=False) @mcp.tool() def get_kline(code: str, period: str = "1d", limit: int = 20) -> str: """获取K线数据。period 可选 1m/5m/15m/30m/60m/1d/1w,limit 为返回条数。""" data = datasource.get_kline(code, period=period, limit=limit) return json.dumps(data, ensure_ascii=False) @mcp.tool() def get_market_snapshot(market: str = "SH") -> str: """获取指定市场的行情总览,返回指数和板块概况。market 当前默认 SH。""" # 实际实现拉取市场总览接口 result = {"market": market, "summary": "示例市场数据"} return json.dumps(result, ensure_ascii=False) if __name__ == "__main__": mcp.run(transport="stdio")代码里几个关键点值得展开说:
FastMCP是 MCP Python SDK 提供的高级封装,用@mcp.tool()装饰器就能把普通函数注册成 AI 可调用的工具。函数名就是工具名,docstring 就是工具描述。描述一定要写清楚参数含义和可选值,模型能不能正确调用,很大程度就看这段描述。- 所有返回值统一用 JSON 字符串。这样对模型来说,解析成本最低,而且和 Python、TypeScript 客户端都能兼容。
- 缓存和限流在适配层做,而不是在工具层做。这样即使以后新增一个工具,也自动继承了限流逻辑。
3.3 调试:先用调试面板验证,再连客户端
写完代码先别急着接客户端,第一步是用 MCP 调试面板验证服务端本身是否正常。装好mcp包后,命令行会自带调试工具,在项目目录直接跑:
mcp dev server.py这条命令会启动一个本地调试面板,你可以在里面直接看到服务注册了哪些工具、工具的输入输出结构是什么,还能手动模拟一次工具调用。我在这一步就发现过问题:某个工具函数里抛了异常,但异常信息没有正确返回给客户端,导致调用后一直转圈。直接在调试面板里复现,比在客户端里猜要高效得多。
调试面板确认通过后,再去配置客户端。这里分享一个排查顺序:先服务端,再协议层,最后客户端。很多“工具加载不出来”的问题,根源其实在服务端根本没启动成功,此时无论怎么改客户端配置都没用。
3.4 接入通用 AI 客户端与效果验证
下面是一个标准 MCP 客户端的配置示例,大多数支持 MCP 的桌面 AI 应用都采用类似的 JSON 配置格式:
{ "mcpServers": { "stock": { "command": "python", "args": ["/path/to/stock-sdk-mcp/server.py"], "env": { "STOCK_API_KEY": "your-key-here", "STOCK_API_BASE": "https://api.example-stock-service.com" } } } }配置完成后,重启客户端让它重新加载服务列表。验证路径很简单:在对话框里输入“帮我看看某只股票现在多少钱”。如果配置正常,你会看到 AI 先调用get_stock_quote工具,再把工具返回的数据整理成自然语言回给你。这一条链路跑通,就等于完成了“把股票数据能力接进 AI”的核心目标。
我还做了一个小验证:让 AI 先搜索代码,再查报价,最后拉 30 天 K 线和市场总览,一步问完。这个场景模拟的是“复盘助手”的常见用法。三个工具按顺序被调用,中间不需要任何人工介入,说明工具链路是通的。这也是 MCP 带来的顺手体验——AI 能自己决定调用顺序,而不是靠你写死一段流程。
3.5 让服务常驻运行:进程守护与日志
本地调试跑通只是第一步,如果你希望这个服务能长期给团队用,进程守护是必须的。直接用命令行前台运行肯定不行,窗口一关服务就没了。我用系统服务配置把它注册成常驻进程,崩溃自动拉起,日志写入文件并做轮转。
部署配置里还有几个容易踩的坑:
- 工作目录:客户端启动 MCP 服务时,要对准 server.py 的绝对路径,避免相对路径带来“找不到模块”的困惑。
- 环境变量:在客户端配置的 env 里显式设置 API Key,不要依赖 shell 环境,否则换个机器就莫名报 401。
- 日志级别:MCP 协议在 stdio 传输模式下,服务端如果把业务日志打到标准输出,会污染协议通信,导致客户端解析失败。日志一定要走 stderr,或者直接写文件。这个坑排查起来相当隐蔽,我第一次踩的时候折腾了半小时。
4. 踩坑记录:接入过程中的典型问题与排查思路
4.1 工具一直没出现在对话列表里
这是接入 MCP 时最常见的问题。现象是客户端配置看起来没问题,重启了好几次,但对话框里就是找不到行情工具。排查顺序应该从服务端往客户端推进:
- 先手动执行一次启动命令,确认 server.py 能正常起来、没有语法错误或 import 错误。
- 用 MCP 调试面板加载同一份 server.py,看工具列表是否正常注册。
- 检查客户端配置里的 command 和 args 是否指向同一份代码;环境变量是否完整。
- 最后才考虑客户端本身的重连问题——有些客户端修改配置后需要完全退出再启动,不一定会热加载。
我遇到过一次很典型的案例:配置里的路径写成了相对路径python server.py,但客户端的实际工作目录压根不在项目目录,服务端一直静默启动失败。改成 server.py 的绝对路径后立刻恢复正常。
4.2 AI 回答说“这是昨天的数据”
行情数据出现“日期错位”,听起来像数据源问题,实际上多数是这几种原因:
- 数据源默认返回最近交易日:股票市场有交易日概念,周六用实时报价接口,返回的往往是周五收盘的数据。这不是 bug,是数据源的合理行为。解决办法是让 AI 工具描述里写清楚“若逢非交易时段,返回最近一个交易日收盘数据”,模型自然会在回答里说明这一点。
- 缓存过期时间设置过长:如果你给报价接口设了 60 秒缓存,那么在这 60 秒内的所有请求都会拿到同一份“旧数据”,对话里看起来就像数据没更新。解决办法是区分报价和 K 线的缓存策略,报价短缓存,K 线长缓存。
- 复权问题:拉日 K 时,如果数据源默认返回前复权数据,除权日前后价格会不连续,看起来像“数据错了”。其实只是复权方式不同。工具描述里要明确说明默认的复权口径,让 AI 在回答里也能把这个信息传达给用户。
4.3 请求被数据源限流,服务报 429
行情数据源服务商几乎都有 QPS 或每日请求量的限制。AI 对话场景的特殊性在于:用户可能一句话触发 AI 连续调用 3 个工具,如果人工再追问几句,瞬间请求量就可能翻几倍。被限流的表现通常是服务端报 HTTP 429,随后客户端显示工具调用失败。
我的解法是双管齐下。一方面是适配层加本地缓存,让高频重复请求直接走内存;另一方面是加节流阀,主动控制对外部接口的访问频率,避免请求量超过服务商阈值。千万不要在 429 报错后在代码里写“立即重试 3 次”——重试只会放大流量,数据源会封得更狠。更合理的做法是退避重试:第一次失败等 1 秒,再失败等 2 秒,还是失败就直接把错误信息返回给 AI,让它如实告诉用户“当前行情服务繁忙”。
4.4 长时间无响应和超时问题
MCP 客户端对工具调用有超时设置。如果你的行情接口偶发缓慢,超过客户端默认等待时间,界面就会显示调用超时。这类问题的根因往往不在 MCP 协议本身,而是数据源响应不稳定。
我在适配层里给不同接口设置了不同的超时:报价接口 5 秒,K 线接口 10 秒。这样既不会让模型等太久,也不至于因为一次网络抖动就整体失败。同时把旧的缓存保留一份,当数据源请求失败时,降级返回最近一次成功的缓存并附上时间戳标记,保证 AI 至少有一个可用的返回,而不是拿到一堆错误堆栈。
4.5 常见问题与错误码速查表
整理一张我实际用到的排查表,供你直接参考:
| 现象 | 可能原因 | 排查/解决方向 |
|---|---|---|
| 对话中找不到工具 | 服务端启动失败,或客户端配置路径错误 | 用调试面板验证服务端;检查 args 绝对路径 |
| 调用报错 connection refused | server 进程退出 | 查看 stderr 日志,配置进程守护 |
| 行情数据不更新 | 缓存 TTL 过长,或非交易时段 | 区分报价/K 线缓存策略;工具描述说明交易日概念 |
| 被数据源返回 429 | 请求频率超过限制 | 加本地缓存、节流阀,配合退避重试 |
| 工具调用一直转圈 | 数据源慢,客户端超时 | 缩短外部请求超时;调整客户端 timeout |
| 返回乱码/非 JSON 结构 | 服务端日志污染 stdio 通道 | 日志必须走 stderr,不能写 stdout |
5. 让这套能力继续生长:扩展与后续演进方向
5.1 数据维度扩展:从行情到基本面
跑通四类行情工具之后,自然想到的是扩展数据维度。比如把财务数据(营收、净利润、毛利率)、北向/板块资金流向、公司公告摘要都接进来。但扩展有一个原则:每个工具都要回答一个明确的问题场景。财务数据适合做“基本面快读”,资金流向适合做“异动分析”,公告摘要适合做“消息面跟踪”。没有场景支撑的工具,AI 大概率不会去用。
我实际体验是,每增加一个工具,模型调用准确率都会有轻微下降。所以正确做法是先用两三个工具跑熟,沉淀一套“哪些数据维度对 AI 问答最有价值”的判断标准,再逐步加。如果后续工具数量超过十个,可以考虑拆分服务:行情服务管行情,基本面服务管财务,客户端按需配置。
5.2 多数据源容错:主备切换
单数据源意味着单点故障。如果要让服务在数据源升级、故障时仍然可用,就要设计多数据源容错。实现思路不复杂:适配层里同时对两个数据源做请求,哪一个先成功返回就用哪一个;或者采用主备模式,主源连续失败 N 次就自动切到备用源,并记录切换日志。
这块值得做,但不必一开始就做。前期只有一个数据源时,把缓存和限流做好就已经覆盖了 90% 的问题。等确实出现了数据源不可用的真实事故,再根据监控数据决定要不要接备用源。提前做的最大风险不是代码复杂度,而是你没有真实故障场景去验证切换逻辑是否正确。
5.3 和已有工具链协作:数据不只是被“问答”
股票数据接进 AI 之后,除了即时问答,还能和其他 MCP 工具组合出更多玩法。比如:让 AI 每天定时拉一批自选股报价,生成一张“今日自选摘要”发到内部群;或者让 AI 结合 K 线数据做一段简单的技术形态描述,辅助你自己做判断。
这些场景的本质是“数据工具 + 流程工具”的组合。MCP 的价值正在于此:行情工具是原子能力,AI 客户端是调度器,你可以像搭积木一样把数据能力编排成更复杂的自动化任务。不过要提醒一句:工具组合越复杂,对模型的提示词和工具描述要求越高,建议先在单工具链路上验证结果稳定,再叠加流程。
5.4 协作场景下的权限与审计
如果是团队使用同一套 MCP 服务,就要考虑权限和审计。方案上可以做一个简单的配置文件,指定每个客户端能使用哪些工具子集,再配合访问日志记录每次工具调用的时间、参数和结果摘要。日志的价值在于:当有人问 AI“为什么拿到了某只股票数据”时,你能回溯到具体是哪次调用产生的,而不是面对一个黑盒。
这个环节不要过度设计。小团队阶段,一个日志文件加一个只读原则就够了;等接入的人多了、数据权限复杂了,再引入服务端鉴权也不迟。把精力优先放在数据稳定和覆盖率上,权限边界用最朴素的规则兜住即可。
实际把 stock-sdk-mcp 这套东西从零跑到能稳定使用,我最有体感的几点收获:第一,MCP 接入的真正门槛不在协议本身,而在数据源的质量和稳定性,先把数据层的缓存、限流、超时做扎实,比研究协议细节更有用;第二,工具数量宁少勿多,先让几个核心工具完全跑熟,再慢慢扩展,对模型调用的准确率友好很多;第三,一定要用好调试面板,它能帮你把“服务端问题”和“客户端问题”快速隔离。最后再分享一个小技巧:每次升级 MCP SDK 或改工具定义之后,先跑一遍调试面板里的工具清单,确认旧客户端的配置没有失效,再重启服务,基本就不会遇到工具神秘消失的问题了。