1. 为什么我要用 MCP 给 LLM 接上投资数据源先说清楚 MCP 是什么、能做什么、适合谁。MCPModel Context Protocol是一套让大模型与外部工具、数据源对话的开放协议你可以把它理解成「给 LLM 装了一个标准化的 USB 接口」模型负责推理和表达MCP Server 负责去拿真实数据、执行真实动作。适合谁适合手里有 Python 基础、想让大模型从「聊天玩具」变成「能干活的助手」的开发者尤其是做金融、投研、个人理财这类需要实时数据的场景。我试过把行情接口、持仓表、资讯源直接塞进 prompt结果上下文又长又容易过期模型还会一本正经地编数字。MCP 的价值就在于把「取数」和「推理」拆开数据由工具函数实时拉取模型只负责基于真实返回值组织语言。这篇就带你从零跑通一个可对话的 AI 投资顾问原型包含行情查询、持仓分析两类工具全程可复制。核心检索词先摆出来MCP 协议接入个人投资数据源、Python 实现 MCP Server、LLM 投资顾问原型。整条链路是「MCP Client对话入口→ MCP Server工具层→ 数据源/LLM」你只要把工具函数写对模型就能调用。下面这张表先帮你建立整体认知后面每一步都会展开角色职责本文对应文件MCP Client发起对话、转发工具调用、渲染结果advisor_client.pyMCP Server注册工具、执行取数逻辑、返回结构化结果advisor_server.py数据源行情接口 / 本地持仓 CSVholdings.csv 行情 APILLM基于工具返回值生成建议通过 TaoToken 统一接入很多人卡在第一步以为要自己实现协议解析。其实用官方 Python SDK注册工具就是加个装饰器的事。真正需要你花心思的是「工具函数怎么设计」——参数要少而明确返回要结构化错误要可读这样模型才不会乱调。2. TaoToken 前置把模型入口和 Key 准备好在写工具之前先把 LLM 这一端打通。MCP Server 本身不产生智能它需要调用一个大模型来完成「根据持仓生成分析」这类动作。我用 TaoToken 作为统一入口原因是它兼容 OpenAI 风格的调用方式改一个base_url就能切换模型不用为每个供应商重写客户端。你需要准备三样东西我把它叫做「三件套」后面所有配置都围绕它Base URLhttps://taotoken.net/apiAPI Key在控制台创建形如sk-开头Model ID比如claude-sonnet-4-5或gpt-4o-mini按你的预算和任务选获取路径很直接先访问官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册然后进控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建 API Key具体入口在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。如果你只是想先验证模型通不通可以直接用模型对话页 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 试一句。把 Key 写进环境变量别硬编码进代码export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意环境变量名建议统一前缀避免和系统里其他API_KEY冲突。团队协作时把.env加进.gitignore。如果你后续要做长期编码或 Agent 类项目可以了解 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它更适合高频调用场景。接入细节可以对照文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。这里有个容易忽略的点MCP Server 调用 LLM 时超时和重试要自己兜底。模型偶尔会慢工具函数里必须包try/except否则一次网络抖动就会让整个对话崩掉。我一般设 30 秒超时、最多重试 2 次。3. 可复制配置MCP Server 与工具函数注册这一节是全文核心给你能直接跑的代码。先装依赖pip install mcp[cli] openai pandas3.1 项目结构与配置文件目录长这样ai-advisor/ ├── advisor_server.py ├── advisor_client.py ├── holdings.csv └── mcp.jsonmcp.json是客户端连接配置路径和字段要和实际一致{ mcpServers: { ai-advisor: { command: python, args: [advisor_server.py], env: { TAOTOKEN_API_KEY: sk-你的key, TAOTOKEN_BASE_URL: https://taotoken.net/api, MODEL_ID: claude-sonnet-4-5 } } } }提示command用绝对路径更稳比如/usr/bin/python3避免虚拟环境切换后找不到解释器。3.2 行情查询工具先写一个最简的行情工具返回结构化 JSON方便模型解析import os import json import httpx from mcp.server.fastmcp import FastMCP mcp FastMCP(ai-advisor) MOCK_QUOTES { 600519: {name: 贵州茅台, price: 1680.5, change_pct: 1.23}, 000001: {name: 平安银行, price: 11.42, change_pct: -0.87}, 510300: {name: 沪深300ETF, price: 3.98, change_pct: 0.45}, } mcp.tool() def get_quote(symbol: str) - str: 查询单只标的的实时行情。symbol 为 6 位代码如 600519。 data MOCK_QUOTES.get(symbol) if not data: return json.dumps({error: f未找到代码 {symbol} 的行情}, ensure_asciiFalse) return json.dumps({symbol: symbol, **data}, ensure_asciiFalse)真实项目里把MOCK_QUOTES换成你的行情 API 调用即可注意加超时和异常捕获。返回用json.dumps(..., ensure_asciiFalse)中文才不会变成转义码。3.3 持仓分析工具持仓从 CSV 读工具负责算总市值和盈亏import csv mcp.tool() def analyze_holdings(csv_path: str holdings.csv) - str: 读取持仓 CSV计算每只标的市值与总盈亏。CSV 列symbol,cost,shares rows [] total_cost 0.0 total_value 0.0 with open(csv_path, newline, encodingutf-8) as f: for row in csv.DictReader(f): symbol row[symbol].strip() cost float(row[cost]) shares float(row[shares]) quote MOCK_QUOTES.get(symbol, {}) price quote.get(price, cost) value price * shares total_cost cost * shares total_value value rows.append({ symbol: symbol, name: quote.get(name, 未知), cost: cost, shares: shares, price: price, pnl: round(value - cost * shares, 2), }) summary { positions: rows, total_cost: round(total_cost, 2), total_value: round(total_value, 2), total_pnl: round(total_value - total_cost, 2), pnl_pct: round((total_value - total_cost) / total_cost * 100, 2) if total_cost else 0, } return json.dumps(summary, ensure_asciiFalse)holdings.csv示例symbol,cost,shares 600519,1500,10 000001,12.5,1000 510300,3.8,50003.4 接入 LLM 的分析工具再注册一个工具把持仓结果交给模型生成点评from openai import OpenAI client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) mcp.tool() def review_portfolio(csv_path: str holdings.csv) - str: 基于持仓数据调用 LLM 生成组合点评与风险提示。 raw analyze_holdings(csv_path) try: resp client.chat.completions.create( modelos.getenv(MODEL_ID, claude-sonnet-4-5), messages[ {role: system, content: 你是严谨的投资组合分析助手只基于给定数据点评不预测涨跌必须提示风险。}, {role: user, content: f以下是持仓数据请分析集中度、盈亏结构并给出风险提示\n{raw}}, ], timeout30, ) return resp.choices[0].message.content except Exception as e: return f调用模型失败{e} if __name__ __main__: mcp.run()到这里三个工具就注册完了。mcp.run()默认走 stdio客户端通过mcp.json拉起。4. 验证请求本地联调与成功结果写完不验证等于没写。先单独测工具逻辑再走完整对话。第一步直接跑 Server 看有没有语法错python advisor_server.py没有报错、进程挂起等待输入说明注册成功。接着用 MCP Inspector 或客户端连接。这里给一个最小客户端import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters(commandpython, args[advisor_server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具, [t.name for t in tools.tools]) quote await session.call_tool(get_quote, {symbol: 600519}) print(行情结果, quote.content[0].text) review await session.call_tool(review_portfolio, {csv_path: holdings.csv}) print(组合点评, review.content[0].text) asyncio.run(main())预期输出类似可用工具 [get_quote, analyze_holdings, review_portfolio] 行情结果 {symbol: 600519, name: 贵州茅台, price: 1680.5, change_pct: 1.23} 组合点评 当前组合总市值约 3.6 万元贵州茅台占比偏高……看到工具列表和真实返回值就说明链路通了。如果review_portfolio返回的是模型生成的点评而不是报错说明 TaoToken 的 Key、Base URL、Model ID 三件套都配对成功。注意call_tool的参数名必须和函数签名一致symbol写成code会直接报参数校验错误。验证阶段建议按「单工具 → 多工具 → 对话串联」三步走每步都确认返回值别一上来就测完整对话出错了很难定位。5. 本篇常见错排查401、local proxy failed 与 reading choices这一节按真实报错来遇到直接对号入座。报错一401 Unauthorized。九成是 Key 没读到。检查echo $TAOTOKEN_API_KEY是否有值mcp.json的env是否传进去了。注意 stdio 模式下子进程不会自动继承你 shell 里的临时变量必须在配置里显式声明。另外确认 Key 没有多余空格或换行。报错二local proxy failed / connection refused。通常是 Server 没起来或端口不对。stdio 模式看command路径HTTP 模式看端口是否被占用。用python advisor_server.py手动跑一遍能复现就说明是启动参数问题不是协议问题。报错三reading choices 相关。这是模型返回结构不符合预期常见于base_url写错或模型名不存在。确认base_url是https://taotoken.net/apiMODEL_ID是控制台里真实存在的模型。如果返回体里没有choices先打印原始响应再判断。报错四OAuth / 鉴权跳转。部分客户端首次连接会走授权流程若卡住检查是否用了需要交互式登录的接入方式。纯 API Key 场景不该出现 OAuth出现就说明配置串了。报错五工具调用参数校验失败。模型传了多余字段或类型不对。解决办法是在工具 docstring 里把参数类型和取值范围写清楚模型会照着填。报错关键词最可能原因处理动作401Key 未注入检查 env 与变量名local proxy failedServer 未启动手动运行确认reading choicesbase_url/模型名错核对三件套OAuth接入方式串了改回 API Key排障时优先看 Server 端日志工具函数的异常一定要print出来否则客户端只看到一句「调用失败」无从下手。6. 继续往下走把原型变成能用的助手原型跑通后真正决定好不好用的是工具设计。我的经验是工具粒度别太细get_quote和analyze_holdings这种「一个工具干一件事」最稳返回一律结构化模型解析成本低所有外部调用都要有超时和降级行情挂了就返回缓存值并标注「数据可能延迟」。如果你要接真实行情把MOCK_QUOTES换成你的数据源注意频率限制和鉴权。持仓 CSV 可以换成数据库查询但别让 MCP 直连生产库用只读账号或中间层隔离。想验证不同模型对同一份持仓的点评差异可以去模型对话页 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 快速对比接入细节查文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。长期做 Agent 项目的话Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 会更省心。最后留一个我踩过的坑别在工具函数里做重计算模型等待超过 30 秒就会超时重试反而更慢。把重活拆成异步任务工具只返回任务 ID 和状态让对话保持轻快。