最近在整理一个挺实在的需求把股票数据能力接进 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 客户端生态已经形成了“工具即插即用”的交互习惯。与其自己维护一套私有协议不如直接使用业界已经统一的标准接入方式。MCPModel 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.10MCP 官方 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 fquote:{code} cached self._get_cached(cache_key, ttl2) if cached: return cached self._throttle() url f{self.base_url}/quote?code{code}api_key{self.api_key} with request.urlopen(url, timeout5) as resp: data json.loads(resp.read()) self._set_cache(cache_key, data, ttl2) return data def get_kline(self, code: str, period: str 1d, limit: int 20) - dict: K 线数据按代码周期做 30 秒缓存 cache_key fkline:{code}:{period}:{limit} cached self._get_cached(cache_key, ttl30) 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, timeout10) as resp: data json.loads(resp.read()) self._set_cache(cache_key, data, ttl30) 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_keySTOCK_API_KEY, base_urlSTOCK_API_BASE, max_qps4, ) mcp.tool() def search_stock_code(keyword: str) - str: 按股票名称或代码片段搜索返回匹配列表。例如输入“白酒”或“600”返回相关股票代码和名称。 # 实际接数据源的搜索接口这里返回示例结构 result {items: [{code: SH600XXX, name: 示例股票}]} return json.dumps(result, ensure_asciiFalse) mcp.tool() def get_stock_quote(code: str) - str: 获取股票实时行情快照包含最新价、涨跌幅、成交量、更新时间。code 形如 SH600XXX。 data datasource.get_quote(code) return json.dumps(data, ensure_asciiFalse) mcp.tool() def get_kline(code: str, period: str 1d, limit: int 20) - str: 获取K线数据。period 可选 1m/5m/15m/30m/60m/1d/1wlimit 为返回条数。 data datasource.get_kline(code, periodperiod, limitlimit) return json.dumps(data, ensure_asciiFalse) mcp.tool() def get_market_snapshot(market: str SH) - str: 获取指定市场的行情总览返回指数和板块概况。market 当前默认 SH。 # 实际实现拉取市场总览接口 result {market: market, summary: 示例市场数据} return json.dumps(result, ensure_asciiFalse) if __name__ __main__: mcp.run(transportstdio)代码里几个关键点值得展开说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 refusedserver 进程退出查看 stderr 日志配置进程守护行情数据不更新缓存 TTL 过长或非交易时段区分报价/K 线缓存策略工具描述说明交易日概念被数据源返回 429请求频率超过限制加本地缓存、节流阀配合退避重试工具调用一直转圈数据源慢客户端超时缩短外部请求超时调整客户端 timeout返回乱码/非 JSON 结构服务端日志污染 stdio 通道日志必须走 stderr不能写 stdout5. 让这套能力继续生长扩展与后续演进方向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 或改工具定义之后先跑一遍调试面板里的工具清单确认旧客户端的配置没有失效再重启服务基本就不会遇到工具神秘消失的问题了。