financial-analyst-deepseek：用 CrewAI + DeepSeek-R1 构建 MCP 驱动的本地金融分析 Agent

financial-analyst-deepseek：用 CrewAI + DeepSeek-R1 构建 MCP 驱动的本地金融分析 Agent

【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub

这篇技术指南聚焦 financial-analyst-deepseek 项目，讲解如何用 CrewAI 编排一个「自然语言 → 结构化解析 → 代码生成 → 代码执行」的股票分析多 Agent 工作流，让本地运行的 DeepSeek-R1（经 Ollama 提供）完成行情获取与可视化脚本编写，最终以 MCP Server 形态接入 Cursor IDE。读完本文，你将掌握完整的安装步骤、三个关键源码文件（finance_crew.py、server.py、pyproject.toml）的逐段原理，以及从「配置 MCP → 提问 → 拿到可执行分析代码」的完整调用链路，可以直接在本地复现这一金融分析 Agent。

项目定位与技术栈总览

该项目构建的是一个MCP 驱动的金融分析 Agent：用户以自然语言提出查询（例如“展示特斯拉过去 3 个月的股价表现”），系统解析出股票代码、时间范围和期望动作，自动生成并执行 Python 可视化代码，最终把结果返回给作为 MCP Host 的 Cursor。整个流程不需要手写任何行情抓取脚本。

依据 README.md 与源码，核心技术栈可以拆成三层：

仓库中的相关文件结构如下（目录：financial-analyst-deepseek）：

• finance_crew.py：业务核心，定义了三个 Agent、三个顺序 Task 及 Crew 组装；
• server.py：MCP Server 出口，基于mcp.server.fastmcp.FastMCP暴露三个工具；
• pyproject.toml：依赖清单与 Python 版本约束；
• uv.lock：锁定依赖版本的锁文件，保证uv sync可复现环境；
• building-financial-analyst.ipynb：从零搭建该工作流的 Jupyter 教学版，含一次真实运行的 Agent 轨迹输出。

从代码结构看，整个项目刻意做了「业务与协议分离」：finance_crew.py只关心 Crew 的构建与执行，server.py只负责把 Crew 的入口函数暴露成 MCP 工具。这也是后续接入其他 Host 时可以复用的核心思路。

环境准备：安装 Ollama 与项目依赖

安装 Ollama 并拉取 DeepSeek-R1

README 提供了 Linux 下的一键安装方式，然后拉取模型：

# 在 Linux 上安装 Ollama curl -fsSL https://ollama.com/install.sh | sh # 拉取 DeepSeek-R1 模型 ollama pull deepseek-r1

需要特别留意的是 finance_crew.py 中实际引用的模型标识为ollama/deepseek-r1:7b，即 CrewAI 通过ollama/前缀把 LLM 请求路由到本地 Ollama 服务（base_url=http://localhost:11434）。因此本地 Ollama 中应存在与代码匹配的模型标签，例如补充拉取 7b 版本：

ollama pull deepseek-r1:7b

同时，运行期间需保证 Ollama 服务处于启动状态，因为 Agent 的每一次推理都会打到localhost:11434。

安装 Python 依赖

项目要求Python 3.12 或更高版本（见 pyproject.toml 中的requires-python = ">=3.12"）。

推荐使用uv直接同步依赖：

uv sync

仓库中已提交 uv.lock，uv sync会依据锁文件还原出完全一致的环境。除此之外，README 也给出了等价的手动安装命令，可直接安装到本地 Python 环境：

pip install crewai crewai-tools ollama mcp pydantic yfinance pandas matplotlib

从 pyproject.toml 可以看到项目声明的完整依赖（版本为该仓库当前锁定的范围），这对理解各库的职责很有帮助：

核心工作流源码解析：finance_crew.py

finance_crew.py 是整个系统的“发动机”。它把一次股票分析拆成三段顺序执行，每段由一个 Agent 负责。

第一步：LLM 连接配置

所有 Agent 共享同一个LLM实例，它指向本地 Ollama 的 DeepSeek-R1：

llm = LLM( model="ollama/deepseek-r1:7b", base_url="http://localhost:11434", # temperature=0.7 )

同一文件里还保留了一段被注释的gpt-4o云端配置（openai/gpt-4o）。这意味着整个 Crew 对 LLM 供应商是无感知的——只要 CrewAI 支持该 provider，把model换成云端模型即可切换后端。若使用 OpenAI 系模型，需另行配置OPENAI_API_KEY环境变量；这是代码结构上可以推断出的扩展点。

第二步：Query 解析 Agent——用 Pydantic 锁定结构化输出

工作流的第一环是解析用户查询。为了让后续「代码生成 Agent」拿到机器可读的输入，项目用 Pydantic 定义了输出模型：

class QueryAnalysisOutput(BaseModel): """Structured output for the query analysis task.""" symbols: list[str] = Field(..., description="List of stock ticker symbols (e.g., ['TSLA', 'AAPL']).") timeframe: str = Field(..., description="Time period (e.g., '1d', '1mo', '1y').") action: str = Field(..., description="Action to be performed (e.g., 'fetch', 'plot').")

随后创建查询解析 Agent 与 Task，并通过output_pydantic=QueryAnalysisOutput要求 Agent 的最终输出必须被解析为该结构：

query_parser_agent = Agent( role="Stock Data Analyst", goal="Extract stock details and fetch required data from this user query: {query}.", backstory="You are a financial analyst specializing in stock market data retrieval.", llm=llm, verbose=True, memory=True, ) query_parsing_task = Task( description="Analyze the user query and extract stock details.", expected_output="A dictionary with keys: 'symbol', 'timeframe', 'action'.", output_pydantic=QueryAnalysisOutput, agent=query_parser_agent, )

这里有两个值得注意的实现细节：

1. goal中以{query}作为插槽变量，运行时由crew.kickoff(inputs={"query": ...})注入具体用户查询；
2. 解析 Agent 开启了memory=True（会话内记忆），便于在多轮对话中保持上下文。

从源码结构看，当前版本模型字段使用复数symbols: list[str]，正好匹配 README 中「Compare Apple and Microsoft stocks」这类多标的对比查询；而教学版 building-financial-analyst.ipynb 中同名模型最初是单数symbol: str。两者并不冲突，说明项目支持单/多股票两种形态，具体字段以当前 finance_crew.py 为准。

第三步：Code Writer Agent——生成可视化代码

解析 Agent 产出{symbols, timeframe, action}后，交给「高级 Python 开发者」角色写代码：

code_writer_agent = Agent( role="Senior Python Developer", goal="Write Python code to visualize stock data.", backstory="""You are a Senior Python developer specializing in stock market data visualization. You are also a Pandas, Matplotlib and yfinance library expert. You are skilled at writing production-ready Python code""", llm=llm, verbose=True, ) code_writer_task = Task( description="""Write Python code to visualize stock data based on the inputs from the stock analyst where you would find stock symbol, timeframe and action.""", expected_output="A clean and executable Python script file (.py) for stock visualization.", agent=code_writer_agent, )

注意该 Agent 不直接持有任何拉数工具——它只负责“写代码”，把 yfinance 取数、pandas 处理、matplotlib 绘图的工作全部写进一段可执行脚本。backstory明确限定了它的专家领域，引导模型只产出高质量、可直接运行的代码。

第四步：Code Execution Agent——审阅并执行

最后一段由执行专家把关。它通过allow_code_execution=True自动挂载 CrewAI 的代码解释能力（源码注释写得很清楚：“This automatically adds the CodeInterpreterTool”），并开启allow_delegation=True，允许在出错时把任务回派给写码 Agent 修正：

code_interpreter_tool = CodeInterpreterTool() code_execution_agent = Agent( role="Senior Code Execution Expert", goal="Review and execute the generated Python code by code writer agent to visualize stock data and fix any errors encountered. It can delegate tasks to code writer agent if needed.", backstory="You are a code execution expert. You are skilled at executing Python code.", # tools=[code_interpreter_tool], allow_code_execution=True, # This automatically adds the CodeInterpreterTool allow_delegation=True, llm=llm, verbose=True, ) code_execution_task = Task( description="""Review and execute the generated Python code by code writer agent to visualize stock data and fix any errors encountered.""", expected_output="A clean, working and executable Python script file (.py) for stock visualization.", agent=code_execution_agent, )

Crew 组装与入口函数

三个 Agent、三个 Task 按顺序拼接成 Crew：

crew = Crew( agents=[query_parser_agent, code_writer_agent, code_execution_agent], tasks=[query_parsing_task, code_writer_task, code_execution_task], process=Process.sequential ) # Function to be wrapped inside MCP tool def run_financial_analysis(query): result = crew.kickoff(inputs={"query": query}) return result.raw

Process.sequential保证 Task 按声明顺序执行：解析 → 写码 → 执行。run_financial_analysis(query)是被 MCP 层包装的对外入口，kickoff返回的result.raw通常是最终生成并校验过的 Python 代码文本。文件底部还保留了独立运行模式：python finance_crew.py时会用内置示例查询 “Plot YTD stock gain of Tesla” 直接跑一遍并打印结果，适合在没有 IDE 时先验证工作流是否通。

MCP 服务层：server.py 如何把工作流变成工具

server.py 是本项目接入 Cursor 的关键。它用 FastMCP 创建名为financial-analyst的 MCP Server，并暴露三个工具，形成「分析 → 存盘 → 绘图」的完整闭环：

from mcp.server.fastmcp import FastMCP from finance_crew import run_financial_analysis mcp = FastMCP("financial-analyst")

工具一analyze_stock(query)——唯一直接调用 Crew 的入口。它接收包含股票代码、时间范围、动作的自然语言查询，内部捕获异常并返回run_financial_analysis(query)的结果，即一段格式良好的 Python 代码：

@mcp.tool() def analyze_stock(query: str) -> str: """ Analyzes stock market data based on the query and generates executable Python code for analysis and visualization. Returns a formatted Python script ready for execution. ... """ try: result = run_financial_analysis(query) return result except Exception as e: return f"Error: {e}"

工具二save_code(code)——把 Crew 返回的代码落盘为工作目录下的stock_analysis.py：

@mcp.tool() def save_code(code: str) -> str: try: with open('stock_analysis.py', 'w') as f: f.write(code) return "Code saved to stock_analysis.py" except Exception as e: return f"Error: {e}"

工具三run_code_and_show_plot()——读取stock_analysis.py并exec执行，从而真正取数、绘图并弹出图表：

@mcp.tool() def run_code_and_show_plot() -> str: with open('stock_analysis.py', 'r') as f: exec(f.read())

服务通过标准输入输出（stdio）方式与 Host 通信：

if __name__ == "__main__": mcp.run(transport='stdio')

可以推断，Cursor 侧拿到analyze_stock的返回值后，会按工具链自动补调save_code与run_code_and_show_plot，实现“对话中直接出图”。analyze_stock的 docstring 中已写明约束：query 字符串必须包含股票代码（如 TSLA、AAPL、NVDA）、时间范围（如1d、1mo、1y）以及动作（如 plot、analyze、compare），这实际上为模型的工具调用做了提示工程。

在 Cursor 中接入 MCP Server（运行步骤）

README 给出了非常明确的 Cursor 接入路径，请按以下顺序操作：

1. 打开 Cursor 设置（Settings）；
2. 选择MCP分类；
3. 点击Add new global MCP server，新增一个全局 MCP 服务。

在弹出的 JSON 配置文件中填入：

{ "mcpServers": { "financial-analyst": { "command": "uv", "args": [ "--directory", "absolute/path/to/project_root", "run", "server.py" ] } } }

其中absolute/path/to/project_root需要替换为本项目目录的绝对路径，也就是包含 server.py 与 pyproject.toml 的那个目录。采用uv --directory <目录> run server.py启动的好处在于：uv run会自动使用项目环境（与uv sync生成的.venv一致），从而保证 Cursor 拉起 MCP Server 时能正确找到 FastMCP 与项目模块，也无需手动激活虚拟环境。

配置完成后，你应当能在 Cursor 的 MCP 设置面板中看到financial-analyst服务出现在列表里。最后一步：在 Cursor MCP 设置中，把该服务对应的开关拨到连接状态，让 Server 与 Host 真正建立连接。至此服务即运行就绪。

在 Cursor 中发起一次金融分析

连接成功后，直接在 Cursor 的对话输入框中提问即可。README 给出的典型示例查询如下：

• “Show me Tesla's stock performance over the last 3 months”
• “Compare Apple and Microsoft stocks for the past year”
• “Analyze the trading volume of Amazon stock for the last month”

用户只需给出「股票 + 时间范围 + 动作」，其余（代码生成、取数、执行、绘图）全部交给 Agent 链自动完成。若查询信息不完整，analyze_stock 的 docstring 会引导模型向用户补充股票代码、时间范围和动作三要素。

一次真实调用的完整执行链路

building-financial-analyst.ipynb 记录了一次以 “Plot YTD stock gain of Tesla” 为输入的完整执行轨迹，可以直接佐证整个链条的运行机制：

1. Stock Data Analyst（解析阶段）输出结构化结论：symbol = TSLA、timeframe = YTD、action = Plot，即用户意图被成功降维成(symbol, timeframe, action)三元组，为后续代码生成提供了确定性输入；
2. Senior Python Developer（写码阶段）基于三元组生成可视化脚本，示例输出片段为：

import yfinance as yf import pandas as pd import matplotlib.pyplot as plt # Extracting the required information from the JSON configuration config = { "symbol": "TSLA", "timeframe": "YTD", "action": "Plot" } # Fetching historical data for Tesla from Yahoo Finance tesla_data = yfinance.download( tickers=config['symbol'], period=config['timeframe'], interval='1d' ) # Creating a simple plot of the stock prices over time plt.figure(figsize=(10, 6)) plt.plot(tesla_data.index, tesla_data['Adj Close']) plt.title(f'Tesla Stock Price (TSLA) - {config["timeframe"]}') plt.xlabel('Date') plt.ylabel('Adjusted Closing Price ($)') plt.grid(True) plt.show()

3. Senior Code Execution Expert（执行阶段）审阅该脚本并完成执行验证。之后这段代码会经save_code写入stock_analysis.py，再由run_code_and_show_plot执行出图。

如果你不想经由 IDE，也可以用纯代码方式观察同一条链路：直接运行python finance_crew.py，脚本会用内置的 Tesla 示例查询执行 Crew 并打印结果；或在 Jupyter 中逐段重放 building-financial-analyst.ipynb，便于调试 Agent 行为、理解 verbose 输出的每一步推理过程。

关键实现细节与可扩展方向

本地与云端模型一键切换：finance_crew.py中同时保留了 DeepSeek-R1（本地 Ollama）与 gpt-4o（云端 OpenAI）两套 LLM 配置。若机器资源不足以本地推理 7B 模型，可切换到云端配置——前提是正确设置对应供应商的环境变量，且按 pyproject.toml 的约定管理好依赖。

代码执行能力的来源：执行 Agent 通过allow_code_execution=True获得代码执行工具（源码注释表明这会自动添加CodeInterpreterTool），同时allow_delegation=True让它在遇到执行错误时可以把修正任务回派给 Code Writer Agent，形成「写码—执行—修错」的自愈闭环。

结构化输出的价值：整个流水线能否稳定工作，很大程度依赖output_pydantic=QueryAnalysisOutput对第一环输出做的强制约束。解析 Agent 即便推理风格再自由（例如 notebook 中 DeepSeek-R1 的<think>思考过程），最终交付物也必须收敛为可校验的{symbols, timeframe, action}结构，从而屏蔽掉后续环节对自由文本的解析不确定性。

运行前提与限制（结合仓库现状整理）：

• 依赖 Python 3.12+，建议用仓库内 uv.lock +uv sync复现环境；
• 本地需保持 Ollama 运行且包含deepseek-r1:7b标签的模型；
• 行情数据来自 yfinance（Yahoo Finance），因此执行阶段需要联网，且数据口径以该数据源为准；
• MCP 接入仅针对 Cursor 这类支持 stdio 传输的 Host 描述，仓库未涉及 HTTP/SSE 等其他传输方式的配置；
• stock_analysis.py由save_code生成在 MCP Server 的工作目录下，绘图脚本与 Agent 的执行环境一致。

如果你想进一步改造，比较自然的切入点包括：把QueryAnalysisOutput扩展出更多字段（如区间粒度、对比基准）来支持更复杂的分析语义；为 Crew 追加“解读图表并输出结论”的第四个 Agent；或为 MCP Server 增加历史上下文工具。当前仓库给出的这套「业务 Crew + FastMCP 包装」结构，已经为这些扩展预留了清晰的边界。

【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub