一、实验背景与核心目标
本次实验为软件工程专业课程实训项目,核心任务是开发一套可对接小红书 MCP(内容平台)服务的 Python 包,需封装小红书登录会话管理、图文 / 视频内容发布、笔记检索、用户信息查询等核心功能。实验要求:
保证代码的向后兼容性,适配团队前期测试代码的调用习惯;
实现核心功能的高内聚低耦合,便于小组协作维护;
最终产出的包可通过pip本地安装,支持命令行调用和 Web 服务两种使用方式。
二、开发环境与目录结构规划
(一)开发环境搭建
本次开发选定 Python 3.9(兼容小红书开放平台 SDK 最低版本要求),先通过pip freeze > requirements.txt反向梳理核心依赖,最终确定核心依赖包及版本:
txt
requests==2.31.0  # 对接小红书MCP HTTP接口
Flask==2.3.3      # 搭建Web服务端
pytest==7.4.2     # 单元测试
python-dotenv==1.0.0 # 环境变量管理
cryptography==41.0.5 # 登录会话加密
执行pip install -r requirements.txt完成环境配置后,参考 PEP 621 规范编写pyproject.toml文件,定义包名称、版本、作者、依赖等核心信息,确保后续可通过pip install .本地打包安装。
(二)目录结构规划与落地
小组初期对目录结构有两种分歧方案:方案一是按功能模块(登录、发布、搜索)划分,方案二是按工程分层(入口层、服务层、管理层、工具层)划分。经讨论,选择方案二(更符合工业界 “分层架构” 思想,便于后续迭代),最终落地的目录结构如下:
plaintext
xiaohongshu_mcp_python/  # 包根目录
├── __init__.py          # 包入口,统一接口暴露与兼容性处理
├── main.py              # 程序入口(命令行+主函数)
├── server/              # Web服务模块
│   ├── __init__.py
│   ├── app.py           # Flask服务实例初始化
│   ├── mcp.py           # MCP核心实例封装
│   ├── mcp_tools.py     # MCP工具函数(登录、发布、搜索等)
│   └── server_run.py    # Web服务启动逻辑
├── services/            # 核心业务服务层
│   ├── __init__.py
│   └── XiaohongshuService.py # 封装MCP底层接口,统一业务逻辑
├── managers/            # 资源管理层
│   ├── __init__.py
│   └── user_session_manager.py # 登录会话管理(创建、校验、清理)
├── tests/               # 测试目录
│   ├── __init__.py
│   ├── test_login.py    # 登录功能测试
│   ├── test_publish.py  # 内容发布测试
│   └── test_search.py   # 内容检索测试
├── requirements.txt     # 依赖管理
└── pyproject.toml       # 包构建配置
搭建过程中需注意:每个子目录必须添加空的__init__.py文件(确保 Python 识别为可导入模块);为tests目录配置conftest.py,统一初始化测试用 MCP 实例、测试账号等夹具,减少测试代码冗余。
三、核心模块开发过程
(一)managers 模块:用户会话管理开发
开发目标
封装登录会话的创建、校验、清理逻辑,解决 “会话状态丢失”“多线程会话冲突” 问题,为登录功能提供底层支撑。
核心实现思路
采用dict存储会话信息(键为唯一 UUID 会话 ID,值为包含 token、创建时间、过期时间的字典),并通过threading.Lock加锁,避免多线程操作字典导致的数据冲突;
会话创建时生成 UUID 作为会话 ID,调用小红书 MCP 登录接口获取 token,同时记录创建时间(设置 2 小时过期阈值);
会话校验分两步:先判断是否过期,再调用 MCP 接口验证 token 有效性;
会话清理时,删除字典中会话信息的同时调用 MCP 接口注销 token,避免无效 token 占用资源。
问题与解决
问题 1:多线程调用get_user_session_manager()时,出现会话信息被覆盖的情况。
排查:未加线程锁,多线程同时写入会话字典导致数据竞争。
解决:在会话增删改查方法中添加threading.Lock(),确保同一时间仅一个线程操作会话字典,核心代码片段:
python
运行
# managers/user_session_manager.py
import threading
_session_lock = threading.Lock()
_user_sessions = {}

def create_session(user_id):
    with _session_lock:
        session_id = str(uuid.uuid4())
        _user_sessions[session_id] = {
            "user_id": user_id,
            "token": "",
            "create_time": time.time(),
            "expire_seconds": 7200
        }
    return session_id
问题 2:会话过期判断逻辑错误,未过期会话被误判为过期。
排查:时间对比逻辑写反(time.time() - create_time < expire_seconds),导致 “未过期” 被判定为 “过期”。
解决:修正逻辑为time.time() - create_time > expire_seconds,并补充单元测试覆盖该场景。
(二)services 模块:XiaohongshuService 类开发
开发目标
封装小红书 MCP 底层 HTTP 接口,统一请求头、参数格式、异常处理,为上层模块提供标准化业务调用接口。
核心实现思路
初始化方法接收 MCP 配置(app_key、app_secret、base_url),封装小红书要求的 HMAC-SHA256 签名算法(接口调用必须的鉴权逻辑);
为每个核心功能定义独立方法(如publish_content()/publish_video()/search_feeds()),统一处理参数拼接、请求发送、响应解析;
自定义异常XiaohongshuAPIException,捕获requests库的网络异常、接口返回的业务异常,统一返回错误码和错误信息,便于上层模块处理。
问题与解决
问题 1:调用发布接口时,MCP 返回 “签名无效”。
排查:签名算法未按小红书要求的 “参数名 ASCII 升序拼接” 规则实现,且未对拼接字符串做 HMAC-SHA256 加密。
解决:参考 MCP 开发文档重构签名算法,核心代码:
python
运行
# services/XiaohongshuService.py
import hmac
import hashlib

def _generate_sign(self, params):
    # 按ASCII升序排序参数
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 拼接参数为字符串
    param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
    # 拼接app_secret并加密
    sign_str = f"{param_str}&app_secret={self.app_secret}"
    sign = hmac.new(
        self.app_secret.encode("utf-8"),
        sign_str.encode("utf-8"),
        digestmod=hashlib.sha256
    ).hexdigest()
    return sign
问题 2:发布视频时出现 “上传超时” 和 “文件格式不支持”。
排查:未设置请求超时时间(默认 5 秒),且视频编码为 avc1(小红书要求 h264 编码)。
解决:① 请求中添加timeout=30延长超时时间;② 在publish_video()方法中添加编码校验,提示用户转换编码后上传。
(三)server 模块:Web 服务与工具函数开发
开发目标
搭建 Flask Web 服务对外提供 HTTP 接口;封装 MCP 工具函数,适配团队前期测试代码的调用习惯。
核心实现思路
app.py中初始化 Flask 实例,配置 CORS 跨域、JSON 请求解析,注册/login//publish//search等路由;
mcp.py中实现 MCP 实例单例模式(避免重复初始化配置),读取.env文件中的 app_key、app_secret;
mcp_tools.py封装工具函数(如xiaohongshu_start_login_session),内部调用 managers 层会话管理和 services 层 API,对外暴露极简接口(无需关注底层逻辑)。
问题与解决
问题 1:Flask 路由调用会话创建函数时,报 “找不到 MCP 实例”。
排查:MCP 实例未全局初始化,每个路由请求都重新创建实例,导致配置未加载。
解决:实现单例模式确保全局唯一实例,核心代码:
python
运行
# server/mcp.py
class MCPInstance:
    _instance = None
    _lock = threading.Lock()

    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    from dotenv import load_dotenv
                    config = load_dotenv()
                    from ..services import XiaohongshuService
                    cls._instance = XiaohongshuService(
                        app_key=config["APP_KEY"],
                        app_secret=config["APP_SECRET"],
                        base_url=config["BASE_URL"]
                    )
        return cls._instance

mcp = MCPInstance.get_instance()
问题 2:工具函数命名不统一(如 “发布图文” 有publish_content/publish_image两种叫法)。
解决:先保留规范命名publish_content,后续在__init__.py中添加别名兼容旧习惯。
(四)main 模块:程序入口开发
开发目标
提供命令行(CLI)和主函数两种入口,支持python -m xiaohongshu_mcp_python调用,或代码中调用main()启动。
核心实现思路
cli_main()使用argparse解析命令行参数(如--login启动登录、--publish发布内容);
main()函数根据参数调用对应工具函数,输出结构化执行结果;
添加if __name__ == "__main__":,支持直接运行main.py。
问题与解决
问题:--publish参数无法识别多个图文文件路径。
排查:argparse未配置nargs参数,默认仅接收单个参数。
解决:配置parser.add_argument("--publish", nargs="+", help="图文文件路径"),支持多路径传入。
(五)init.py:包入口与兼容性封装
开发目标
统一对外接口,简化调用路径,保证向后兼容,符合 Python 包规范。
核心实现思路
定义包元信息(__version__/__author__/__description__),遵循语义化版本规则;
从子模块导入核心对象(main、app、XiaohongshuService 等),提供 “一层式” 导入路径;
导出 mcp_tools.py 工具函数,适配前期测试代码的路径习惯;
定义函数别名解决命名分歧;
配置__all__列表,明确公共接口(避免内部对象被外部导入)。
问题与解决
问题:导入子模块时触发 Pylint “循环导入” 误报。
排查:实际无循环导入(仅导入对象而非模块),是工具误判。
解决:在 Pylint 配置文件添加disable=cyclic-import,同时确认导入顺序(先导入模块,再导入对象)。
四、测试与调试过程
(一)单元测试
基于 pytest 编写用例,覆盖核心场景:
会话管理:测试创建、校验、过期、清理全流程;
API 调用:使用小红书测试环境 app_key,验证登录、发布、搜索接口返回;
兼容性:测试旧路径(from xiaohongshu_mcp_python.server.mcp_tools import ...)和新路径(from xiaohongshu_mcp_python import ...)是否均能调用。
(二)集成测试
启动 Flask 服务后,用 Postman 调用接口验证端到端流程:
调用/api/login获取会话 ID;
调用/api/publish传入会话 ID 和图文参数,验证内容发布到测试环境;
调用/api/search检索发布内容,验证返回结果正确性。
(三)bug 修复记录
1    多线程会话信息被覆盖    加线程锁确保会话字典操作原子性    已修复
2    发布接口签名无效    按 MCP 要求重构签名算法    已修复
3    命令行无法识别多文件路径    配置 argparse 的 nargs 参数为 +    已修复
4    视频上传超时 / 格式报错    延长超时时间 + 添加编码校验    已修复
五、开发总结与后续计划
(一)开发总结
本次开发从目录规划到模块落地,完整实现了小红书 MCP Python 包核心功能,核心收获:
掌握 Python 包分层架构设计,实现 “管理层→服务层→工具层→入口层” 低耦合设计,便于扩展;
学会向后兼容的实现方法(统一接口、别名、路径转发),兼顾团队前期开发习惯;
解决多线程、API 签名、超时处理等实际问题,提升代码健壮性;
遵循 PEP8/PEP621 规范,保证代码可读性与可维护性。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐