目录

1. 服务层功能完善
   • 图文发布功能实现
   • 视频发布功能实现
2. MCP工具接口开发
   • FastMCP集成
   • 工具接口设计与优化
3. 浏览器自动化核心功能
   • 浏览器管理器优化
   • 页面控制器实现
4. 本周开发总结

服务层功能完善

图文发布功能实现

1. 开发过程

图文发布功能是小红书MCP核心能力之一，需兼顾本地/网络图片处理、发布流程闭环、参数合法性校验三大核心诉求，具体实现如下：

• 图片处理逻辑设计：
  针对本地图片和网络图片两种输入形式，设计ImageProcessor类并实现process_images核心方法。其中网络图片下载模块支持HTTP/HTTPS协议，新增图片格式验证逻辑，仅放行JPG、PNG、GIF、WebP格式文件。
• 发布流程实现：
  封装_publish_content方法统一管理发布全流程，标准化执行步骤：启动浏览器 → 导航至发布页 → 图片上传 → 内容填写 → 发布触发 → 结果等待，全流程嵌入错误捕获与状态反馈机制。
• 参数验证逻辑：
  严格约束发布内容参数：标题长度限制1-20字、图片数量1-9张、内容支持文本+表情混合格式，前置校验避免平台端异常。

2. 核心问题与解决方案

问题1：网络图片下载失败

• 现象：HTTPS链接图片下载成功率低，偶发超时/无响应；
• 根因：网络超时、目标服务器反爬、格式不兼容；
• 解决方案：
  ① 实现3次重试机制，失败后间隔1秒重试；
  ② 模拟浏览器User-Agent请求头；
  ③ 统一转换图片为JPG格式；
• 核心代码：

async def download_image(self, url, timeout=30):
    """下载网络图片，支持重试"""
    for attempt in range(3):
        try:
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }
            async with aiohttp.ClientSession() as session:
                async with session.get(url, headers=headers, timeout=timeout) as response:
                    if response.status == 200:
                        content = await response.read()
                        return content
        except Exception as e:
            logger.warning(f"下载图片失败 (尝试 {attempt+1}/3): {e}")
            await asyncio.sleep(1)
    return None

问题2：图片上传超时

• 现象：多图上传时频繁触发超时，发布流程中断；
• 根因：图片体积过大、上传速度慢、固定等待时间不足；
• 解决方案：
  ① 图片压缩限制单张体积；
  ② 动态调整等待时间（基础10秒+每张图片5秒，上限60秒）；
  ③ 分批上传优化传输效率；
• 核心代码：

async def _wait_for_upload_complete(self, image_count):
    """等待图片上传完成，根据图片数量调整等待时间"""
    wait_time = min(60, 10 + image_count * 5)
    start_time = time.time()
    while time.time() - start_time < wait_time:
        if await self._is_upload_complete():
            return True
        await asyncio.sleep(2)
    return False

视频发布功能实现

1. 开发过程

视频发布模块聚焦格式兼容性、流程标准化、状态监控三大方向：

• 视频格式验证：支持MP4/MOV/AVI格式，校验文件存在性、可读性，建议单文件不超过1GB；
• 发布流程：实现_publish_video方法，流程为：启动浏览器 → 发布页导航 → 视频发布入口选择 → 视频上传 → 平台处理等待 → 内容填写 → 发布；
• 状态监控：实时监控视频处理进度，捕获转码失败场景，输出结构化错误信息。

2. 核心问题与解决方案

问题1：视频处理时间过长

• 现象：视频上传后平台处理超时，发布流程终止；
• 根因：视频体积大、平台队列拥堵、网络不稳定；
• 解决方案：
  ① 延长最大等待时间至10分钟；
  ② 每10秒轮询处理状态；
  ③ 超时后执行优雅降级策略；
• 核心代码：

async def _wait_for_video_processing(self):
    """等待视频处理完成"""
    max_wait_time = 600  # 10分钟
    start_time = time.time()
    
    while time.time() - start_time < max_wait_time:
        status = await self._get_video_processing_status()
        if status == "completed":
            return True
        elif status == "failed":
            return False
        await asyncio.sleep(10)
    
    logger.warning("视频处理超时")
    return False

问题2：视频格式不兼容

• 现象：部分MP4视频上传后无法转码，发布失败；
• 根因：编码格式不支持、分辨率/比特率过高；
• 解决方案：
  ① 检测视频编码（仅支持h264/h265/mpeg4）；
  ② 限制分辨率≤1920x1080；
  ③ 输出标准化兼容检测报告；
• 核心代码：

def check_video_compatibility(self, video_path):
    """检查视频兼容性"""
    try:
        import ffmpeg
        probe = ffmpeg.probe(video_path)
        video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None)
        
        if not video_stream:
            return False, "未找到视频流"
        
        codec_name = video_stream['codec_name']
        if codec_name not in ['h264', 'h265', 'mpeg4']:
            return False, f"不支持的视频编码: {codec_name}"
        
        width = int(video_stream['width'])
        height = int(video_stream['height'])
        if width > 1920 or height > 1080:
            return False, "视频分辨率过高，建议不超过1920x1080"
        
        return True, "视频格式兼容"
    except Exception as e:
        return False, f"检查视频格式时出错: {str(e)}"

MCP工具接口开发

FastMCP集成

1. 开发过程

完成FastMCP框架的全量集成，实现小红书能力的标准化工具封装：

• 服务初始化：创建FastMCP实例，配置服务名称/版本，完成工具函数注册；
• 工具接口设计：

  工具名称	功能描述
  xiaohongshu_publish	图文内容发布
  xiaohongshu_publish_video	视频内容发布
  xiaohongshu_search	小红书内容搜索
  xiaohongshu_get_user_info	用户信息获取

• 参数标准化：定义统一参数格式，实现入参校验，输出完整接口文档。

2. 核心问题与解决方案

问题1：工具注册失败

• 现象：注册时触发AttributeError，工具无法加载；
• 根因：FastMCP版本不兼容、装饰器使用错误、参数类型注解不规范；
• 解决方案：
  ① 升级至最新版FastMCP；
  ② 修正装饰器使用方式，规范参数类型注解；
• 核心代码：

# 正确的工具注册方式
@mcp.tool
async def xiaohongshu_publish(
    title: str,
    content: str,
    images: List[str],
    tags: Optional[List[str]] = None,
    username: Optional[str] = None
) -> dict:
    """
    发布图文内容到小红书
    """
    # 业务逻辑实现

问题2：流式输出实现

• 现象：工具无进度反馈，用户体验差；
• 根因：缺少上下文对象、无进度报告机制、响应格式不规范；
• 解决方案：
  ① 集成Context上下文对象；
  ② 分阶段上报进度（10%/30%/50%/70%/100%）；
• 核心代码：

async def xiaohongshu_publish(..., context: Optional[Context] = None):
    """
    发布图文内容，支持流式输出
    """
    if context:
        await context.report_progress(progress=10, total=100, message="开始发布")
    
    # 处理图片
    if context:
        await context.report_progress(progress=30, total=100, message="处理图片中")
    
    # 上传图片
    if context:
        await context.report_progress(progress=50, total=100, message="上传图片中")
    
    # 填写内容
    if context:
        await context.report_progress(progress=70, total=100, message="填写内容中")
    
    # 发布完成
    if context:
        await context.report_progress(progress=100, total=100, message="发布完成")

浏览器自动化核心功能

浏览器管理器优化

1. 开发过程

构建BrowserManager类实现浏览器全生命周期管理，核心能力包括：

• 多浏览器支持：兼容Chromium/Firefox/WebKit，支持无头模式、用户数据目录配置；
• 反检测措施：集成playwright-stealth插件，实现浏览器指纹随机化、人类行为模拟（随机延迟）；
• 会话管理：自动保存/加载Cookies，支持会话恢复，实现浏览器实例池调度。

2. 核心问题与解决方案

问题1：浏览器被检测为自动化工具

• 现象：登录时触发滑块验证，甚至账号封禁；
• 根因：浏览器指纹特征明显、操作无随机延迟、缺失真实浏览器特性；
• 解决方案：
  ① 集成playwright-stealth插件模拟真实浏览器环境；
  ② 添加随机操作延迟；
  ③ 禁用自动化特征标识；
• 核心代码：

async def start(self):
    """启动浏览器，应用反检测措施"""
    self._playwright = await async_playwright().start()
    
    if self.browser_type == "chromium":
        browser_type = self._playwright.chromium
    elif self.browser_type == "firefox":
        browser_type = self._playwright.firefox
    else:
        browser_type = self._playwright.webkit
    
    self._browser = await browser_type.launch(
        headless=self.headless,
        executable_path=self.executable_path,
        args=[
            "--disable-blink-features=AutomationControlled",
            "--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        ]
    )
    
    self._context = await self._browser.new_context()
    await Stealth(self._context)  # 应用反检测插件
    self._page = await self._context.new_page()
    
    if self.cookie_storage:
        await self.load_cookies()

问题2：浏览器启动失败

• 现象：Playwright无法定位浏览器可执行文件；
• 根因：浏览器未安装、路径配置错误、权限不足；
• 解决方案：
  ① 自动检测系统内置浏览器（Edge/Chrome）；
  ② 提供默认路径兜底；
  ③ 输出精准错误提示；
• 核心代码：

def _find_browser_executable(self):
    """自动查找浏览器可执行文件"""
    if self.executable_path and os.path.exists(self.executable_path):
        return self.executable_path
    
    if platform.system() == "Windows":
        # 查找Edge浏览器
        edge_paths = [
            r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
            r"C:\Program Files\Microsoft\Edge\Application\msedge.exe"
        ]
        for path in edge_paths:
            if os.path.exists(path):
                return path
        
        # 查找Chrome浏览器
        chrome_paths = [
            r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
            r"C:\Program Files\Google\Chrome\Application\chrome.exe"
        ]
        for path in chrome_paths:
            if os.path.exists(path):
                return path
    
    return None

页面控制器实现

1. 开发过程

实现页面操作标准化封装，核心能力：

• 元素定位：支持XPath/CSS选择器，集成智能等待机制；
• 页面操作：封装点击、输入、上传等操作，支持导航/刷新；
• 异常处理：捕获元素未找到、操作超时等异常，输出结构化错误日志。

2. 核心问题与解决方案

问题1：元素定位不稳定

• 现象：XPath选择器偶发失效，操作失败；
• 根因：页面动态渲染、XPath路径过深、元素加载延迟；
• 解决方案：
  ① 使用相对XPath路径；
  ② 实现元素智能等待（最长30秒）；
  ③ 多策略定位兜底；
• 核心代码：

async def wait_for_element(self, selector, timeout=30):
    """等待元素出现"""
    try:
        await self.page.wait_for_selector(
            selector,
            timeout=timeout * 1000
        )
        return True
    except Exception as e:
        logger.warning(f"等待元素 {selector} 失败: {e}")
        return False

async def click_element(self, selector, timeout=30):
    """点击元素，支持智能等待"""
    if await self.wait_for_element(selector, timeout):
        try:
            await self.page.click(selector)
            return True
        except Exception as e:
            logger.error(f"点击元素 {selector} 失败: {e}")
            return False
    return False

问题2：文件上传失败

• 现象：图片/视频上传时，文件选择对话框处理异常；
• 根因：上传元素定位错误、路径格式不兼容、输入框隐藏；
• 解决方案：
  ① 精准定位文件输入元素；
  ② 兼容多系统路径格式；
  ③ 使用Playwright原生上传API；
• 核心代码：

async def upload_file(self, selector, file_path):
    """上传文件"""
    try:
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"文件不存在: {file_path}")
        
        input_element = await self.page.query_selector(selector)
        if not input_element:
            raise Exception("未找到文件输入元素")
        
        await input_element.set_input_files(file_path)
        return True
    except Exception as e:
        logger.error(f"上传文件失败: {e}")
        return False

本周开发总结

本周完成小红书MCP服务三大核心模块开发：

1. 服务层：落地图文/视频发布全流程，解决图片下载/上传超时、视频格式兼容等核心问题；
2. MCP接口：完成FastMCP集成，实现标准化工具注册与流式输出，支撑上层调用；
3. 浏览器自动化：优化浏览器管理器与页面控制器，攻克反检测、元素定位不稳定等关键问题。