Agent-study项目教程（10）：基于MCP的智能数据爬取与分析Agent

一、理论部分

1. MCP 的核心价值：把“工具能力”从 Agent 中解耦出来

在传统 Agent 实现中，“工具”通常以内置函数的形式直接写在 Agent 代码里。这样做虽然简单，但会带来三个工程问题：

• 耦合度高：Agent 与工具强绑定，工具升级、迁移或复用成本高。
• 部署受限：工具只能跟随 Agent 同进程、同语言运行，跨语言/跨进程复用困难。
• 能力难共享：同一套工具能力难以被 IDE、其他应用或其他 Agent 体系复用。

MCP（Model Context Protocol）的思路是：把工具能力封装成标准化的MCP Server，由 Agent 作为MCP Client去调用。工具成为独立进程（甚至可远程部署）的服务，Agent 只关心“调用接口”和“拿到结果”。

2. MCP 的运行形态：Client/Server + Transport

一个 MCP 体系最关键的结构是：

• Server：提供工具（Tools）。每个工具有名字、参数与返回值。
• Client：发起工具调用请求，并消费返回结果。
• Transport：通信通道。本项目使用最易本地调试的stdio（标准输入/输出）作为进程间通信方式：Client 启动 Server 子进程，通过 stdin/stdout 传递协议消息。

stdio 模式的优势是：易调试、零网络配置；不足是：跨机器调用需要改为 SSE/HTTP 等远程传输方式。

3. 将 MCP 工具接入 Agent 编排：用 LangGraph 管理“调用-再推理”的循环

在“工具型 Agent”里，典型模式是：

1. 模型决定是否要调用工具（并给出工具名与参数）
2. 程序执行工具并拿到结果
3. 把工具结果反馈给模型，继续推理直到完成任务

本项目使用 LangGraph 组织这个循环：agent节点负责决策，tools节点负责执行工具；当最后一条消息包含tool_calls时进入 tools，否则结束。

4. 数据分析工具的关键工程点：受控执行 + 可视化落盘

“让模型写分析代码再执行”属于高风险能力，本项目做了两个关键约束以保证可运行性与可复现性：

• 非交互式绘图后端：强制matplotlib使用Agg，避免在无 GUI / 非主线程环境调用图形后端导致报错。
• 统一保存接口：提供save_plot(filename)，要求模型生成图表时只保存，不调用plt.show()。

同时，执行环境预置pd/plt/sns，并自动注入中文字体兼容配置，保证中文标题与标签可正常显示。

二、实战部分

1. 项目目标

实现一个“智能数据爬取与分析 Agent”，具备：

• 通过 MCP 工具爬取 GitHub Trending（指定语言）
• 让模型生成分析代码：整理数据、绘制图表（柱状图）
• 执行分析代码并将图表保存到output/目录

默认示例任务为：分析 GitHub Trending 上 Python 热门项目，并绘制 Star 数最高的 Top-5 柱状图。

2. 运行准备

请确保本地已安装依赖并配置环境变量（不要把真实密钥写进代码或公开仓库）：

依赖安装（示例）：

pipinstall-rrequirements.txt

环境变量（示例）：

• DEEPSEEK_API_KEY
• DEEPSEEK_BASE_URL

说明：

• 本项目依赖mcp、requests、beautifulsoup4、pandas、matplotlib、seaborn等。

3. 项目代码（核心实现）

本项目由三部分组成：MCP Server（暴露工具）、工具实现（爬虫与分析执行器）、MCP Client/Agent（LangGraph 编排）。

3.1 MCP Server：用 FastMCP 暴露两个工具

Server 负责把“爬取”和“分析执行”封装成标准工具接口：

importosimportsys sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__),"../..")))frommcp.server.fastmcpimportFastMCPfromstage_06_advanced.project_10_data_analysis.tools.crawlerimportcrawl_github_trendingfromstage_06_advanced.project_10_data_analysis.tools.analysisimportexecute_analysis_code mcp=FastMCP("DataAnalysisServer")@mcp.tool()defget_github_trending(language:str="python")->str:results=crawl_github_trending(language)returnstr(results)@mcp.tool()defanalyze_data(code:str)->str:returnexecute_analysis_code(code)if__name__=="__main__":mcp.run()

工具接口设计要点：

• get_github_trending(language)：返回热门仓库列表（名称、stars、描述）
• analyze_data(code)：执行模型生成的 Python 分析代码，输出执行日志与保存路径

3.2 爬虫工具：抓取 GitHub Trending 并结构化为列表

爬虫侧输出的是 Python 列表（每个元素是字典），便于后续转换为 DataFrame：

importrequestsfrombs4importBeautifulSoupfromtypingimportList,Dictdefcrawl_github_trending(language:str="python")->List[Dict[str,str]]:url=f"https://github.com/trending/{language}?since=daily"headers={"User-Agent":"Mozilla/5.0"}response=requests.get(url,headers=headers,timeout=10)response.raise_for_status()soup=BeautifulSoup(response.text,"html.parser")articles=soup.find_all("article",class_="Box-row")repos=[]forarticleinarticles:title_tag=article.find("h2",class_="h3 lh-condensed")repo_name=title_tag.get_text().strip().replace("\n","").replace(" ","")iftitle_tagelse"Unknown"desc_tag=article.find("p",class_="col-9 color-fg-muted my-1 pr-4")description=desc_tag.get_text().strip()ifdesc_tagelse"No description"footer=article.find("div",class_="f6 color-fg-muted mt-2")stars=0iffooter:forlinkinfooter.find_all("a"):if"stargazers"inlink.get("href",""):star_text=link.get_text().strip().replace(",","")stars=int(star_text)ifstar_text.isdigit()else0breakrepos.append({"name":repo_name,"stars":stars,"description":description})returnrepos

3.3 分析执行器：受控执行 + save_plot 落盘

分析执行器的目标是：让模型可以写 pandas/matplotlib/seaborn 代码，并在受控环境中执行，且图表可保存到项目目录：

importsysimportioimportosimportmatplotlib matplotlib.use("Agg")importpandasaspdimportmatplotlib.pyplotaspltimportseabornassnsimporttracebackdefexecute_analysis_code(code:str)->str:old_stdout=sys.stdout redirected_output=io.StringIO()sys.stdout=redirected_output output_dir=os.path.join(os.path.dirname(os.path.dirname(__file__)),"output")os.makedirs(output_dir,exist_ok=True)defsave_plot(filename="analysis_result.png"):filepath=os.path.join(output_dir,filename)try:plt.savefig(filepath)print(f"图表已保存至:{filepath}")finally:plt.close()local_scope={"pd":pd,"plt":plt,"sns":sns,"save_plot":save_plot}font_fix=""" import platform system_name = platform.system() if system_name == 'Windows': plt.rcParams['font.sans-serif'] = ['Microsoft YaHei', 'SimHei', 'SimSun', 'Malgun Gothic'] elif system_name == 'Darwin': plt.rcParams['font.sans-serif'] = ['Arial Unicode MS', 'PingFang SC', 'Heiti TC'] else: plt.rcParams['font.sans-serif'] = ['WenQuanYi Micro Hei', 'Noto Sans CJK SC', 'SimHei'] plt.rcParams['axes.unicode_minus'] = False """try:exec(font_fix+"\n"+code,local_scope)result=redirected_output.getvalue()returnresultifresultelse"代码执行成功 (无输出)。"exceptException:returnf"执行代码出错:\n{traceback.format_exc()}"finally:sys.stdout=old_stdout

3.4 MCP Client + LangGraph：把工具调用接入 Agent 决策循环

Client 侧使用 stdio 启动 Server，并把 MCP 工具包装成 LangChain 工具供 LangGraph 调用：

importosimportasynciofromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessagefromlangchain_core.toolsimportStructuredToolfromlanggraph.graphimportStateGraph,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.prebuiltimportToolNodefromtypingimportTypedDict,Annotated,Listfromlangchain_core.messagesimportBaseMessagefrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_client load_dotenv()server_params=StdioServerParameters(command="python",args=[os.path.join(os.path.dirname(__file__),"server.py")],env=os.environ.copy(),)asyncdef_call_mcp_tool(tool_name:str,arguments:dict):asyncwithstdio_client(server_params)as(read,write):asyncwithClientSession(read,write)assession:awaitsession.initialize()result=awaitsession.call_tool(tool_name,arguments=arguments)texts=[c.textforcinresult.contentifc.type=="text"]return"\n".join(texts)defcall_get_github_trending(language:str="python"):returnasyncio.run(_call_mcp_tool("get_github_trending",{"language":language}))defcall_analyze_data(code:str):returnasyncio.run(_call_mcp_tool("analyze_data",{"code":code}))get_github_trending_tool=StructuredTool.from_function(func=call_get_github_trending,name="get_github_trending",description="获取 GitHub Trending 数据。",)analyze_data_tool=StructuredTool.from_function(func=call_analyze_data,name="analyze_data",description="执行 Python 分析代码并保存图表。",)tools=[get_github_trending_tool,analyze_data_tool]classAgentState(TypedDict):messages:Annotated[List[BaseMessage],add_messages]llm=ChatOpenAI(model="deepseek-chat",temperature=0,openai_api_base="https://api.deepseek.com",openai_api_key=os.getenv("DEEPSEEK_API_KEY"),)llm_with_tools=llm.bind_tools(tools)defagent_node(state:AgentState):response=llm_with_tools.invoke(state["messages"])return{"messages":[response]}defshould_continue(state:AgentState):last_message=state["messages"][-1]return"tools"iflast_message.tool_callselseEND workflow=StateGraph(AgentState)workflow.add_node("agent",agent_node)workflow.add_node("tools",ToolNode(tools))workflow.set_entry_point("agent")workflow.add_conditional_edges("agent",should_continue,{"tools":"tools",END:END})workflow.add_edge("tools","agent")app=workflow.compile()

4. 如何运行

在项目根目录执行：

python stage_06_advanced/project_10_data_analysis/main.py

程序会自动以 stdio 启动 MCP Server，并执行默认示例任务。图表输出将保存到：

• stage_06_advanced/project_10_data_analysis/output/

5. 运行结果示例

运行过程会输出 MCP 连接信息与 LangGraph 节点流转信息（示例）：

🚀 启动 MCP 架构数据分析 Agent... 🔗 连接到 MCP Server: python ['.../server.py'] 用户: 请帮我分析一下 GitHub 上 Python 的热门项目，画一个柱状图展示 Star 数最高的 5 个项目，并保存图表。 ---- 节点: agent ---- ---- 节点: tools ---- ---- 节点: agent ---- ---- 节点: tools ---- ---- 节点: agent ---- ✅ 任务完成。请检查 'stage_06_advanced/project_10_data_analysis/output' 目录查看结果。

在output/目录下，你会看到生成的图表文件，并在工具返回日志中看到保存路径，例如：

图表已保存至: .../stage_06_advanced/project_10_data_analysis/output/python_top5_stars.png

6. 项目总结

• MCP 把“工具能力”封装为标准化 Server，使 Agent 与工具解耦，便于跨进程、跨语言、跨应用复用。
• stdio 是最适合本地演示的传输方式：零配置、易调试；若要远程调用，可扩展为 SSE/HTTP。
• 对“代码执行型工具”必须做工程约束：非交互式后端、统一保存接口、受控命名空间与中文字体处理，保证可运行性与可复现性。
• 将 MCP 工具接入 LangGraph 后，Agent 可以以“可控循环”的方式完成：检索数据 → 生成分析代码 → 执行 → 输出图表的完整闭环。