ai-proj-helper/skills-dev/ai-chat-plugin/skills/SKILL.md

name, description, arguments

name	description	arguments
ai-chat	AI Chat 测试与管理。发送消息测试 AI Chat 工具调用链路，管理工具开关和 Provider 配置，支持 local/staging 环境切换。	<subcommand> [args]

AI Chat Skill

测试和管理 Coolbuy PaaS AI Chat 服务的 Claude Code skill。

Quick Reference

命令	用途
/ai-chat send <message>	发送消息到 AI Chat，实时显示工具调用 + AI 回复
/ai-chat env [local|staging]	切换/查看目标环境（默认 local）
/ai-chat tools [category]	列出当前环境已注册的工具
/ai-chat config	查看 AI 配置（Provider、工具开关等）
/ai-chat history	显示本次会话的历史消息

---

Environment Config

两套环境，通过 /ai-chat env 切换：

环境	Auth URL	AI URL	登录账号
local (默认)	http://localhost:7089	http://localhost:7092	lining_admin / admin123
staging	http://39.105.150.219:7089	http://39.105.150.219:7092	lining_admin / admin123

状态文件

环境状态保存在 /tmp/ai-chat-state.json，格式：

{
  "env": "local",
  "token": "eyJ...",
  "token_env": "local",
  "history": []
}

---

Commands

/ai-chat env

切换或查看当前环境。

用法：

• /ai-chat env — 显示当前环境
• /ai-chat env local — 切换到本地环境
• /ai-chat env staging — 切换到 staging 环境

实现步骤：

1. 读取 /tmp/ai-chat-state.json（不存在则默认 {"env":"local","history":[]}）
2. 如果提供了参数，更新 env 字段并清空 token（环境变了 token 失效）
3. 写回状态文件
4. 输出当前环境信息表格

---

/ai-chat send

发送消息到 AI Chat 并实时显示流式响应。

用法：/ai-chat send <message>

实现步骤：

Step 1: 读取状态

# 读取状态文件
cat /tmp/ai-chat-state.json 2>/dev/null || echo '{"env":"local","history":[]}'

确定环境变量：

• local: AUTH_URL=http://localhost:7089, AI_URL=http://localhost:7092
• staging: AUTH_URL=http://39.105.150.219:7089, AI_URL=http://39.105.150.219:7092

Step 2: 获取 Token

如果状态文件中没有 token 或 token_env 与当前 env 不匹配，执行登录：

curl -s -X POST "$AUTH_URL/api/v1/auth/login" \
  -H "Content-Type: application/json" \
  -d '{"username":"lining_admin","password":"admin123"}'

从响应中提取 access_token：

# 响应格式
# {"access_token":"eyJ...","refresh_token":"...","token_type":"Bearer","expires_in":7200,"user_info":{...}}

用 python3 -c "import json,sys; print(json.load(sys.stdin)['access_token'])" 提取 token。

将 token 和 token_env 保存到状态文件。

Step 3: 构造请求并发送 SSE 流

curl -s -N -X POST "$AI_URL/api/v1/ai/chat/stream" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d "{\"message\":\"$MSG\",\"history\":$HISTORY}" 2>&1

重要: history 字段传入之前的会话历史（从状态文件读取），实现多轮对话。

Step 4: 解析 SSE 事件

用 Python 脚本解析 SSE 流（比 bash while read 更可靠）：

#!/usr/bin/env python3
"""解析 AI Chat SSE 流并格式化输出"""
import sys, json

full_content = ""
tool_calls = []

for line in sys.stdin:
    line = line.strip()
    if not line.startswith("data:"):
        continue

    data_str = line[5:].strip()
    if not data_str:
        continue

    try:
        event = json.loads(data_str)
    except json.JSONDecodeError:
        continue

    evt_type = event.get("type", "")

    if evt_type == "content":
        chunk = event.get("content", "")
        full_content += chunk
        # 实时输出内容片段
        sys.stdout.write(chunk)
        sys.stdout.flush()

    elif evt_type == "tool_call":
        tc = event.get("tool_call", {})
        tool_name = tc.get("name", "unknown")
        tool_args = tc.get("arguments", {})
        tool_id = tc.get("id", "")
        tool_calls.append({"id": tool_id, "name": tool_name})
        # 输出工具调用标记
        args_str = json.dumps(tool_args, ensure_ascii=False)
        if len(args_str) > 200:
            args_str = args_str[:200] + "..."
        print(f"\n🔧 Tool Call: {tool_name}", file=sys.stderr)
        print(f"   Args: {args_str}", file=sys.stderr)

    elif evt_type == "tool_result":
        tr = event.get("tool_result", {})
        tool_name = tr.get("name", "unknown")
        content = tr.get("content", "")
        is_error = tr.get("is_error", False)
        # 截断长结果
        if len(content) > 500:
            content = content[:500] + f"... ({len(content)} chars total)"
        status = "❌ Error" if is_error else "✅ Result"
        print(f"   {status} [{tool_name}]: {content}", file=sys.stderr)

    elif evt_type == "done":
        usage = event.get("usage") or {}
        prompt_t = usage.get("prompt_tokens", 0)
        completion_t = usage.get("completion_tokens", 0)
        total_t = usage.get("total_tokens", 0)
        print(f"\n\n--- Done ---", file=sys.stderr)
        if total_t > 0:
            print(f"Tokens: {prompt_t} prompt + {completion_t} completion = {total_t} total", file=sys.stderr)
        if tool_calls:
            print(f"Tool calls: {len(tool_calls)} ({', '.join(tc['name'] for tc in tool_calls)})", file=sys.stderr)

    elif evt_type == "error":
        err = event.get("error", "unknown error")
        print(f"\n❌ Error: {err}", file=sys.stderr)

# 输出换行
print()
# 将 full_content 输出到 fd 3 用于状态更新（如果 fd 3 打开）
try:
    with open("/tmp/ai-chat-response.txt", "w") as f:
        f.write(full_content)
except:
    pass

Step 5: 更新会话历史

发送完成后，将用户消息和 AI 回复追加到状态文件的 history 数组中：

[
  {"role": "user", "content": "<用户消息>"},
  {"role": "assistant", "content": "<AI 完整回复>"}
]

完整 bash 执行流程

#!/usr/bin/env bash
set -euo pipefail

MSG="$1"
STATE_FILE="/tmp/ai-chat-state.json"

# 1. 读取状态
if [ -f "$STATE_FILE" ]; then
  STATE=$(cat "$STATE_FILE")
else
  STATE='{"env":"local","history":[]}'
fi

ENV=$(echo "$STATE" | python3 -c "import json,sys; print(json.load(sys.stdin).get('env','local'))")
TOKEN=$(echo "$STATE" | python3 -c "import json,sys; print(json.load(sys.stdin).get('token',''))")
TOKEN_ENV=$(echo "$STATE" | python3 -c "import json,sys; print(json.load(sys.stdin).get('token_env',''))")
HISTORY=$(echo "$STATE" | python3 -c "import json,sys; print(json.dumps(json.load(sys.stdin).get('history',[])))")

# 2. 确定 URL
if [ "$ENV" = "staging" ]; then
  AUTH_URL="http://39.105.150.219:7089"
  AI_URL="http://39.105.150.219:7092"
else
  AUTH_URL="http://localhost:7089"
  AI_URL="http://localhost:7092"
fi

# 3. 获取 token（如果需要）
if [ -z "$TOKEN" ] || [ "$TOKEN_ENV" != "$ENV" ]; then
  echo "🔐 Logging in to $ENV environment..."
  LOGIN_RESP=$(curl -s -X POST "$AUTH_URL/api/v1/auth/login" \
    -H "Content-Type: application/json" \
    -d '{"username":"lining_admin","password":"admin123"}')

  TOKEN=$(echo "$LOGIN_RESP" | python3 -c "import json,sys; d=json.load(sys.stdin); print(d.get('access_token',d.get('data',{}).get('access_token','')))")

  if [ -z "$TOKEN" ]; then
    echo "❌ Login failed: $LOGIN_RESP"
    exit 1
  fi
  echo "✅ Login successful"

  # 更新状态中的 token
  STATE=$(echo "$STATE" | python3 -c "
import json,sys
s=json.load(sys.stdin)
s['token']='$TOKEN'
s['token_env']='$ENV'
print(json.dumps(s,ensure_ascii=False))
")
fi

# 4. 发送 SSE 请求并解析
echo ""
echo "📤 Sending to $ENV: $MSG"
echo "---"

# 转义消息中的特殊字符
MSG_JSON=$(python3 -c "import json; print(json.dumps('$MSG'))")

curl -s -N -X POST "$AI_URL/api/v1/ai/chat/stream" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d "{\"message\":$MSG_JSON,\"history\":$HISTORY}" 2>&1 | python3 -c "
import sys, json

full_content = ''
tool_calls = []

for line in sys.stdin:
    line = line.strip()
    if not line.startswith('data:'):
        continue
    data_str = line[5:].strip()
    if not data_str:
        continue
    try:
        event = json.loads(data_str)
    except json.JSONDecodeError:
        continue

    evt_type = event.get('type', '')

    if evt_type == 'content':
        chunk = event.get('content', '')
        full_content += chunk
        sys.stdout.write(chunk)
        sys.stdout.flush()
    elif evt_type == 'tool_call':
        tc = event.get('tool_call', {})
        tool_name = tc.get('name', 'unknown')
        tool_args = tc.get('arguments', {})
        tool_calls.append({'name': tool_name})
        args_str = json.dumps(tool_args, ensure_ascii=False)
        if len(args_str) > 200:
            args_str = args_str[:200] + '...'
        print(f'\n🔧 Tool Call: {tool_name}', file=sys.stderr)
        print(f'   Args: {args_str}', file=sys.stderr)
    elif evt_type == 'tool_result':
        tr = event.get('tool_result', {})
        tool_name = tr.get('name', 'unknown')
        content = tr.get('content', '')
        is_error = tr.get('is_error', False)
        if len(content) > 500:
            content = content[:500] + f'... ({len(content)} chars total)'
        status = '❌' if is_error else '✅'
        print(f'   {status} [{tool_name}]: {content}', file=sys.stderr)
    elif evt_type == 'done':
        usage = event.get('usage') or {}
        pt = usage.get('prompt_tokens', 0)
        ct = usage.get('completion_tokens', 0)
        tt = usage.get('total_tokens', 0)
        print(f'\n\n--- Done ---', file=sys.stderr)
        if tt > 0:
            print(f'Tokens: {pt} prompt + {ct} completion = {tt} total', file=sys.stderr)
        if tool_calls:
            names = ', '.join(tc['name'] for tc in tool_calls)
            print(f'Tool calls: {len(tool_calls)} ({names})', file=sys.stderr)
    elif evt_type == 'error':
        err = event.get('error', 'unknown error')
        print(f'\n❌ Error: {err}', file=sys.stderr)

print()
with open('/tmp/ai-chat-response.txt', 'w') as f:
    f.write(full_content)
"

# 5. 更新历史
RESPONSE=$(cat /tmp/ai-chat-response.txt 2>/dev/null || echo "")
python3 -c "
import json
state_file = '$STATE_FILE'
try:
    with open(state_file) as f:
        state = json.load(f)
except:
    state = {'env': 'local', 'history': []}

msg = $MSG_JSON
resp = '''$RESPONSE'''

state['history'].append({'role': 'user', 'content': msg})
if resp:
    state['history'].append({'role': 'assistant', 'content': resp})
state['token'] = '$TOKEN'
state['token_env'] = '$ENV'

with open(state_file, 'w') as f:
    json.dump(state, f, ensure_ascii=False, indent=2)
"

echo ""
echo "💬 History: $(echo "$STATE" | python3 -c "import json,sys; h=json.load(sys.stdin).get('history',[]); print(len(h)//2 + 1)") messages in session"

重要注意事项：

• 上面的脚本是逻辑参考，不要原样执行。Claude Code 应按步骤逐一执行 bash 命令。
• 消息中的引号和特殊字符需要用 python3 json.dumps 转义。
• 如果 token 过期（401 响应），自动重新登录。
• SSE 超时设置 --max-time 120。

---

/ai-chat tools

列出当前环境已注册的工具。

用法：

• /ai-chat tools — 列出所有工具
• /ai-chat tools <category> — 按分类过滤

实现步骤：

1. 读取状态获取环境和 token（必要时先登录）
2. 发送请求：

# 列出所有工具
curl -s "$AI_URL/api/v1/ai/tools" \
  -H "Authorization: Bearer $TOKEN"

# 按分类过滤
curl -s "$AI_URL/api/v1/ai/tools?category=order" \
  -H "Authorization: Bearer $TOKEN"

3. 响应格式：

{
  "tools": [
    {
      "name": "list_orders",
      "description": "查询订单列表",
      "category": "order",
      "enabled": true,
      "parameters": {...}
    }
  ]
}

4. 按 category 分组，输出表格：

📋 AI Tools (142 total)

order (15 tools)
  ├── list_orders         查询订单列表
  ├── get_order_detail    获取订单详情
  └── ...

product (12 tools)
  ├── list_products       查询商品列表
  └── ...

已知工具分类：order, product, sku, inventory, task, brand, requirement, customer, dashboard, distribution, finance, discount, channel, approval, organization, feature_gap

---

/ai-chat config

查看 AI 服务配置。

实现步骤：

1. 根据当前环境读取对应配置文件：

   • local: 读取 ai-service/api/etc/ai-api-local.yaml
   • staging: SSH 到 staging 读取 /opt/coolbuy/configs/ai-api.yaml

2. 显示关键配置项：

   • AI Provider 和 Model
   • 各工具分类的开关状态
   • API 端口

3. 输出格式：

⚙️  AI Config (local)

Provider:  deepseek
Model:     deepseek-chat
Port:      7092

Tool Categories:
  ✅ order          ✅ product        ✅ sku
  ✅ inventory      ✅ task           ✅ brand
  ✅ requirement    ✅ customer       ✅ dashboard
  ✅ distribution   ✅ finance        ✅ discount
  ✅ channel        ✅ approval       ✅ organization
  ✅ feature_gap

---

/ai-chat history

显示当前会话的历史消息。

实现步骤：

1. 读取 /tmp/ai-chat-state.json 的 history 数组
2. 按时间顺序显示：

💬 Chat History (3 messages)

[1] 👤 User: 你好
    🤖 AI: 你好！我是 AI 助手...

[2] 👤 User: 搜索组织架构找到大客户部
    🤖 AI: 我来帮你搜索... (used: search_organizations)

[3] 👤 User: 最近的订单
    🤖 AI: 以下是最近的订单列表...

3. 如果需要清空历史：/ai-chat history clear
   • 删除状态文件中的 history 数组，重置为空

---

SSE Event Format Reference

AI Chat SSE 流使用 event: message + data: {json} 格式：

type	数据字段	说明
content	content: "<text>"	增量文本内容
tool_call	tool_call: {id, name, arguments}	AI 请求调用工具
tool_result	tool_result: {tool_call_id, name, content, is_error}	工具执行结果
done	usage: {prompt_tokens, completion_tokens, total_tokens} (可能为 null), finish_reason	流结束
error	error: "<message>"	错误

---

Troubleshooting

Token 过期 (401)

如果请求返回 401，删除缓存 token 重新登录：

# 清除 token 强制重新登录
python3 -c "
import json
with open('/tmp/ai-chat-state.json') as f: s=json.load(f)
s['token']=''
with open('/tmp/ai-chat-state.json','w') as f: json.dump(s,f)
"

连接失败

• local: 确认本地服务已启动 (./scripts/start_dev.sh)
• staging: 确认 staging 服务运行中 (ssh coolbuy-staging "docker ps | grep ai-service")

SSE 流中断

• 检查 AI 服务日志
• local: 查看终端输出
• staging: ssh coolbuy-staging "docker logs coolbuy-ai-service --tail 50"

消息中包含特殊字符

务必用 python3 -c "import json; print(json.dumps(msg))" 转义消息内容，避免 JSON 解析失败。