AI Agent 写行情监控?先让它接上 MCP 工具查一次真实数据
作者: TickDB Research · 发布: 2026/6/17 · 阅读: 9
标签: M07, 思否 / A024
AI Agent 写行情脚本前,必须先接上外部行情工具,完成一次可复核的查询。 第一步不是直接生成完整监控系统,而是确认工具可见、查询一个真实 symbol、核对 symbol、last_price、timestamp 等字段。MCP 工具调用是单次查询,不是 WebSocket 持续推送,也不构成投资建议。
你用 Cursor 或 Claude Code 让 AI 写一个 A 股实时行情监控脚本。AI 洋洋洒洒输出了几十行代码,import 了一堆库,还贴心地加上了异常处理。但你一跑就报错——它用了一个根本不存在的 API 端点,或者捏造了 get_stock_price 这样根本不存在的函数。AI 在凭记忆"编造"行情接口,而你让它"写代码"之前,从没让它先看一眼真正的行情工具长什么样。
正确的顺序是:先接工具,再做一次可复核的查询,最后才生成脚本骨架。
一、行情工具长什么样:以 TickDB 为例
在让 AI 动手之前,先理解你要给它接的是什么。以 TickDB 为例,它是一个统一行情数据平台,把 A 股、港股、美股、期货、数字货币等市场的实时行情整合到一个入口里,然后通过多种方式向外提供——其中就包括专门给 AI 编程工具用的 MCP 接口。
TickDB 给 AI Agent 提供了三种接入形态,对应三种使用场景:
| 接入形态 | 适合的场景 | 能力边界 |
|---|---|---|
| MCP 工具调用 | 在 Cursor、Claude Code 等 AI IDE 里直接查行情 | 单次查询,AI 按需调用;不是持续推送 |
| REST API | 写后端服务、自动化脚本、定时任务 | 一次 HTTP 请求拿一次快照或历史数据 |
| WebSocket 推送 | 监控看板、实时告警、需要持续接收变化的场景 | 长连接持续推送,断线需自行处理重连 |
这三种形态各有分工,不能混用。本文聚焦的是第一种——MCP 工具调用,因为它的场景最特殊:你不是在写一个独立应用,而是在让 AI 帮你写应用,而 AI 需要先"看见"真实的行情工具,才知道代码该长什么样。
TickDB 的 MCP 工具通过远端 HTTP 端点接入,配置 Header 为 X-TickDB-Key,官方远端地址为 https://mcp.tickdb.ai/。在 Cursor 或 Claude Code 中配置完成后,AI 就能在对话中直接调用 get_ticker、get_kline 等工具。MCP 接口的意义在于:让 AI 在生成代码之前,先完成一次真实的工具调用,而不是靠训练记忆去猜测某个行情接口的返回格式。
本文以 TickDB 作为示例。如果你用的是其他数据源,原则是一样的——确认 AI 环境里有可用的工具,做一次可复核查询,再进入生成环节。
二、流程:先查后写
让 AI Agent 对接实时行情,按下面这五步走,每一步验证上一步的结果,不要跳:
用户任务:监控某只 A 股当前价格
│
▼
① 工具可见:确认 AI 环境已加载 TickDB MCP 行情工具
│
▼
② 单次查询:让 Agent 调 get_ticker 查一个真实 symbol
│
▼
③ 字段核对:确认返回的 symbol、last_price、timestamp 存在且合法
│
▼
④ 脚本骨架:基于已验证的工具调用方式生成监控脚本框架
│
▼
⑤ 扩展与告警:加入多 symbol 循环、失败分支与结果汇总
每一步的输出,都是下一步的输入。跳步的直接后果:AI 在没验证工具存在的情况下就开始"编"代码,后续所有工作都建立在虚构的 API 之上。
三、第一步:确保 AI 能看见行情工具
在 Cursor 或 Claude Code 中配置好 TickDB MCP 之后,AI 就能在对话中直接调用 get_ticker 等工具。你可以用下面这个自然语言任务模板,直接复制给你的 AI Agent:
你现在可以访问 TickDB MCP 的行情工具。首先,列出当前可用的工具名称和描述,确认 get_ticker 已存在。
然后,用 get_ticker 查询 600519.SH,type=stock。
如果调用成功,按顺序回答:
1. 返回的 symbol 是否与请求一致?
2. last_price 是否存在且为可用数值?
3. timestamp 是否存在且为正整数?
不要补充建议,不要发散解释,只回答这三个问题的结果。
为什么用 600519.SH 做首次查询? 它是一个 A 股常见品种,在 2026-06-14 的 TickDB MCP 测试中查询成功,返回
code=0,data 中包含该 symbol,且字段symbol、type、last_price、timestamp均可见。这个已知的可复核结果,是你判断工具是否正常工作的基准。
四、第二步:确认查询返回的关键字段
当 AI 返回查询结果后,不要直接让它写监控脚本。先核对以下三个字段:
| 核对项 | 为什么重要 | 如果缺失 |
|---|---|---|
symbol 与请求一致 | 防止返回了错误品种的价格 | 后续监控盯错了对象 |
last_price 存在且为合法数值 | 这是价格监控的核心字段 | 监控没有意义 |
timestamp 为正整数 | 标记行情时间,判断数据新旧 | 无法判断行情时效 |
这三个字段全部确认后,才进入第三步。任何一个缺失,都意味着工具调用不正常,需要回头检查配置或网络。
注意:MCP 工具调用是单次查询,不是持续推送。一次
get_ticker成功,只证明当前这次查询可用——不等于后续每次都会成功,也不等于数据是实时推送的。你需要把错误处理写进脚本骨架。
五、第三步:生成监控脚本骨架
确认工具可用、字段正常后,再让 AI 生成脚本。这次它就不是凭空编造,而是基于已验证的工具调用方式去写。
下面是一个可参考的自然语言任务模板:
基于刚才 get_ticker 的成功调用,帮我生成一个 Python 监控脚本的框架,要求:
1. 使用 dict 定义待监控的 symbol 列表:["600519.SH", "000001.SZ"]
2. 用循环逐个调用 get_ticker
3. 对每个返回结果检查:symbol 是否匹配、last_price 是否存在且可转为数值、timestamp 是否为正整数
4. 成功的结果放入 results 列表,失败的结果放入 errors 列表并记录 symbol 和失败原因
5. 最后输出一个状态表,包含 symbol、状态(成功/失败)、last_price(仅成功时显示)
不要写假的 API 调用,直接用 get_ticker 的真实调用方式。
AI 生成的脚本骨架应该大致是这个结构(伪代码,只展示逻辑流程,不是生产级代码):
# 监控脚本骨架(伪代码,不可直接运行,非生产级代码)
# 基于 TickDB MCP get_ticker 工具的实际调用方式生成
from decimal import Decimal, InvalidOperation
SYMBOLS = ["600519.SH", "000001.SZ"]
results = []
errors = []
for sym in SYMBOLS:
try:
# 调用 get_ticker 工具,此处为 MCP 工具调用
resp = call_tool("get_ticker", symbols=sym, type="stock")
# 检查返回码
if resp.get("code") != 0:
errors.append({"symbol": sym, "reason": f"业务码异常: {resp.get('code')}"})
continue
# 提取 data 第一条
items = resp.get("data", [])
if not items:
errors.append({"symbol": sym, "reason": "data 为空"})
continue
item = items[0]
if item.get("symbol") != sym:
errors.append({"symbol": sym, "reason": "symbol 不匹配"})
continue
# last_price 校验:必须存在、可转为有限 Decimal
price_raw = item.get("last_price")
if not isinstance(price_raw, str) or not price_raw.strip():
errors.append({"symbol": sym, "reason": "last_price 缺失或为空"})
continue
try:
price = Decimal(price_raw)
except (InvalidOperation, ValueError):
errors.append({"symbol": sym, "reason": f"last_price 无法解析: {price_raw}"})
continue
if not price.is_finite():
errors.append({"symbol": sym, "reason": f"last_price 非有限数: {price_raw}"})
continue
# timestamp 校验:必须为正整数且非 bool
ts = item.get("timestamp")
if isinstance(ts, bool) or not isinstance(ts, int) or ts <= 0:
errors.append({"symbol": sym, "reason": "timestamp 无效"})
continue
results.append({"symbol": sym, "last_price": str(price), "timestamp": ts})
except Exception as e:
errors.append({"symbol": sym, "reason": str(e)})
# 输出状态表
print("=" * 40)
print(f"{'symbol':<15} {'status':<10} {'last_price':<12}")
print("-" * 40)
for r in results:
print(f"{r['symbol']:<15} {'成功':<10} {r['last_price']:<12}")
for e in errors:
print(f"{e['symbol']:<15} {'失败':<10} {e['reason']:<12}")
print("=" * 40)
这是骨架,不是生产级代码。 你需要补充:日志记录、重试逻辑、告警阈值、持久化存储和错误通知。骨架的任务是让你确认 AI 理解了工具的正确调用方式,而不是直接把输出部署上线。
六、检查卡:从工具可见到脚本骨架
每完成一步,对照这张表打钩:
- [ ] 工具可见:AI 能列出可用工具,
get_ticker在列表中 - [ ] 单次查询成功:用
600519.SH查询,返回code=0,data非空 - [ ] symbol 匹配:返回数据中的
symbol与请求一致 - [ ]
last_price存在:字段存在且可解析为有限 Decimal - [ ]
timestamp存在:字段为正整数且非 bool - [ ] 脚本骨架生成:基于上述验证结果,AI 生成了循环查询和状态表逻辑
- [ ] 失败分支:脚本包含了 symbol 不匹配、data 为空、字段缺失、数值解析失败等异常处理
七、适合谁,不适合谁
适合的场景:
- AI 工具链首次接入行情数据,需要验证工具可用性和返回结构
- 快速生成监控脚本骨架,后续手动补充生产逻辑
- 按需查询场景:每日定时任务查询一批 symbol,汇总状态
不适合的场景:
- WebSocket 持续推送——MCP
get_ticker是单次查询,不是流式推送 - 生产级低延迟监控——本文只验证字段可读性,不承诺延迟或稳定性
- 交易执行——本文不涉及下单、撤单、仓位管理
八、下一步
你已验证了工具可用、字段结构正确、脚本骨架能生成。接下来该做什么?
- 用你自己的 symbol 列表跑一次完整循环,确认每个品种都能正常返回。
- 补充你需要的业务逻辑:价格变动阈值告警、存储到数据库、定时触发。
- 查阅 TickDB 官方文档,了解 MCP 工具完整列表以及 REST 和 WebSocket 接入方式的适用场景——单次查询用 MCP 或 REST,需要持续推送时再考虑 WebSocket。
别再让 AI 凭记忆写行情代码。先接上工具,查一个真实 symbol,核对三个关键字段,再生成脚本骨架。 你的 AI 助手能列出当前可用的行情工具吗?在评论区晒出它第一次 get_ticker 的返回摘要——symbol、last_price 是否有值、timestamp 是否合理——我们看看它走对了第几步。
本文仅讨论 AI 工具链的工程接入方法,不构成任何投资建议。所有价格字段均为接口返回值占位说明,不代表实时报价。
通过 TickDB API 获取实时行情数据
一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送,免费开始使用。
免费领取 API Key查看 API 文档