xiaoli-cli/开发文档/开发文档.md

小狸 Pro-CLI 开发文档

版本：2.0 | 更新日期：2026年3月

---

目录

1. 系统架构概述
2. AI引擎开发规范
3. Plugin协议开发
4. Skill协议开发
5. MCP协议支持
6. WebSocket API
7. 已加载模块列表
8. 最佳实践
9. 调试与测试
10. 常见问题

---

1. 系统架构概述

1.1 三大协议支持

小狸 Pro-CLI 的工具系统支持三种协议：

协议	目录	说明
Plugin 协议	plugins/	小狸自有协议，简单易用
Skill 协议	skills/	Agent Skills 开源协议，支持 Markdown 和 Python 格式
MCP 协议	-	Model Context Protocol，Anthropic/OpenAI 标准

1.2 统一工具管理器

unified_tool_manager.py 协调三种协议，提供统一的工具调用接口：

from unified_tool_manager import UnifiedToolManager

manager = UnifiedToolManager(cli_instance)
manager.initialize("plugins", "skills")

# Plugin 格式调用
result = manager.execute("file_manager", "read test.txt")

# MCP 格式调用
result = manager.mcp_call("file_manager", {"operation": "read", "path": "test.txt"})

1.3 路由策略

• 优先使用 Skill 协议（如果存在同名技能）
• 回退到 Plugin 协议
• 支持强制指定协议
• MCP 格式调用统一入口

---

2. AI引擎开发规范

2.1 文件与类命名规范

规范	说明
文件命名	必须以 _engine.py 结尾
类命名	使用驼峰命名，如 Qwen3CoderPlusAI

示例：

• 文件：qwen3-coder-plus_engine.py
• 类名：Qwen3CoderPlusAI

2.2 引擎配置声明

class MyEngineAI:
    # 配置属性
    requires_api_key = True  # 是否需要API密钥
    default_base_url = "https://api.example.com/v1"
    default_model = "my-model"
    
    def __init__(self):
        self.name = "my_engine"
        # 初始化代码...

2.3 核心接口

def generate_response(self, user_input, tool_results=None, system_prompt=None):
    """
    生成AI响应
    
    参数:
        user_input: 用户输入文本
        tool_results: 工具执行结果（可选）
        system_prompt: 系统提示词（可选，由主程序生成）
    
    返回:
        AI响应字符串
    """
    pass

def get_help_info(self):
    """返回引擎帮助信息"""
    return "引擎帮助信息..."

2.4 禁止设置超时

# ✅ 正确
response = requests.post(url, json=body, headers=headers, timeout=None)

# ❌ 错误
response = requests.post(url, json=body, headers=headers, timeout=30)

2.5 SDK快速开发

使用 SDK，只需几行代码就能完成引擎开发：

from SDK_openAI import *

class MyAIAI:
    def __init__(self):
        engine_Online("on")  # 云模式
        ai_name("my_ai")
        ai_api_url("https://api.example.com/v1")
        ai_api_key("your_api_key")
        ai_model_id("gpt-3.5-turbo")
        
        self.name = "my_ai"
        self.cli = None
        self.shared_conversation_history = None
        
    def set_cli(self, cli):
        set_cli(cli)
        self.cli = cli
        
    def generate_response(self, user_input, tool_results=None, system_prompt=None):
        set_shared_history(self.shared_conversation_history)
        return generate_response(user_input, tool_results, system_prompt)

2.6 SDK方法说明

方法	说明
engine_Online("on"/"off")	云模式/本地模式
ai_name(name)	设置引擎名称
ai_api_url(url)	设置API地址
ai_api_key(key)	设置API密钥
ai_model_id(model)	设置模型名称
ai_thinking_engine(...)	深度思考引擎配置

2.7 深度思考引擎配置

适用于 QwQ、DeepSeek-R1 等会输出思考过程的模型：

ai_thinking_engine(
    enable=True,           # 启用深度思考
    start_marker="<thinking>",  # 思考开始标记
    end_marker="</thinking>",   # 思考结束标记
    show_thinking=True     # 显示思考过程
)

---

3. Plugin协议开发

3.1 基本结构

class Plugin:
    """插件描述"""
    
    def __init__(self):
        self.usage = "使用方法说明"
        self.cli = None
    
    def set_cli(self, cli):
        """设置CLI引用（可选）"""
        self.cli = cli
    
    def get_tool_info(self):
        """返回工具信息（必需）"""
        return {
            "name": "tool_name",
            "description": "工具描述",
            "keywords": ["关键词1", "关键词2"],
            "usage": self.usage
        }
    
    def get_mcp_definition(self):
        """返回MCP格式定义（可选）"""
        return {
            "name": "tool_name",
            "description": "工具描述",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "operation": {"type": "string"},
                    "path": {"type": "string"}
                },
                "required": ["operation"]
            }
        }
    
    def convert_mcp_args(self, arguments):
        """转换MCP参数（可选）"""
        return f"{arguments.get('operation', '')} {arguments.get('path', '')}"
    
    def handle(self, args: str) -> str:
        """处理工具调用（必需）"""
        try:
            # 实现逻辑
            return "处理结果"
        except Exception as e:
            return f"错误: {str(e)}"

3.2 完整示例：文件管理器

import os
from typing import Dict, Any, List

class Plugin:
    """文件管理器插件"""
    
    def __init__(self):
        self.usage = """文件管理器使用方法：
file_manager <操作> <参数>

操作类型:
- list <路径>           - 列出目录内容
- read <文件路径>       - 读取文件内容
- write <文件路径> -c "内容" - 写入文件
- delete <路径>         - 删除文件或目录
"""
        self.cli = None
    
    def set_cli(self, cli):
        self.cli = cli
    
    def get_tool_info(self):
        return {
            "name": "file_manager",
            "description": "文件和目录管理工具",
            "keywords": ["文件", "目录", "管理", "file"],
            "usage": self.usage
        }
    
    def get_mcp_definition(self) -> Dict[str, Any]:
        return {
            "name": "file_manager",
            "description": "文件和目录管理工具",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "operation": {
                        "type": "string",
                        "enum": ["list", "read", "write", "delete"],
                        "description": "操作类型"
                    },
                    "path": {"type": "string", "description": "路径"},
                    "content": {"type": "string", "description": "内容"}
                },
                "required": ["operation"]
            }
        }
    
    def convert_mcp_args(self, arguments: Dict[str, Any]) -> str:
        op = arguments.get("operation", "")
        path = arguments.get("path", "")
        content = arguments.get("content", "")
        if content:
            return f'{op} {path} -c "{content}"'
        return f"{op} {path}"
    
    def handle(self, args: str) -> str:
        parts = args.strip().split(maxsplit=2)
        if not parts:
            return "错误：请提供操作类型"
        
        operation = parts[0].lower()
        
        if operation == "list":
            return self._op_list(parts[1:] if len(parts) > 1 else ["."])
        elif operation == "read":
            return self._op_read(parts[1:] if len(parts) > 1 else [])
        elif operation == "write":
            return self._op_write(parts[1:])
        elif operation == "delete":
            return self._op_delete(parts[1:])
        
        return f"错误：不支持的操作 '{operation}'"
    
    def _op_list(self, args: List[str]) -> str:
        path = args[0] if args else "."
        if not os.path.exists(path):
            return f"错误：路径不存在: {path}"
        # 实现列出目录逻辑...
        return f"目录内容: {path}"
    
    def _op_read(self, args: List[str]) -> str:
        if not args:
            return "错误：请提供文件路径"
        # 实现读取文件逻辑...
        return "文件内容..."
    
    def _op_write(self, args: List[str]) -> str:
        if len(args) < 2:
            return "错误：请提供文件路径和内容"
        # 实现写入文件逻辑...
        return "写入成功"
    
    def _op_delete(self, args: List[str]) -> str:
        if not args:
            return "错误：请提供路径"
        # 实现删除逻辑...
        return "删除成功"

3.3 参数格式规范

多操作插件格式：

<操作类型> <目标对象> [可选内容]

示例：read file.txt、write file.txt 内容

单参数插件格式：

<待处理内容>

示例：2+3、hello world

---

4. Skill协议开发

4.1 Python格式技能

# skills/my_skill/__init__.py
from skills.base import Skill, SkillResult

class MySkill(Skill):
    """我的技能"""
    
    name = "my_skill"
    description = "技能描述"
    tags = ["标签1", "标签2"]
    
    def get_tool_definition(self):
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        "operation": {"type": "string"}
                    }
                }
            }
        }
    
    def run(self, **kwargs) -> SkillResult:
        try:
            operation = kwargs.get("operation", "")
            # 实现逻辑
            return SkillResult(
                success=True,
                message="执行成功",
                data={"result": "..."}
            )
        except Exception as e:
            return SkillResult(
                success=False,
                error=str(e)
            )

4.2 Markdown格式技能

# skills/my_skill/SKILL.md

---
name: my_skill
description: 技能描述
tags: [标签1, 标签2]
---

# 技能说明

这是技能的详细使用说明...

## 使用方法

1. 步骤一
2. 步骤二

---

5. MCP协议支持

5.1 MCP工具定义格式

{
    "name": "tool_name",
    "description": "工具描述",
    "inputSchema": {
        "type": "object",
        "properties": {
            "param1": {"type": "string", "description": "参数说明"},
            "param2": {"type": "integer", "description": "参数说明"}
        },
        "required": ["param1"]
    }
}

5.2 MCP调用格式

请求：

{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
        "name": "file_manager",
        "arguments": {
            "operation": "read",
            "path": "test.txt"
        }
    }
}

响应：

{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "content": [
            {"type": "text", "text": "文件内容..."}
        ],
        "isError": false
    }
}

5.3 统一工具管理器MCP接口

# 获取MCP工具列表
tools = manager.get_mcp_tools()

# MCP格式调用
result = manager.mcp_call("file_manager", {"operation": "read", "path": "test.txt"})

# 处理MCP请求
response = manager.handle_mcp_request({
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {...}
})

---

6. WebSocket API

6.1 消息类型

类型	方向	说明
heartbeat	双向	心跳检测
auth	客户端→服务端	认证请求
auth_success / auth_failed	服务端→客户端	认证结果
message	双向	文本消息
file	客户端→服务端	文件上传
image	服务端→客户端	图片发送
tool_status	服务端→客户端	工具状态

6.2 消息格式示例

认证请求：

{"type": "auth", "password": "your_password"}

发送消息：

{"type": "message", "content": "你好"}

上传文件：

{
    "type": "file",
    "file": "base64_encoded_data",
    "filename": "example.txt",
    "text": "附加说明"
}

6.3 启动远程服务

# 本地模式
/remote start 9079 your_password

# 代理模式
/remote proxy relay.example.com:9079 server_pwd 12345 pc_pwd

---

7. 已加载模块列表

7.1 AI引擎

引擎名	类型	说明
ollama	本地	Ollama本地模型，无需API密钥
glm	云端	智谱GLM
spark	云端	讯飞星火
qwq	云端	QwQ深度思考引擎
qwen	云端	通义千问

7.2 插件

插件名	功能
file_manager	文件管理（14种操作）
ai_search	AI搜索
audio_player	音频播放
cmd_executor	命令执行
frontend_tester	前端测试
glados_tts_plugin	GLaDOS语音合成
napcat_qq	QQ机器人
network_tools	网络工具
ollama_image	Ollama图像
send_image	发送图片
speech_recognition	语音识别
task_manager	任务管理
tool_search	工具搜索

---

8. 最佳实践

8.1 安全性

# ❌ 避免使用危险函数
result = eval(user_input)  # 危险！

# ✅ 安全的替代方案
import ast
try:
    tree = ast.parse(user_input, mode='eval')
    result = eval(compile(tree, '<string>', 'eval'))
except:
    return "无效表达式"

8.2 输入验证

def handle(self, args: str) -> str:
    # 参数验证
    if not args:
        return "错误：参数不能为空"
    
    # 长度限制
    if len(args) > 10000:
        return "错误：输入过长"
    
    # 格式验证
    if not re.match(r'^[\w\s\-\.]+$', args):
        return "错误：包含非法字符"
    
    # 继续处理...

8.3 错误处理

def handle(self, args: str) -> str:
    try:
        # 业务逻辑
        return "结果"
    except FileNotFoundError:
        return "错误：文件不存在"
    except PermissionError:
        return "错误：权限不足"
    except Exception as e:
        return f"错误: {str(e)}"

8.4 资源管理

# 使用上下文管理器
with open(file_path, 'r', encoding='utf-8') as f:
    content = f.read()

# 或确保资源释放
try:
    f = open(file_path, 'r')
    content = f.read()
finally:
    f.close()

---

9. 调试与测试

9.1 日志记录

def handle(self, args: str) -> str:
    if self.cli:
        print(f"[DEBUG] 插件被调用: {args}")
    # 处理逻辑
    return "结果"

9.2 单元测试

import unittest
from plugins.file_manager import Plugin

class TestFileManager(unittest.TestCase):
    def setUp(self):
        self.plugin = Plugin()
    
    def test_get_tool_info(self):
        info = self.plugin.get_tool_info()
        self.assertIn("name", info)
        self.assertEqual(info["name"], "file_manager")
    
    def test_handle_list(self):
        result = self.plugin.handle("list .")
        self.assertIsInstance(result, str)

if __name__ == "__main__":
    unittest.main()

9.3 集成测试

# integration_test.py
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))

from ai_cli import AICLI

def test_integration():
    cli = AICLI()
    # 测试插件调用...

---

10. 常见问题

Q1: 插件不加载？

检查以下几点：

• 文件是否放在 plugins/ 目录
• 类名是否为 Plugin
• 是否实现了 get_tool_info() 和 handle() 方法
• 文件名是否为 .py 结尾

Q2: 插件不被调用？

检查：

• 关键词是否匹配用户输入
• keywords 列表是否正确设置
• 插件是否成功加载（查看启动日志）

Q3: MCP格式调用失败？

确保：

• 实现了 get_mcp_definition() 方法
• 实现了 convert_mcp_args() 方法（如果需要）
• inputSchema 格式正确

Q4: 如何调试？

• 在 handle() 方法中添加 print() 语句
• 检查返回值是否为字符串
• 查看主程序日志

Q5: 如何处理大文件？

# 限制输出长度
MAX_LENGTH = 10000
if len(content) > MAX_LENGTH:
    content = content[:MAX_LENGTH] + "\n... (内容已截断)"

---

附录：插件开发检查清单

• 插件类名是 Plugin
• 实现了 get_tool_info() 方法
• 实现了 handle(args) 方法
• 返回字符串结果
• 包含 usage 属性
• 可选：实现 get_mcp_definition() 支持 MCP
• 可选：实现 convert_mcp_args() 转换参数
• 所有异常都被捕获
• 遵循参数格式规范
• 遵循 PEP 8 代码规范
• 进行了基本测试