AI Agent 接行情的第一个坑:没查工具就写代码
一、问题边界:一次排障的记录
同事在 Cursor 里让 AI 写一个 A 股实时行情监控脚本。AI 输出了几十行代码,import 了一堆库,异常处理也加了。一跑就报错——它用了一个不存在的 API 端点,还捏造了 get_stock_price 这个函数。
排查只花了五分钟:AI 在生成代码之前,没有被明确告知可用的行情工具是什么,也没有做过一次真实的工具调用验证。它不知道真正的接口长什么样,只能凭训练记忆去猜。猜错了。
这暴露的不是 AI 的能力问题,而是一个被反复跳过的工程步骤:在让 AI 生成任何依赖外部数据的代码之前,必须先确认数据入口真实存在、能调通、返回结构符合预期。 跳了这一步,后面所有的代码生成、异常处理、告警逻辑都建立在不可靠的基础上。
本文只覆盖这个过程的第一步——确认工具可用、完成一次可复核的查询、核对关键字段。后续的持续监控、WebSocket 推送、生产级部署不在范围内。这一步跑通了,脚本骨架才是长在土里的,不是飘在天上的。
二、接入边界:AI 不是行情源,工具入口才是
AI Agent 不能凭空知道当前价格。训练记忆是历史快照,不是实时数据。让它获取当前行情的唯一可靠方式,是接上一个真实的外部工具。
TickDB 的 MCP 接口就是为这个场景设计的——把行情查询封装成 AI 编程工具可调用的函数。在 Cursor 或 Claude Code 中配置完成后,AI 环境中会多出一组可被 Agent 直接调用的行情工具,包括 get_ticker、get_kline 等。这些不是 AI 自己编造的函数名,而是指向真实行情服务的入口。
配置方式:通过远端 HTTP 端点接入,Header 为 X-TickDB-Key,官方远端地址 https://mcp.tickdb.ai/。配置成功后,AI 对话中就能直接调用行情工具。
MCP 工具调用是单次查询,不是持续推送。 一次
get_ticker成功只证明当前这次查询可用,不等于后续每次都会成功,也不等于数据是实时推送的。需要持续监控的场景,应另行评估 WebSocket 方案。
三、查询边界:用真实 symbol 做一次最小调用
工具接入后,第一步不是写脚本,是做一次最小查询——用 get_ticker 查一个真实品种,验证整个链路是通的。
选 600519.SH 作为首次查询对象,是因为它常见,且在已有的 TickDB MCP 测试中查询成功,返回 code=0,symbol、type、last_price、timestamp 等字段均可见。这个已知可复核的结果,是你判断工具是否正常工作的基准。
把下面这个自然语言任务模板直接复制给 AI Agent:
你现在可以访问 TickDB MCP 的行情工具。首先,列出当前可用的工具名称和描述,确认 get_ticker 已存在。
然后,用 get_ticker 查询 600519.SH。
如果调用成功,按顺序回答:
1. 返回的 symbol 是否与请求一致?
2. last_price 是否存在且为可用数值?
3. timestamp 是否存在且为正整数?
不要补充建议,不要发散解释,只回答这三个问题的结果。
这个模板刻意限制了 AI 的回答范围——三个问题,不多不少。目的是把首次查询变成一个精确的、可复核的检查点,而不是让 AI 拿着一知半解的成功结果开始发挥。
四、契约边界:三个字段与失败分支
查询返回后,先核对三个关键字段,再进入代码生成环节:
| 核对项 | 为什么重要 | 如果缺失 |
|---|---|---|
symbol 与请求一致 |
防止返回了错误品种的价格 | 后续监控盯错了对象 |
last_price 存在且为合法数值 |
价格监控的核心字段 | 监控没有意义 |
timestamp 为正整数 |
标记行情时间,判断数据新旧 | 无法判断行情时效 |
三个字段全部确认后,再让 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 生成的脚本骨架应包含以下核心逻辑。任何校验失败都进 errors 列表,不补默认值,不静默跳过——价格缺失就是缺失,timestamp 非法就是非法,不能让不可信数据流入下游。
# 监控脚本骨架——基于 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)
# 检查返回码
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存在:字段存在且可解析为有限 Decimaltimestamp存在:字段为正整数且非 bool- 脚本骨架生成:基于验证结果生成循环查询和状态表逻辑
- 失败分支:包含 symbol 不匹配、data 为空、字段缺失、数值解析失败等异常处理
六、能做什么 / 不能做什么
本文覆盖的范围:
- AI 工具链首次接入行情数据,验证工具可用性和返回结构
- 用最少步骤跑通一次
get_ticker查询并核对关键字段 - 基于已验证的工具调用方式生成监控脚本骨架
本文不覆盖的范围:
- WebSocket 持续推送——MCP
get_ticker是单次查询,不是流式推送。持续监控场景应另行评估 WebSocket 方案 - 生产级低延迟监控——本文只验证字段可读性和工具可用性,不验证延迟、稳定性或 SLA
- 交易执行——本文不涉及下单、撤单、仓位管理
- REST 或 WebSocket 的接入方式——本文只讨论 MCP 工具调用这一种入口
七、下一步
工具调通了,字段核对了,骨架生成了。接下来三件事:
- 用自己的 symbol 列表跑一次完整循环,不要只依赖一个
600519.SH的验证结果——品种不同,返回行为可能不同。 - 补充业务逻辑:阈值告警、数据库写入、定时触发——这些是骨架之外的业务层,需要你自己设计。
- 评估其他接入方式:单次查询 MCP 足够,需要持续推送时再评估 WebSocket。TickDB 的官方文档提供了完整工具列表和多入口配置说明,可搜索查阅。
更多推荐



所有评论(0)