zhuchaowe/qwen_agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

自修改w

Browse Source
This commit is contained in:
朱潮 2026-03-17 13:43:26 +08:00
parent 09c4d5e804
commit 7117fccb7a
3 changed files with 660 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

190
skills/bot-self-modifier/SKILL.md Normal file
View File

@ -0,0 +1,190 @@
---
name: bot-self-modifier
description: |
Bot 自修改技能，允许当前 bot 在对话过程中通过 API 读取和修改自身的全部配置。
使用场景：
- 用户要求 bot 调整行为或切换人设时（如"把你的角色改成英语老师"、"在系统提示词里加上XXX规则"） → 修改系统提示词
- 用户要求 bot 更换头像、修改自我介绍、设置开场建议问题时（如"把你的头像换成这个链接"、"更新你的描述"） → 修改头像/描述/建议问题
- 用户要求 bot 接入新工具或移除已有工具时（如"帮我接入 Jina 搜索"、"把那个 MCP 服务器删掉"） → 添加/删除 MCP 服务器
- 用户要求 bot 安装、启用、禁用或卸载技能时（如"帮我装上这个技能包"、"把 XX 技能关掉"） → 管理技能列表
- 用户要求 bot 配置 API 密钥或运行参数时（如"把 JINA_API_KEY 设置成 xxx"） → 修改环境变量
- bot 需要自主进化、动态调整自身能力边界的自动化场景
---
# Bot Self-Modifier
## Skill Structure
```
bot-self-modifier/
├── SKILL.md # Core instruction file (this file)
├── scripts/
│ └── bot_modifier.py # Main bot modification script
```
## Overview
支持以下功能：
1. **系统提示词** - 读取和修改
2. **Bot 基本信息** - 头像、描述、建议问题的读取和修改
3. **MCP 服务器** - 列表查看、添加和删除
4. **技能列表** - 读取、上传、启用、禁用、删除
5. **环境变量** - 读取和修改
## Quick Start
所有操作都通过 `scripts/bot_modifier.py` 脚本执行，使用 `--action` 参数指定操作。
### 系统提示词操作
```bash
# 读取当前系统提示词
scripts/bot_modifier.py --action get_prompt
# 修改系统提示词
scripts/bot_modifier.py --action set_prompt --value "你是一个专业的客服助手"
```
### Bot 基本信息操作
```bash
# 读取当前 bot 基本信息（名称、头像、描述、建议问题）
scripts/bot_modifier.py --action get_info
# 修改 bot 标题名称
scripts/bot_modifier.py --action set_name --value "智能客服助手"
# 设置头像（URL）
scripts/bot_modifier.py --action set_avatar --value "https://example.com/avatar.png"
# 设置描述
scripts/bot_modifier.py --action set_description --value "这是一个智能客服助手"
# 设置建议问题（JSON 数组）
scripts/bot_modifier.py --action set_suggestions --value '["你能做什么？", "帮我查一下订单", "如何退款"]'
```
### MCP 服务器操作
```bash
# 列出所有 MCP 服务器
scripts/bot_modifier.py --action list_mcp
# 添加 MCP 服务器（config 为 JSON 字符串）
scripts/bot_modifier.py --action add_mcp --name "my-server" --mcp-type "sse" --config '{"url": "https://example.com/sse"}'
# 删除 MCP 服务器
scripts/bot_modifier.py --action delete_mcp --mcp-id "<MCP_SERVER_ID>"
```
### 技能操作
```bash
# 列出所有技能
scripts/bot_modifier.py --action list_skills
# 上传技能（zip 文件）
scripts/bot_modifier.py --action upload_skill --file /path/to/skill.zip
# 启用技能（逗号分隔的技能名）
scripts/bot_modifier.py --action enable_skill --value "skill-name-1,skill-name-2"
# 禁用技能（从已启用列表中移除）
scripts/bot_modifier.py --action disable_skill --value "skill-name-to-remove"
# 删除已上传的技能
scripts/bot_modifier.py --action delete_skill --value "skill-name"
```
### 环境变量操作
```bash
# 读取当前环境变量
scripts/bot_modifier.py --action get_env
# 修改环境变量（JSON 格式）
scripts/bot_modifier.py --action set_env --value '{"API_KEY": "xxx", "SECRET": "yyy"}'
```
## Script Usage
### bot_modifier.py
主脚本，通过 HTTP API 与服务端交互。
```bash
scripts/bot_modifier.py [OPTIONS]
```
**Options:**
| Option | Required | Description | Default |
|--------|----------|-------------|---------|
| `--action` | Yes | 操作类型（见下方列表） | - |
| `--value` | Depends | 操作值（提示词内容/技能名/环境变量 JSON） | - |
| `--name` | For add_mcp | MCP 服务器名称 | - |
| `--mcp-type` | For add_mcp | MCP 服务器类型 (sse/streamable-http) | sse |
| `--config` | For add_mcp | MCP 服务器配置 JSON | - |
| `--mcp-id` | For delete_mcp | MCP 服务器 ID | - |
| `--file` | For upload_skill | Skill zip 文件路径 | - |
**Available Actions:**
| Action | Description |
|--------|-------------|
| `get_prompt` | 读取系统提示词 |
| `set_prompt` | 修改系统提示词（需要 --value） |
| `get_info` | 读取 bot 基本信息（名称、头像、描述、建议问题） |
| `set_name` | 修改 bot 标题名称（需要 --value） |
| `set_avatar` | 设置头像 URL（需要 --value） |
| `set_description` | 设置描述（需要 --value） |
| `set_suggestions` | 设置建议问题（需要 --value，JSON 数组） |
| `list_mcp` | 列出 MCP 服务器 |
| `add_mcp` | 添加 MCP 服务器（需要 --name, --config） |
| `delete_mcp` | 删除 MCP 服务器（需要 --mcp-id） |
| `list_skills` | 列出所有技能 |
| `upload_skill` | 上传技能 zip（需要 --file） |
| `enable_skill` | 启用技能（需要 --value，逗号分隔技能名） |
| `disable_skill` | 禁用技能（需要 --value，技能名） |
| `delete_skill` | 删除已上传技能（需要 --value） |
| `get_env` | 读取环境变量 |
| `set_env` | 修改环境变量（需要 --value，JSON 格式） |
## Common Workflows
### 工作流 1: 查看并修改 Bot 配置
```bash
# 1. 先查看当前配置
scripts/bot_modifier.py --action get_prompt
scripts/bot_modifier.py --action get_info
scripts/bot_modifier.py --action get_env
scripts/bot_modifier.py --action list_mcp
scripts/bot_modifier.py --action list_skills
# 2. 修改需要的配置
scripts/bot_modifier.py --action set_prompt --value "新的提示词内容"
```
### 工作流 2: 添加新 MCP 服务器并配置环境变量
```bash
# 1. 添加 MCP 服务器
scripts/bot_modifier.py --action add_mcp --name "jina-search" --mcp-type "sse" --config '{"url": "https://mcp.jina.ai/sse"}'
# 2. 配置需要的环境变量
scripts/bot_modifier.py --action set_env --value '{"JINA_API_KEY": "jina_xxx"}'
```
### 工作流 3: 上传并启用新技能
```bash
# 1. 上传技能包
scripts/bot_modifier.py --action upload_skill --file /path/to/my-skill.zip
# 2. 启用技能
scripts/bot_modifier.py --action enable_skill --value "my-skill"
# 3. 配置技能所需的环境变量
scripts/bot_modifier.py --action set_env --value '{"SKILL_API_KEY": "xxx"}'
```

436
skills/bot-self-modifier/scripts/bot_modifier.py Executable file
View File

@ -0,0 +1,436 @@
#!/usr/bin/env python3
"""
Bot Self-Modifier Script
通过 API 读取和修改 bot 配置：系统提示词、MCP服务器、技能、环境变量
"""
import argparse
import json
import os
import sys
import urllib.request
import urllib.error
import urllib.parse
def get_config():
"""获取配置，下面的MASTERKEY和ASSISTANT_ID是从环境变量自动获取的，不需要用户提供"""
masterkey = os.environ.get("MASTERKEY", "master")
bot_id = os.environ.get("ASSISTANT_ID", "")
if not masterkey:
print("ERROR: MASTERKEY environment variable is required")
sys.exit(1)
return masterkey, "http://localhost:8001", bot_id
def api_request(method, url, headers=None, data=None, is_multipart=False):
"""发送 HTTP 请求"""
if headers is None:
headers = {}
if data is not None and not is_multipart:
data = json.dumps(data).encode("utf-8")
headers.setdefault("Content-Type", "application/json")
req = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req) as resp:
body = resp.read().decode("utf-8")
return json.loads(body) if body else {}
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8")
try:
detail = json.loads(body)
except Exception:
detail = body
print(f"ERROR: HTTP {e.code} - {detail}")
sys.exit(1)
def build_multipart_data(fields, files):
"""构建 multipart/form-data 请求体"""
boundary = "----BotModifierBoundary"
lines = []
for key, value in fields.items():
lines.append(f"--{boundary}".encode())
lines.append(f'Content-Disposition: form-data; name="{key}"'.encode())
lines.append(b"")
lines.append(value.encode() if isinstance(value, str) else value)
for key, (filename, file_data, content_type) in files.items():
lines.append(f"--{boundary}".encode())
lines.append(
f'Content-Disposition: form-data; name="{key}"; filename="{filename}"'.encode()
)
lines.append(f"Content-Type: {content_type}".encode())
lines.append(b"")
lines.append(file_data)
lines.append(f"--{boundary}--".encode())
lines.append(b"")
body = b"\r\n".join(lines)
content_type = f"multipart/form-data; boundary={boundary}"
return body, content_type
# ============ 系统提示词 ============
def get_prompt(bot_id, masterkey, api_base):
"""读取系统提示词"""
url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
result = api_request("GET", url, headers)
prompt = result.get("system_prompt", "")
print("=== System Prompt ===")
print(prompt if prompt else "(empty)")
return prompt
def set_prompt(bot_id, masterkey, api_base, value):
"""修改系统提示词"""
url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
data = {"system_prompt": value}
result = api_request("PUT", url, headers, data)
print(f"OK: System prompt updated. {result.get('message', '')}")
# ============ Bot 基本信息 ============
def get_info(bot_id, masterkey, api_base):
"""读取 bot 基本信息（头像、描述、建议问题）"""
url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
result = api_request("GET", url, headers)
print("=== Bot Info ===")
print(f" Name: {result.get('name', '(empty)')}")
print(f" Avatar URL: {result.get('avatar_url') or '(empty)'}")
print(f" Description: {result.get('description') or '(empty)'}")
suggestions = result.get("suggestions") or []
print(f" Suggestions: {json.dumps(suggestions, ensure_ascii=False) if suggestions else '(empty)'}")
return result
def set_avatar(bot_id, masterkey, api_base, value):
"""设置头像 URL"""
url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
data = {"avatar_url": value}
result = api_request("PUT", url, headers, data)
print(f"OK: Avatar updated. {result.get('message', '')}")
def set_name(bot_id, masterkey, api_base, value):
"""设置 bot 标题名称"""
url = f"{api_base}/api/v1/bots/{bot_id}"
headers = {"Authorization": f"Bearer {masterkey}"}
data = {"name": value}
result = api_request("PUT", url, headers, data)
print(f"OK: Name updated to '{result.get('name', value)}'.")
def set_description(bot_id, masterkey, api_base, value):
"""设置描述"""
url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
data = {"description": value}
result = api_request("PUT", url, headers, data)
print(f"OK: Description updated. {result.get('message', '')}")
def set_suggestions(bot_id, masterkey, api_base, value):
"""设置建议问题（JSON 数组）"""
try:
suggestions = json.loads(value)
except json.JSONDecodeError:
print("ERROR: --value must be a valid JSON array for set_suggestions")
sys.exit(1)
if not isinstance(suggestions, list):
print("ERROR: --value must be a JSON array, e.g. '[\"question1\", \"question2\"]'")
sys.exit(1)
url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
data = {"suggestions": suggestions}
result = api_request("PUT", url, headers, data)
print(f"OK: Suggestions updated. {result.get('message', '')}")
# ============ MCP 服务器 ============
def list_mcp(bot_id, masterkey, api_base):
"""列出 MCP 服务器"""
url = f"{api_base}/api/v1/bots/{bot_id}/mcp"
headers = {"Authorization": f"Bearer {masterkey}"}
result = api_request("GET", url, headers)
if not result:
print("No MCP servers configured.")
return
print("=== MCP Servers ===")
for server in result:
enabled_str = "enabled" if server.get("enabled") else "disabled"
print(f" [{enabled_str}] {server['name']} (id: {server['id']}, type: {server.get('type', 'N/A')})")
config = server.get("config", {})
if config:
print(f" config: {json.dumps(config, ensure_ascii=False)}")
return result
def add_mcp(bot_id, masterkey, api_base, name, mcp_type, config_str):
"""添加 MCP 服务器"""
try:
config = json.loads(config_str)
except json.JSONDecodeError:
print("ERROR: --config must be valid JSON")
sys.exit(1)
url = f"{api_base}/api/v1/bots/{bot_id}/mcp"
headers = {"Authorization": f"Bearer {masterkey}"}
data = {
"name": name,
"type": mcp_type,
"config": config,
"enabled": True,
}
result = api_request("POST", url, headers, data)
print(f"OK: MCP server '{name}' added (id: {result.get('id', 'N/A')})")
def delete_mcp(bot_id, masterkey, api_base, mcp_id):
"""删除 MCP 服务器"""
url = f"{api_base}/api/v1/bots/{bot_id}/mcp/{mcp_id}"
headers = {"Authorization": f"Bearer {masterkey}"}
result = api_request("DELETE", url, headers)
print(f"OK: MCP server deleted. {result.get('message', '')}")
# ============ 技能操作 ============
def list_skills(bot_id, masterkey, api_base):
"""列出所有技能"""
# 获取技能文件列表
url = f"{api_base}/api/v1/skill/list?bot_id={urllib.parse.quote(bot_id)}"
headers = {"Authorization": f"Bearer {masterkey}"}
result = api_request("GET", url, headers)
# 获取当前启用的技能
settings_url = f"{api_base}/api/v1/bots/{bot_id}/settings"
settings = api_request("GET", settings_url, headers)
enabled_skills_str = settings.get("skills", "") or ""
enabled_skills = [s.strip() for s in enabled_skills_str.split(",") if s.strip()]
skills = result.get("skills", [])
print(f"=== Skills (total: {result.get('total', len(skills))}) ===")
print(f"Enabled: {', '.join(enabled_skills) if enabled_skills else '(none)'}")
print()
for skill in skills:
is_user = "[user]" if skill.get("user_skill") else "[official]"
is_enabled = "*" if skill["name"] in enabled_skills else " "
print(f" {is_enabled} {is_user} {skill['name']}: {skill.get('description', '')}")
return skills
def upload_skill(bot_id, masterkey, api_base, file_path):
"""上传技能 zip"""
if not os.path.exists(file_path):
print(f"ERROR: File not found: {file_path}")
sys.exit(1)
filename = os.path.basename(file_path)
with open(file_path, "rb") as f:
file_data = f.read()
body, content_type = build_multipart_data(
fields={"bot_id": bot_id},
files={"file": (filename, file_data, "application/zip")},
)
url = f"{api_base}/api/v1/skill/upload"
headers = {
"Authorization": f"Bearer {masterkey}",
"Content-Type": content_type,
}
result = api_request("POST", url, headers, body, is_multipart=True)
print(f"OK: Skill uploaded. {result.get('message', '')} (name: {result.get('skill_name', 'N/A')})")
def enable_skill(bot_id, masterkey, api_base, skill_names):
"""启用技能 - 将技能名添加到 settings.skills"""
# 先获取当前启用的技能
settings_url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
settings = api_request("GET", settings_url, headers)
current = settings.get("skills", "") or ""
current_list = [s.strip() for s in current.split(",") if s.strip()]
new_skills = [s.strip() for s in skill_names.split(",") if s.strip()]
for s in new_skills:
if s not in current_list:
current_list.append(s)
updated = ",".join(current_list)
data = {"skills": updated}
result = api_request("PUT", settings_url, headers, data)
print(f"OK: Skills enabled. Current: {updated}")
def disable_skill(bot_id, masterkey, api_base, skill_name):
"""禁用技能 - 从 settings.skills 中移除"""
settings_url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
settings = api_request("GET", settings_url, headers)
current = settings.get("skills", "") or ""
current_list = [s.strip() for s in current.split(",") if s.strip()]
to_remove = [s.strip() for s in skill_name.split(",") if s.strip()]
current_list = [s for s in current_list if s not in to_remove]
updated = ",".join(current_list)
data = {"skills": updated}
result = api_request("PUT", settings_url, headers, data)
print(f"OK: Skills disabled. Current: {updated if updated else '(none)'}")
def delete_skill(bot_id, masterkey, api_base, skill_name):
"""删除已上传的技能"""
url = (
f"{api_base}/api/v1/skill/remove"
f"?bot_id={urllib.parse.quote(bot_id)}"
f"&skill_name={urllib.parse.quote(skill_name)}"
)
headers = {"Authorization": f"Bearer {masterkey}"}
result = api_request("DELETE", url, headers)
print(f"OK: Skill '{skill_name}' deleted. {result.get('message', '')}")
# 同时从启用列表中移除
disable_skill(bot_id, masterkey, api_base, skill_name)
# ============ 环境变量 ============
def get_env(bot_id, masterkey, api_base):
"""读取环境变量"""
url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
result = api_request("GET", url, headers)
shell_env = result.get("shell_env", {})
print("=== Environment Variables ===")
if not shell_env:
print("(none)")
else:
for key, value in shell_env.items():
display_val = value if value else "(empty)"
print(f" {key}={display_val}")
return shell_env
def set_env(bot_id, masterkey, api_base, env_json):
"""修改环境变量（合并更新，不会覆盖未指定的变量）"""
# 先获取当前环境变量
settings_url = f"{api_base}/api/v1/bots/{bot_id}/settings"
headers = {"Authorization": f"Bearer {masterkey}"}
settings = api_request("GET", settings_url, headers)
current_env = settings.get("shell_env", {}) or {}
try:
new_env = json.loads(env_json)
except json.JSONDecodeError:
print("ERROR: --value must be valid JSON for set_env")
sys.exit(1)
# 合并
current_env.update(new_env)
data = {"shell_env": current_env}
result = api_request("PUT", settings_url, headers, data)
print(f"OK: Environment variables updated. {result.get('message', '')}")
for key, value in current_env.items():
print(f" {key}={value if value else '(empty)'}")
# ============ Main ============
def main():
parser = argparse.ArgumentParser(description="Bot Self-Modifier: read and modify bot configuration via API")
parser.add_argument(
"--action",
required=True,
choices=[
"get_prompt", "set_prompt",
"get_info", "set_name", "set_avatar", "set_description", "set_suggestions",
"list_mcp", "add_mcp", "delete_mcp",
"list_skills", "upload_skill", "enable_skill", "disable_skill", "delete_skill",
"get_env", "set_env",
],
help="Action to perform",
)
parser.add_argument("--value", help="Value for the action (prompt text, skill name, env JSON, etc.)")
parser.add_argument("--name", help="MCP server name (for add_mcp)")
parser.add_argument("--mcp-type", default="sse", help="MCP server type (for add_mcp, default: sse)")
parser.add_argument("--config", help="MCP server config JSON (for add_mcp)")
parser.add_argument("--mcp-id", help="MCP server ID (for delete_mcp)")
parser.add_argument("--file", help="Skill zip file path (for upload_skill)")
args = parser.parse_args()
masterkey, api_base, bot_id = get_config()
action_map = {
"get_prompt": lambda: get_prompt(bot_id, masterkey, api_base),
"set_prompt": lambda: set_prompt(bot_id, masterkey, api_base, args.value),
"get_info": lambda: get_info(bot_id, masterkey, api_base),
"set_name": lambda: set_name(bot_id, masterkey, api_base, args.value),
"set_avatar": lambda: set_avatar(bot_id, masterkey, api_base, args.value),
"set_description": lambda: set_description(bot_id, masterkey, api_base, args.value),
"set_suggestions": lambda: set_suggestions(bot_id, masterkey, api_base, args.value),
"list_mcp": lambda: list_mcp(bot_id, masterkey, api_base),
"add_mcp": lambda: add_mcp(bot_id, masterkey, api_base, args.name, args.mcp_type, args.config),
"delete_mcp": lambda: delete_mcp(bot_id, masterkey, api_base, args.mcp_id),
"list_skills": lambda: list_skills(bot_id, masterkey, api_base),
"upload_skill": lambda: upload_skill(bot_id, masterkey, api_base, args.file),
"enable_skill": lambda: enable_skill(bot_id, masterkey, api_base, args.value),
"disable_skill": lambda: disable_skill(bot_id, masterkey, api_base, args.value),
"delete_skill": lambda: delete_skill(bot_id, masterkey, api_base, args.value),
"get_env": lambda: get_env(bot_id, masterkey, api_base),
"set_env": lambda: set_env(bot_id, masterkey, api_base, args.value),
}
# 参数验证
if args.action == "set_prompt" and not args.value:
print("ERROR: --value is required for set_prompt")
sys.exit(1)
if args.action in ("set_avatar", "set_name", "set_description", "set_suggestions") and not args.value:
print(f"ERROR: --value is required for {args.action}")
sys.exit(1)
if args.action == "add_mcp" and (not args.name or not args.config):
print("ERROR: --name and --config are required for add_mcp")
sys.exit(1)
if args.action == "delete_mcp" and not args.mcp_id:
print("ERROR: --mcp-id is required for delete_mcp")
sys.exit(1)
if args.action == "upload_skill" and not args.file:
print("ERROR: --file is required for upload_skill")
sys.exit(1)
if args.action in ("enable_skill", "disable_skill", "delete_skill") and not args.value:
print(f"ERROR: --value is required for {args.action}")
sys.exit(1)
if args.action == "set_env" and not args.value:
print("ERROR: --value is required for set_env (JSON format)")
sys.exit(1)
action_map[args.action]()
if __name__ == "__main__":
main()

34
skills/bot-self-modifier/skill.yaml Normal file
View File

@ -0,0 +1,34 @@
name: bot-self-modifier
version: 1.0.0
description: Bot 自修改技能，允许 bot 通过 API 读取和修改自身配置（系统提示词、MCP服务器、技能、环境变量）
author:
name: sparticle
email: support@gbase.ai
license: MIT
tags:
- bot-management
- self-modifier
- configuration
runtime:
python: ">=3.7"
dependencies: []
entry_point: scripts/bot_modifier.py
env:
MASTERKEY:
type: string
required: true
description: Master Key for API authentication
BOT_MODIFIER_API_BASE:
type: string
required: false
default: http://localhost:8001
description: API server base URL
config:
bot_id:
type: string
required: true
description: Target Bot ID
action:
type: string
required: true
description: Action to perform