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 或 .js
    is_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 用 node
    command = "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_filename

        result = await self.session.call_tool(tool_name, tool_args)
        tool_output[tool_name] = result.content[0].text
        messages.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.com
    smtp_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'] = subject
    msg['From'] = sender_email
    msg['To'] = to
    msg.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就可以运行整个项目了。