✨解锁 AI Agent 新姿势！手把手教你用 Python 搭建 MCP 服务，对接沪深数据 API，量化交易MCP 服务 (保姆级教程)✨-开发者社区

1. 为什么我要给 AI Agent 接一个行情 MCP 服务

先说清楚这篇要解决的事：用 Python 从零写一个 MCP（Model Context Protocol）服务，把沪深行情数据 API 包装成 AI Agent 能直接调用的工具，然后在 Cline 里发起一次真实的行情查询。适合谁？适合已经会用 Python 写脚本、想让 Cursor / Cline 这类 AI 客户端“自己动手查数据”的开发者，尤其是做量化、盯盘工具、投研助手方向的朋友。

MCP 是什么？你可以把它理解成 AI 和外部世界之间的“标准插座”。AI 客户端（比如 Cline）不会直接去访问某个行情 API，它只会按 JSON-RPC 2.0 的格式，通过标准输入输出（stdin/stdout）跟你的 MCP 服务对话。你的服务收到请求后，去调真正的沪深数据 API，把结果打包成 JSON 再吐回去。整个过程里，AI 负责“想调哪个工具”，你的 Python 脚本负责“怎么调、调完怎么回”。

我试过把这套链路跑通之后，最直观的变化是：以前要手动复制股票代码、粘贴到网页查、再把结果贴回对话框；现在直接在 Cline 里说一句“帮我查一下 000001 的公司简介”，Agent 会自己选中get_company_profile这个工具、传参、拿结果、组织成人话回给我。下面按“环境准备 → 写服务 → 配置客户端 → 验证 → 排障”的顺序走一遍，代码可以直接复制。

2. TaoToken 前置：统一 Key 与 API 通道

在写 MCP 服务之前，先把“模型侧”的通道理顺。MCP 服务本身只负责行情数据，但你的 AI Agent 要能跑起来，背后得有一个稳定的大模型调用入口。TaoToken 在这里的角色就是统一 Key 和 API 通道：你不需要在 Cline、Cursor、脚本里各维护一套配置，而是通过一个 Key 走同一个 API 入口。

官网入口：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

API 地址（不带 UTM）：https://taotoken.net/api

具体操作上，你需要先拿到一个可用的 Key。进入控制台创建 API Key：

控制台：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite
API Keys 管理：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite

拿到 Key 之后，建议先做一次模型对话验证，确认通道是通的，再去折腾 MCP：

模型对话：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite

如果你后面要长期跑编码类 Agent（比如让 Cline 持续调用工具、多轮改代码），可以看下 Coding Plan，它更适合高频、长会话的场景：

Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite

接入文档在这里，配置项、参数含义都以它为准：

接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite

注意：MCP 服务负责的是“行情数据工具”，TaoToken 负责的是“模型调用通道”，两者是配合关系，不要混在一起配。先把模型通道跑通，再写 MCP，排障时能少一半干扰。

3. 可复制配置：Python MCP 服务骨架

3.1 环境与依赖

Python 建议 3.8 以上，核心依赖只有requests：

pip install requests

目录结构建议这样，方便后面配 Cline：

mcp-quant/ ├── mcp_hs_server.py └── requirements.txt

3.2 服务端完整骨架

下面这份代码是可直接运行的 MCP 服务骨架。它做了三件事：监听 stdin 的 JSON-RPC 请求、把工具名映射到 Python 函数、调用沪深数据 API 并回写 stdout。行情 API 的 Token 请替换成你自己的，示例里用占位符。

#!/usr/bin/env python # -*- coding: utf-8 -*- import sys import json import logging import traceback import requests # 日志走 stderr，避免污染 stdout 的 JSON-RPC 响应 logging.basicConfig( level=logging.INFO, stream=sys.stderr, format="%(asctime)s - %(levelname)s - %(message)s", ) HS_API_BASE = "https://api.example-hs.com/hs" HS_TOKEN = "YOUR_HS_API_TOKEN" # 替换为你自己的行情 API Token if HS_TOKEN == "YOUR_HS_API_TOKEN": logging.warning("当前使用的是占位 Token，请替换为有效 Token 后再调用。") def call_hs_api(endpoint_path, params=None): """通用行情 API 调用，返回解析后的 JSON。""" url = f"{HS_API_BASE}/{endpoint_path}?token={HS_TOKEN}" logging.info(f"准备调用行情 API: {url}") try: resp = requests.get(url, params=params, timeout=20) resp.raise_for_status() return resp.json() except requests.exceptions.Timeout: raise TimeoutError(f"请求行情 API 超时: {url}") except requests.exceptions.RequestException as e: raise ConnectionError(f"行情 API 请求失败: {e}") except json.JSONDecodeError: raise ValueError("行情 API 返回的不是合法 JSON") def get_stock_list(**kwargs): """获取沪深 A 股股票列表（代码、名称、交易所）。""" logging.info("执行工具: get_stock_list") return call_hs_api("list/all") def get_company_profile(stock_code, **kwargs): """获取指定股票代码的上市公司简介。 Args: stock_code (str): 股票代码，例如 '000001'。 """ if not stock_code: raise ValueError("工具 get_company_profile 需要 stock_code 参数。") logging.info(f"执行工具: get_company_profile, stock_code={stock_code}") return call_hs_api(f"gs/gsjj/{stock_code}") def get_capital_trend(stock_code, **kwargs): """获取指定股票代码的每日资金流入趋势。 Args: stock_code (str): 股票代码，例如 '000001'。 """ if not stock_code: raise ValueError("工具 get_capital_trend 需要 stock_code 参数。") logging.info(f"执行工具: get_capital_trend, stock_code={stock_code}") return call_hs_api(f"capital/lrqs/{stock_code}") TOOLS = { "get_stock_list": get_stock_list, "get_company_profile": get_company_profile, "get_capital_trend": get_capital_trend, } logging.info(f"已注册 MCP 工具: {list(TOOLS.keys())}") def handle_request(request_data): request_id = request_data.get("id") if request_data.get("jsonrpc") != "2.0" or "method" not in request_data: return json.dumps({ "jsonrpc": "2.0", "error": {"code": -32600, "message": "无效请求"}, "id": request_id, }) method = request_data.get("method") params = request_data.get("params", {}) payload = None if method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) logging.info(f"工具调用 -> {tool_name}, 参数: {arguments}") if tool_name in TOOLS: try: result = TOOLS[tool_name](**arguments) payload = {"result": json.dumps(result, ensure_ascii=False)} except (ValueError, TypeError) as e: payload = {"error": {"code": -32602, "message": f"参数错误: {e}"}} except (ConnectionError, TimeoutError, ValueError) as e: payload = {"error": {"code": -32000, "message": f"外部 API 失败: {e}"}} except Exception as e: logging.error(traceback.format_exc()) payload = {"error": {"code": -32000, "message": f"内部错误: {e}"}} else: payload = {"error": {"code": -32601, "message": f"工具不存在: {tool_name}"}} elif method == "tools/list": tool_list = [] for name, func in TOOLS.items(): doc = func.__doc__ or "无描述" tool_list.append({ "name": name, "description": doc.split("Args:")[0].strip(), }) payload = {"result": json.dumps(tool_list, ensure_ascii=False)} else: payload = {"error": {"code": -32601, "message": f"方法未找到: {method}"}} final = {"jsonrpc": "2.0", "id": request_id} final.update(payload) return json.dumps(final, ensure_ascii=False) def main(): logging.info("MCP 服务已启动，等待 stdin 请求...") while True: try: line = sys.stdin.readline() if not line: logging.info("stdin 关闭，服务退出。") break line = line.strip() if not line: continue try: request_data = json.loads(line) except json.JSONDecodeError: logging.error(f"无法解析 JSON: {line}") continue response = handle_request(request_data) print(response, flush=True) except KeyboardInterrupt: logging.info("收到 Ctrl+C，退出。") break except BrokenPipeError: logging.warning("管道断开，退出。") break except Exception: logging.error(traceback.format_exc()) continue if __name__ == "__main__": main()

几个关键点解释一下。日志全部走stderr，因为stdout是留给 JSON-RPC 响应的，混在一起客户端会解析失败。tools/call里result字段必须是字符串，所以用json.dumps把 API 返回的 dict 序列化，ensure_ascii=False保证中文不乱码。print(..., flush=True)的flush很关键，不刷新缓冲区客户端会一直等。

3.3 Cline 的 settings.json 配置

Cline 里配置 MCP Server，核心是告诉它“用什么命令启动这个脚本”。在 Cline 的 MCP 设置里新增一个 Server，或者直接编辑它的settings.json：

{ "mcpServers": { "hs-quant": { "command": "python", "args": ["/absolute/path/to/mcp-quant/mcp_hs_server.py"], "env": { "PYTHONUNBUFFERED": "1" } } } }

args里必须是脚本的绝对路径，相对路径在客户端启动子进程时经常找不到。Windows 下路径写成C:\\path\\to\\mcp_hs_server.py，或者用原始字符串。PYTHONUNBUFFERED=1是保险起见，避免输出被缓冲。

3.4 config.toml 形式的等价配置

有些客户端（或你自己的启动器）习惯用 TOML，等价写法如下：

[mcp_servers.hs-quant] command = "python" args = ["/absolute/path/to/mcp-quant/mcp_hs_server.py"] [mcp_servers.hs-quant.env] PYTHONUNBUFFERED = "1"

两种格式表达的是同一件事：启动命令 + 参数 + 环境变量。选你客户端支持的那种即可。

4. 验证请求：用 Cline 发起一次行情查询

配置保存后，重启 Cline 或重新加载 MCP Server。你可以在 MCP 管理界面看到hs-quant的状态，正常应该是已连接。如果显示红色或报错，先看第 5 节的排障。

4.1 先手动验证服务本身

在配客户端之前，建议先在终端手动喂一条 JSON 给服务，确认它能正常响应：

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' | python mcp_hs_server.py

预期输出是一行 JSON，result里是工具列表的字符串，包含get_stock_list、get_company_profile、get_capital_trend。如果这步就失败，问题在服务本身，跟客户端无关。

再测一次工具调用：

echo '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"get_company_profile","arguments":{"stock_code":"000001"}}}' | python mcp_hs_server.py

成功的话，result字段里是该公司简介的 JSON 字符串。注意这里返回的是字符串，不是嵌套对象，这是 MCP 规范要求的。

4.2 在 Cline 里发起真实查询

服务验证通过后，回到 Cline 对话框，直接说：

帮我查一下 000001 的公司简介，用 hs-quant 这个 MCP 工具。

Cline 会先调用tools/list拿到可用工具，然后选中get_company_profile，传入stock_code: "000001"，拿到结果后组织成自然语言回复你。你可以在 Cline 的工具调用日志里看到完整的请求和响应，确认参数传对了、API 返回了数据。

再试一个带趋势的：

看看 000001 最近的资金流入趋势。

这次它会调get_capital_trend。如果两个工具都能正常返回，说明整条链路——Cline → MCP 服务 → 行情 API → 回传——已经打通。

5. 本篇常见错排查

5.1 客户端显示 MCP Server 启动失败

最常见的原因是args里的路径不对。Cline 启动子进程时的工作目录不一定是你项目目录，所以必须用绝对路径。另一个原因是command写成了python3但系统里只有python，或者反过来。先在终端用同样的命令手动跑一遍，能跑通再填进配置。

5.2 服务启动了但工具列表为空

检查TOOLS字典是否在main()之前就完成了注册。如果工具函数定义在TOOLS之后，字典里会是空的。另外确认tools/list分支里result是字符串，有些客户端对类型敏感，返回对象会直接忽略。

5.3 调用工具报“参数错误”

MCP 客户端传参时，arguments是一个字典。如果你的函数签名是def get_company_profile(stock_code, **kwargs)，那arguments里必须有stock_code键。如果 AI 传的是code或symbol，就会触发ValueError。解决办法是在 docstring 里把参数名写清楚，AI 会参考描述来传参。

5.4 中文返回乱码

两个地方要检查。一是json.dumps有没有加ensure_ascii=False，不加的话中文会变成\uXXXX。二是print的输出编码，Windows 终端默认可能是 GBK，建议在脚本开头加：

import io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

5.5 请求超时或连接失败

行情 API 的 Token 无效、过期、或者调用频率超限，都会表现为超时或 4xx。先在浏览器或 curl 里直接请求一次 API，确认 Token 本身可用。另外timeout=20可以根据你的网络情况调整，但不要设得太短，行情接口偶尔会慢。

5.6 stdout 被日志污染

如果你看到客户端报“无法解析 JSON”，大概率是某行日志打到了stdout。检查所有print和logging的输出目标，logging.basicConfig里必须指定stream=sys.stderr。第三方库如果自己往 stdout 打印，也会造成同样的问题，必要时重定向。

6. 把工具侧接入收个尾

到这里，你的 Python MCP 服务已经能跑、能配、能验证了。行情数据工具走的是你自己的 API Token，模型调用通道走 TaoToken 统一 Key，两边各司其职。如果你在接入过程中遇到 Key 配置、通道报错、或者想让 Agent 长期跑编码任务，可以按场景分流处理：

排障与接入细节，先看 API Keys 和接入文档：
- https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite
- https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
想先验证模型通道是否正常，用模型对话：
- https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite
长期编码、Agent 多轮工具调用，看 Coding Plan：
- https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite

最后留一个实操建议：把HS_TOKEN从代码里挪到环境变量，用os.environ.get("HS_API_TOKEN")读取，脚本里只留占位。这样你把代码分享出去或者提交到仓库时，不会把 Token 一起带出去。MCP 服务本身不复杂，难的是把每个环节的边界划清楚——谁负责数据、谁负责模型、谁负责调度，分清楚了，后面加工具就是复制粘贴的事。