1. 項目概述

項目目標

• 構建一個本地智能輿論分析系統。

• 利用自然語言處理和多工具協作，實現用戶查詢意圖的自動理解。

• 進行新聞檢索、情緒分析、結構化輸出和郵件推送。

系統流程

1. 用戶查詢：用戶輸入查詢請求。

2. 提取關鍵詞：從用戶查詢中提取關鍵詞。

3. 使用Google Serper API搜索：利用API獲取新聞前5篇文章。

4. 分析情感傾向：對獲取的文章進行情緒分析。

5. 保存為Markdown文件：將分析結果保存為Markdown格式。

6. 發送郵件：將結果通過郵件發送給用戶。

系統架構

• Client-Server架構：

  • 客戶端（Client）：用戶交互入口，負責接收輸入和調用大語言模型進行任務規劃。

  • 服務器端（Server）：工具能力提供者，處理數據和響應客戶端請求。

項目執行流程

1. 客戶端加載模型：加載本地模型配置，與服務器建立連接。

2. 用戶輸入查詢：客戶端自動調用大語言模型，將自然語言請求轉化為結構化工具調用。

3. 客戶端驅動服務器端：完成關鍵詞搜索、新聞采集、情緒傾向分析、報告生成和郵件發送。

2. MCP的環境準備

MCP的開發需要借助uv（虛擬環境管理工具）進行虛擬環境創建和依賴的管理。

2.1 安裝uv

• 提供了兩種安裝uv的方法：

  1. 使用pip安裝：pip install uv

  2. 使用conda安裝（針對已安裝Anaconda環境的用戶）：conda install uv

2.2 創建MCP項目

• 通過cd命令進入要創建項目的空間，然后使用以下命令創建一個空的MCP項目

  uv init mcp-project

• 這將在指定目錄下創建一個名為mcp-project的文件夾，其中包含初始化的項目結構。

• 在mcp-project目錄下，創建兩個Python文件，分別是client.py和server.py：

  • client.py是客戶端，用戶與客戶端進行交互。

  • server.py是服務端，其中包含了多種工具函數，客戶端會調用其中的工具函數進行操作。

這樣，MCP項目的創建便完成了。

3. 代碼實現

3.1 確定大模型參數

• 創建一個.env文件，在該文件中添加相關的環境變量，這些變量分別代表阿里百煉平臺的URL、選擇的模型名稱、個人的百煉平臺API。

  • BASE_URL:

    • 指定用于API請求的基礎URL，例如它可以是阿里云的DashScope服務的兼容模式地址：https://dashscope.aliyuncs.com/compatible-mode/v1。

  • MODEL:

    • 指定要使用的模型名稱。

  • DASHSCOPE_API_KEY:

    • DashScope服務的API密鑰，用于認證和授權訪問DashScope平臺的API。

  • SERPER_API_KEY:

    • Serper服務的API密鑰，Serper是一個提供搜索引擎結果頁面（SERP）數據的API服務，允許開發者通過HTTP請求獲取搜索引擎的結果。

  • SMTP_SERVER:

    • 指定用于發送電子郵件的SMTP服務器地址。在您的例子中，它是：smtp.163.com，這是163郵箱的SMTP服務器。

  • SMTP_PORT:

    • 指定SMTP服務器的端口號。在您的例子中，端口號是：465，這是一個常用的SMTP服務端口，通常用于SSL加密連接。

  • EMAIL_USER:

    • 用于SMTP認證的電子郵件用戶名，通常是您的電子郵件地址。。

  • EMAIL_PASS:

    • 用于SMTP認證的電子郵件密碼。

3.2 client.py的構建

3.2.1 功能分析

• 首先從客戶端入手，進行client.py的構建。其總體架構如下：

  [配置初始化]
  [連接工具服務器（MCP Server）]
  [用戶提問] -> chat_loop()[[LLM 規劃工具調用鏈]][順序執行工具鏈]
  [保存分析結果 & 最終回答]

• 運行過程中有以下幾個關鍵步驟：

  1. 客戶端從本地配置文件中讀取必要的信息，完成大模型參數的設定（見3.2.2 確定大模型參數），并初始化所需的運行環境（見3.2.2 初始化客戶端配置）。

  2. 程序啟動服務端腳本并與其建立通信，獲取可用的工具信息（見3.2.3 啟動MCP工具服務連接）。

  3. 完成連接后，客戶端將根據用戶輸入的請求，協調內部調度器對工具鏈任務進行統一管理（見3.2.4 工具鏈任務調度器）。

  4. 在與用戶交互的過程中，系統會持續監聽用戶輸入（見3.2.5 用戶交互循環），并調用大模型對任務進行智能拆解，規劃合適的工具鏈執行順序（見3.2.6 智能規劃工具鏈）。

  5. 每次任務執行完畢后，客戶端將自動釋放相關資源，確保系統穩定運行與退出（見3.2.7 關閉資源）。

  6. 整個流程由主函數串聯驅動，形成完整的一條執行主線（見3.2.8 主流程函數）。

3.2.2 初始化客戶端配置

在client.py中創建一個MCPClient類，用于封裝和管理與MCP協議相關的所有客戶端邏輯，隨后在里面編寫各種相關函數。

class MCPClient:def __init__(self):# 創建 AsyncExitStack, 用于托管所有異步資源釋放，這是為了后續連接 MCP Server 時使用 'async with' 語法自動管理上下文。self.exit_stack = AsyncExitStack()# 從環境中讀取配置項self.openai_api_key = os.getenv("DASHSCOPE_API_KEY")self.base_url = os.getenv("BASE_URL")self.model = os.getenv("MODEL")# 對 LLM 相關配置進行初始化if not self.openai_api_key:raise ValueError("未找到 OpenAI API Key, 請在 .env 文件中設置 DASHSCOPE_API_KEY")# 初始化 OpenAI 客戶端對象self.client = OpenAI(api_key=self.openai_api_key,base_url=self.base_url)# 初始化 MCP Session（用于延遲賦值），等待連接 MCP Server 后再初始化它self.session: Optional[ClientSession] = None

3.2.3 啟動MCP工具服務連接

connect_to_server 函數的作用是連接并啟動本地的服務器腳本。它會先判斷腳本類型（必須是 .py 或 .js），再根據類型選擇對應的啟動方式（Python或Node.js）。接著，它會通過MCP提供的方式啟動服務端腳本，并建立起與服務端的通信通道。建立連接后，客戶端會初始化會話，并獲取服務器上有哪些工具可以使用，方便后續根據任務調用這些工具。整個過程相當于“把工具服務開起來，并準備好對話”。

async def connect_to_server(self, server_script_path: str):# 對服務器腳本進行判斷，只允許是 .py 或 .jsis_python = server_script_path.endswith('.py')is_js = server_script_path.endswith('.js')if not (is_python or is_js):raise ValueError("服務器腳本必須是 .py 或 .js 文件")# 確定啟動命令，.py 用 python，.js 用 nodecommand = "python" if is_python else "node"# 構造 MCP 所需的服務器參數，包括啟動參數、腳本路徑參數、環境變量（為 None 表示默認）server_params = StdioServerParameters(command=command, args=(server_script_path,), env=None)# 啟動 MCP 工具服務進程（并建立 stdio 通信）self.stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))# 封裝通信通道，讀取服務器返回的數據，并向服務器發送請求self.stdio, self.write = stdio.transport# 創建 MCP 客戶端會話對象self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))# 初始化會話await self.session.initialize()# 獲取工具列表并打印response = await self.session.list_tools()if not ("MC是服務器，支持以下工具：", {tool_name for tool in tools})

詳細步驟

1. 判斷腳本類型：

   • 檢查 server_script_path 是否以 .py 或 .js 結尾，否則拋出 ValueError。

2. 確定啟動命令：

   • 如果是 .py 文件，使用 python 命令；如果是 .js 文件，使用 node 命令。

3. 構造服務器參數：

   • 使用 StdioServerParameters 構造服務器參數，包括命令、腳本路徑和環境變量。

4. 啟動 MCP 工具服務進程：

   • 使用 stdio_client 啟動 MCP 工具服務進程，并建立 stdio 通信。

5. 封裝通信通道：

   • 讀取服務器返回的數據，并向服務器發送請求。

6. 創建 MCP 客戶端會話對象：

   • 使用 ClientSession 創建 MCP 客戶端會話對象。

7. 初始化會話：

   • 調用 session.initialize() 初始化會話。

8. 獲取工具列表并打印：

   • 調用 session.list_tools() 獲取工具列表，并打印支持的工具。

3.2.4??工具鏈任務調度器

process_query 函數是客戶端處理用戶提問的核心部分，負責從接收問題到規劃任務、調用工具、生成回復，再到保存結果的整個閉環。

功能步驟

1. 獲取支持的工具列表

   • 向服務器請求當前支持的工具列表，例如“新聞搜索”、“情感分析”、“發送郵件”等。

2. 提取關鍵詞

   • 從用戶問題中提取關鍵詞，生成統一的文件名，后續所有工具都會使用這個名字保存或讀取文件，保證流程一致。

3. 工具鏈規劃

   • 將問題交給大語言模型，決定如何使用這些工具（如先查新聞，再分析情感，再發郵件）。

4. 調用服務器上的工具

   • 按順序調用服務器上的工具，并在調用前動態地填入一些信息（如文件名或路徑）。

5. 收集執行結果

   • 收集所有工具執行完畢后的結果，程序會再調用一次大模型，讓它根據整個過程總結一個回答。

6. 保存對話記錄

   • 將對話記錄（包括用戶的提問和模型的回答）自動保存成一個 .txt 文件，方便后續查閱。

?

async def process_query(self, query: str) -> str:# 準備初始消息和獲取工具列表messages = {"role": "user", "content": query}response = await self.session.list_tools()available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"input_schema": tool.inputSchema,},} for tool in response.tools]# 提取問題的關鍵詞，對文件名進行生成keyword_match = re.search(r"(關于|分析|查詢|搜索|查看)(.+?)(\n|$)", query)keyword = keyword_match.group(2) if keyword_match else "分析對象"safe_keyword = re.sub(r'[\\/*?:"<>|]', '', keyword)[:20]timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")md_filename = f"Sentiment_{safe_keyword}_{timestamp}.md"md_path = os.path.join("./sentiment-reports/", md_filename)# 更新查詢，將文件名添加到原始查詢中，使大模型在調用工具鏈時可以識別到該信息messages = {"role": "user", "content": query}md_path = (query.strip() + f" [md_filename={md_filename}]")[md_path + md_path]messages = {"role": "user", "content": query}tool_plan = await self.plan_tool_usage(query, available_tools)tool_outputs = {}messages = [{"role": "user", "content": query}]# 依次執行工具調用，并收集結果for step in tool_plan:tool_name = step["name"]tool_args = step["arguments"]for key, val in tool_args.items():if isinstance(val, str) and val.startswith("{{") and val.endswith("}})"):ref_key = val.strip("{{").strip("}}")resolved_val = tool_outputs.get(ref_key, val)tool_args[key] = resolved_val# 注入統一的文件名或路徑（用于分析和郵件）if tool_name == "analyze_sentiment" and "filename" not in tool_args:tool_args["filename"] = md_filenameresult = await self.session.call_tool(tool_name, tool_args)tool_output[tool_name] = result.content[0].textmessages.append({"role": "tool","tool_called": tool_name,"content": result.content[0].text})# 調用大模型生成回復信息，并輸出保存結果final_response = self.client.chat_completions.create(model=self.model,messages=messages)final_output = final_response.choices[0].message.content# 對輔助函數進行定義，目的是把文本清理成合法的文件名def clean_filename(text: str) -> str:text = text.strip().replace("\n", "").replace("\r", "")return text[:50]# 使用清理函數處理用戶查詢，生成用于文件命名的前綴，并添加時間戳、設置輸出目錄safe_filename = clean_filename(query)timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")filename = f"{safe_filename}_{timestamp}.txt"output_dir = "./lin_outputs"os.makedirs(output_dir, exist_ok=True)file_path = os.path.join(output_dir, filename)# 將對話內容寫入 md 文檔，其中包含用戶的初始提問以及模型的最終回復結果with open(file_path, 'w', encoding='utf-8') as f:f.write(f"* 用戶提問：{query}\n\n")f.write(f"* 模型回復：\n{final_output}\n")print(f"📄 對話記錄已保存為：{file_path}")return final_output

3.2.5 用戶交互循環筆記

概述

chat_loop 函數是客戶端的“對話主入口”，負責程序和用戶之間的交互。它是一個無限循環，不斷等待用戶輸入問題，并處理這些輸入。

主要功能

1. 提示用戶輸入：

   • 程序啟動時，打印提示信息，告知用戶系統已啟動，可以開始提問（輸入 quit 可退出）。

2. 無限循環等待輸入：

   • 進入一個無限循環，不斷等待用戶輸入問題。

3. 處理用戶輸入：

   • 每當用戶輸入一句話，程序會將這個問題傳遞給 process_query() 函數，自動規劃任務、調用工具、生成回復。

4. 打印結果：

   • 處理完畢后，將結果打印出來。

5. 錯誤處理：

   • 如果在運行過程中出現錯誤（如連接失敗、參數出錯等），程序會捕獲錯誤信息并打印出來，而不會直接崩潰。

async def chat_loop(self):# 初始化提示信息print("\n💬 MCP 客戶端已啟動！輸入 'quit' 退出")while True:try:# 進入主循環中等待用戶輸入query = input("\n你：").strip()if query.lower() == 'quit':break# 處理用戶的提問，并返回結果response = await self.process_query(query)print(f"\n🤖 AI：{response}")except Exception as e:print(f"\n?? 發生錯誤：{str(e)}")

3.2.6 智能規劃工具鏈

概述

plan_tool_usage 函數的作用是讓大模型根據用戶的問題，自動規劃出一組需要使用的工具和調用順序。這個過程確保了用戶的問題可以自動轉化為結構化的工具執行步驟，方便后續依次調用處理。

主要功能

1. 整理當前可用的工具列表：

   • 將可用的工具整理為列表，并寫入系統提示中，引導模型只能從這些工具中選擇。

2. 發送提示內容給大模型：

   • 將提示內容和用戶的問題一起發送給大模型，請求模型生成一個工具調用計劃。

3. 解析大模型的回復：

   • 從大模型的回復中提取出合法的 JSON 內容，并進行解析。如果解析成功，就將結果作為工具調用鏈返回；如果解析失敗，則打印錯誤信息并返回一個空的計劃。

代碼實現

async def plan_tool_usage(self, query: str, tools: list[dict]) -> List[dict]:# 構造系統提示詞 system_prompt# 將所有可用工具組織為文本列表輸入提示中，并明確指出工具名。# 限定使用格式是 JSON，防止大模型輸出錯誤格式的數據。print("\n🤖 正在生成工具調用計劃...")print(json.dumps(tools, ensure_ascii=False, indent=2))tool_list_text = "\n".join([f"{{'function': {{'name': '{tool['function']['name']}', 'description': '{tool['function']['description']}'}}}}"for tool in tools])system_prompt = {"role": "system","content": f"""你是一個智能任務規劃助手，用戶會給出一句自然語言請求。\n你只能從以下工具中選擇（嚴格使用工具名稱）：\n{tool_list_text}\n如果多個工具需要串聯，后續步驟中可以使用【下一步工具名】占位。\n返回格式：JSON 數組，每個對象包含 name 和 arguments 字段。\n不要返回自然語言，不要使用未列出的工具。"""}# 構造對話上下文并調用模型planning_messages = [system_prompt,{"role": "user", "content": query}]response = self.client.chat_completions.create(model=self.model,messages=planning_messages,tool=tools,tool_choice="none")# 提取出模型返回的 JSON 內容content = response.choices[0].message.content.strip()match = re.search(r"(?<json\)\[(.*)\]\s*\)(\s*\)\s*content\)", content)if match:json_text = match.group(1)else:json_text = content# 在解析 JSON 之后返回調用計劃try:plan = json.loads(json_text)return plan if isinstance(plan, list) else []except Exception as e:print(f"? 工具調用鏈規劃失敗：{e}\n原始返回：{content}")return []

3.2.7 關閉資源

概述： 該函數用于在程序結束時關閉并清理所有已打開的資源，確保程序收尾干凈、退出徹底。

功能：

• 調用之前創建的 AsyncExitStack，這個工具會自動管理在程序運行過程中建立的連接，如與服務器的通信通道。

• 通過調用 aclose()，可以確保所有資源都被優雅地釋放，避免出現內存泄漏或卡住進程的問題。

代碼實現：

async def cleanup(self):await self.exit_stack.aclose()

3.2.8 主流程函數

概述： 這是程序的主入口，控制整個客戶端的運行流程。

功能：

• 程序一開始會創建一個 MCPClient 實例，也就是之前封裝的客戶端對象。

• 然后指定服務端腳本的位置，并嘗試連接服務器。

• 一旦連接成功，就進入對話循環，開始等待用戶輸入并處理問題。

• 無論程序中途正常退出還是出錯，最后都會執行 cleanup()，確保所有資源都被安全關閉。

代碼實現：

async def main():server_script_path = "F:\\mcp-project\\server.py"client = MCPClient()try:await client.connect_to_server(server_script_path)await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())

3.3 server.py 的構建

概述： server.py 是服務器端的主要腳本，負責提供新聞搜索、情感分析、郵件發送等基礎工具能力，供客戶端調用。

3.3.1 功能分析

核心工具：

1. search_google_news：

   • 用于在 Google 上搜索相關新聞。

2. analyze_sentiment：

   • 用于對語句進行情感分析。

3. send_email_with_attachment：

   • 用于將本地的文件發送至目標郵箱。

核心功能剖析：

1. 啟動時加載環境變量：

   • Server 會首先加載環境變量，配置必要的 API 密鑰和服務信息。

2. 注冊功能模塊：

   • 注冊一組功能模塊，包括：

     • 調用 Server API 搜索新聞內容。

     • 基于大模型分析文本情感。

     • 發送帶有分析報告的郵件（對應各自的工具函數）。

3. 工具接口暴露：

   • 每個工具均以標準接口形式暴露，客戶端可以根據任務需要按需調用。

4. 標準輸入輸出 (stdio) 模式運行：

   • 程序以標準輸入輸出 (stdio) 模式運行，確保與客戶端實現穩定、實時的交互。

3.3.2?search_google_news() 函數

概述

• 該函數通過 Serper API 使用關鍵詞從 Google 上搜索獲取新聞，返回前五條新聞并保存到本地文件中。

主要內容

1. 申請 Serper API：

   • 需要先申請 Serper 的 API，訪問 Serper 官網 注冊并獲取 API Key。

2. 配置環境變量：

   • 在 .env 文件中配置 Serper 的 API Key。

3. 函數作用：

   • search_google_news() 函數的作用是根據用戶提供的關鍵詞，調用 Serper API 搜索 Google 新聞，并返回前 5 條結果。

4. 執行過程：

   1. 讀取 API 密鑰：從環境變量中獲取用于訪問 Serper API 的密鑰。

   2. 向新聞搜索接口發起請求：將用戶輸入的關鍵詞打包成請求體，發送給 Serper 提供的 Google News 接口。

   3. 提取新聞信息：從返回的數據中提取前 5 條新聞的標題、簡介和鏈接。

   4. 保存為 JSON 文件：將這些新聞內容保存成一個本地 .json 文件，文件名帶有時間戳，方便歸檔。

   5. 返回內容與保存路徑：最后，工具會將獲取到的新聞數據、提示信息和保存路徑一起返回，供客戶端展示或傳遞給下一個工具使用。

代碼實現

@mcp.tool()
async def search_google_news(keyword: str) -> str:# 從環境中獲取 API 密鑰并進行檢查api_key = os.getenv("SERPER_API_KEY")if not api_key:return "? 未配置 SERPER_API_KEY，請在 .env 文件中設置"# 設置請求參數并發送請求url = "https://google.serper.dev/news"headers = {"X-API-KEY": api_key,"Content-Type": "application/json"}payload = {"q": keyword}async with httpx.AsyncClient() as client:response = await client.post(url, headers=headers, json=payload)json_response = response.json()# 檢查數據，并按照格式提取新聞，返回前五條新聞if "news" not in data:return "? 未獲取到搜索結果"articles = [{"title": item.get("title"),"desc": item.get("snippet"),"url": item.get("link")} for item in data["news"][:5]]# 將新聞結果以帶有時間戳命名的 .JSON 格式文件的形式保存在本地指定的路徑output_dir = "/google_news"os.makedirs(output_dir, exist_ok=True)filename = f"google_news_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.json"file_path = os.path.join(output_dir, filename)with open(file_path, "w", encoding="utf-8") as f:json.dump(articles, f, ensure_ascii=False, indent=2)return (f"📰 已獲取與 {keyword} 相關的前5條 Google 新聞。\n"f"📄 已保存到: {file_path}")

詳細步驟

1. 獲取 API 密鑰：

   • 從環境變量中獲取 Serper API 的密鑰。

2. 設置請求參數：

   • 構造請求 URL、請求頭和請求體。

3. 發送請求：

   • 使用 httpx.AsyncClient() 發送 POST 請求。

4. 解析響應：

   • 將響應內容解析為 JSON 格式。

5. 提取新聞信息：

   • 從解析后的 JSON 中提取前 5 條新聞的標題、簡介和鏈接。

6. 保存為 JSON 文件：

   • 將提取的新聞信息保存為一個本地 .json 文件，文件名帶有時間戳。

7. 返回結果：

   • 返回保存路徑和新聞信息。

3.3.3?analyze_sentiment() 函數

概述

analyze_sentiment() 函數用于對一段新聞文本或任意內容進行情感傾向分析，并將分析結果保存為 Markdown 格式的報告文件。

主要內容

1. 功能流程：

   • 讀取大模型配置：從環境變量中加載大模型的 API 密鑰、模型名稱和服務器地址，用于后續調用語言模型。

   • 構造分析指令：將用戶輸入的文本內容整理成標準格式，調用大模型進行情感分析。

   • 獲取模型回復：調用大模型，發送分析指令并獲取分析結果。

   • 生成 Markdown 報告：將原始文本與分析結果整理成結構化的 Markdown 報告，包含時間戳、原文、分析結果。

   • 保存到本地文件：將生成的報告保存到本地，文件名由用戶指定，或默認由程序生成。

   • 返回報告路徑：返回生成的報告文件路徑，方便后續工具（如郵件發送）使用。

代碼實現

@mcp.tool()
async def analyze_sentiment(text: str, filename: str) -> str:# 讀取大模型配置openai_key = os.getenv("DASHSCOPE_API_KEY")client = OpenAI(api_key=openai_key, base_url=os.getenv("BASE_URL"))# 構造情感分析的提示詞prompt = f"請對以下新聞內容進行情感傾向分析，并說明原因。\n\n{text}"# 向模型發送請求，并處理返回的結果response = client.chat.completions.create(model="gpt-3.5-turbo",messages=[{"role": "user", "content": prompt}])result = response.choices[0].message.content.strip()# 生成 Markdown 格式的分析報告，并指定是設置好的輸出目錄markdown = f"**分析時間**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n**原文**:\n\n{text}\n\n**分析結果**:\n\n{result}\n"# 創建輸出目錄output_dir = "./sentiment_report"os.makedirs(output_dir, exist_ok=True)# 生成文件名filename = f"sentiment_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.md"file_path = os.path.join(output_dir, filename)# 將分析結果寫入文件with open(file_path, "w", encoding="utf-8") as f:f.write(markdown)return file_path

詳細步驟

1. 讀取 API 密鑰：

   • 從環境變量中獲取 OpenAI 的 API 密鑰。

2. 構造提示詞：

   • 將用戶輸入的文本內容構造成提示詞，發送給大模型進行情感分析。

3. 發送請求并獲取結果：

   • 使用 OpenAI 客戶端發送請求，并獲取情感分析結果。

4. 生成 Markdown 報告：

   • 將原始文本和分析結果整理成 Markdown 格式的報告。

5. 創建輸出目錄：

   • 創建用于保存報告文件的輸出目錄。

6. 生成文件名：

   • 生成帶有時間戳的文件名。

7. 保存報告文件：

   • 將生成的 Markdown 報告保存到本地文件中。

8. 返回文件路徑：

   • 返回生成的報告文件路徑。

3.3.4?send_email_with_attachment()函數

概述

send_email_with_attachment() 是一個工具類，用于通過獲取本地路徑下的文件，然后將其發送給指定的郵箱。

主要功能

1. 讀取 SMTP 配置：

   • 從環境變量中讀取 SMTP 服務器地址、端口、發件人郵箱和授權碼。

2. 拼接附件路徑并檢查是否存在：

   • 程序會在指定的 sentiment_reports 文件夾中查找附件，如果找不到文件，就會提示失敗。

3. 構造郵件內容：

   • 創建郵件對象，設置主題、正文、收件人等基本信息。

4. 添加附件：

   • 將 Markdown 報告文件讀取為二進制，并以附件形式加入郵件中。

5. 連接 SMTP 服務器并發送郵件：

   • 通過 SSL 安全連接登錄郵箱服務器，并發送郵件。如果發送成功會返回確認信息，如果失敗則返回錯誤說明。

執行流程

1. 讀取發件郵箱配置：

   • 從環境變量中讀取 SMTP 服務器地址、端口、發件人郵箱和授權碼，這些信息是發送郵件的基礎。

2. 拼接附件路徑并檢查是否存在：

   • 程序會在默認的 sentiment_reports 文件夾中查找附件，如果找不到文件，就會提示失敗。

3. 構造郵件內容：

   • 創建郵件對象，設置主題、正文、收件人等基本信息。

4. 添加附件：

   • 將 Markdown 報告文件讀取為二進制，并以附件形式加入郵件中。

5. 連接 SMTP 服務器并發送郵件：

   • 通過 SSL 安全連接登錄郵箱服務器，并發送郵件。如果發送成功會返回確認信息，如果失敗則返回錯誤說明。

代碼實現

@mcp.tool()
async def send_email_with_attachment(to: str, subject: str, body: str, filename: str) -> str:# 讀取并配置 SMTP 相關信息smtp_server = os.getenv("SMTP_SERVER")  # 例如 smtp.qq.comsmtp_port = int(os.getenv("SMTP_PORT", 465))sender_email = os.getenv("EMAIL_USER")sender_pass = os.getenv("EMAIL_PASS")# 獲取附件文件的路徑，并進行檢查是否存在full_path = os.path.abspath(os.path.join("./sentiment_reports", filename))if not os.path.exists(full_path):return f"? 附件路徑無效，未找到文件：{full_path}"# 創建郵件并設置內容msg = EmailMessage()msg['Subject'] = subjectmsg['From'] = sender_emailmsg['To'] = tomsg.set_content(body)# 添加附件并發送郵件try:with open(full_path, "rb") as f:file_name = os.path.basename(full_path)msg.add_attachment(file_data, maintype="application", subtype="octet-stream", filename=file_name)except Exception as e:return f"? 附件讀取失敗：{str(e)}"try:with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:server.login(sender_email, sender_pass)server.send_message(msg)return f"? 郵件已成功發送給 {to}，附件路徑：{full_path}"except Exception as e:return f"? 郵件發送失敗：{str(e)}"

4、測試

在運行的時候只需要運行client.py就可以運行整個項目了。