196 lines
7.2 KiB
Python
196 lines
7.2 KiB
Python
"""
|
||
快速创建新 Agent 脚本
|
||
从 luna 模板复制出一个新 agent 目录，自动替换所有唯一标识符，并：
|
||
- 生成 agents/<dir>/ 下的 agent.py / app.py / api_server.py / chat.py / .env / __init__.py
|
||
- 生成 mcp_dev_agent/<dir>_server.py 入口
|
||
- 更新 C:/Users/nzy/.codebuddy/.mcp.json 追加 MCP server 条目
|
||
|
||
命名约定（沿用 luna/qwen）：
|
||
DIR = 用户输入（如 claude）
|
||
AGENT = {DIR}_agent （App name / agent name）
|
||
DB = sessions_{DIR}.db
|
||
MCP = kebab-case（如 claude-agent）
|
||
|
||
用法：
|
||
python create_agent.py claude --model opcode/claude-sonnet
|
||
python create_agent.py claude --model opcode/claude-sonnet --port 8005 --display "Claude Dev Agent"
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import sys
|
||
|
||
# 强制 UTF-8（Windows 控制台默认 GBK，无法打印 ✓/→ 等字符）
|
||
if sys.platform == "win32":
|
||
os.environ.setdefault("PYTHONUTF8", "1")
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
sys.stderr.reconfigure(encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
# 仓库根目录
|
||
REPO_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||
AGENTS_DIR = os.path.join(REPO_ROOT, "agents")
|
||
MCP_DIR = os.path.join(REPO_ROOT, "mcp_dev_agent")
|
||
TEMPLATE_DIR = os.path.join(AGENTS_DIR, "luna")
|
||
MCP_CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".codebuddy", ".mcp.json")
|
||
VENV_PYTHON = os.path.join(REPO_ROOT, ".venv", "Scripts", "python.exe")
|
||
|
||
# 模板中需要替换的标识符（按顺序执行，先长后短避免误伤）
|
||
REPLACEMENTS = [
|
||
("luna_agent", "{AGENT}"), # 覆盖所有 app/agent name 引用
|
||
("luna", "{DIR}"), # 剩余：模块路径、db 名、instruction 内称呼
|
||
("Luna", "{TITLE}"), # 首字母大写：Luna Agent / Luna: / === Luna
|
||
("8002", "{PORT}"), # 默认端口
|
||
("opcode/deepseek-v4-flash", "{MODEL}"), # .env 的 VLLM_MODEL
|
||
]
|
||
|
||
# 复制时排除的目录/文件
|
||
SKIP_DIR_NAMES = {"__pycache__", ".adk", ".git", ".idea"}
|
||
SKIP_FILENAMES = {".gitignore"}
|
||
|
||
|
||
def interact(arg_dir: str, model: str, port: int, display: str) -> tuple[str, str, int, str]:
|
||
"""补齐缺失参数（缺省时交互式询问）"""
|
||
d = arg_dir
|
||
if not d:
|
||
d = input("Agent 目录名（如 claude）: ").strip()
|
||
if not model:
|
||
model = input("模型名（如 opcode/claude-sonnet）: ").strip()
|
||
if port is None:
|
||
port = auto_next_port()
|
||
if not display:
|
||
display = f"{d.title()} Agent ({model}) 全栈开发助手"
|
||
return d, model, port, display
|
||
|
||
|
||
def auto_next_port() -> int:
|
||
"""扫描 agents/*/.env 的 API_SERVER_PORT，取最大值 +1"""
|
||
max_port = 8000
|
||
if os.path.isdir(AGENTS_DIR):
|
||
for entry in os.listdir(AGENTS_DIR):
|
||
env_path = os.path.join(AGENTS_DIR, entry, ".env")
|
||
if os.path.isfile(env_path):
|
||
m = re.search(r"API_SERVER_PORT\s*=\s*(\d+)", open(env_path, encoding="utf-8").read())
|
||
if m:
|
||
max_port = max(max_port, int(m.group(1)))
|
||
return max_port + 1
|
||
|
||
|
||
def copy_tree(src: str, dst: str) -> None:
|
||
"""递归复制，跳过 __pycache__/.adk/.git 等"""
|
||
os.makedirs(dst, exist_ok=True)
|
||
for name in os.listdir(src):
|
||
s = os.path.join(src, name)
|
||
d = os.path.join(dst, name)
|
||
if os.path.isdir(s):
|
||
if name in SKIP_DIR_NAMES:
|
||
continue
|
||
copy_tree(s, d)
|
||
else:
|
||
if name in SKIP_FILENAMES:
|
||
continue
|
||
shutil.copy2(s, d)
|
||
|
||
|
||
def apply_replacements(path: str, mapping: dict) -> None:
|
||
"""对文件内容按顺序做字符串替换"""
|
||
with open(path, encoding="utf-8") as f:
|
||
content = f.read()
|
||
for old, key in REPLACEMENTS:
|
||
content = content.replace(old, mapping.get(key, ""))
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
|
||
def update_mcp_config(mcp_key: str, server_file: str, display: str) -> None:
|
||
"""在 .mcp.json 的 mcpServers 中追加一条"""
|
||
if not os.path.isfile(MCP_CONFIG_PATH):
|
||
print(f"[warn] 未找到 {MCP_CONFIG_PATH}，跳过 MCP 配置更新")
|
||
return
|
||
|
||
with open(MCP_CONFIG_PATH, encoding="utf-8") as f:
|
||
cfg = json.load(f)
|
||
|
||
if mcp_key in cfg.get("mcpServers", {}):
|
||
print(f"[warn] .mcp.json 已存在 {mcp_key} 条目，跳过")
|
||
return
|
||
|
||
cfg.setdefault("mcpServers", {})[mcp_key] = {
|
||
"type": "stdio",
|
||
"command": VENV_PYTHON,
|
||
"args": [server_file],
|
||
"description": display,
|
||
}
|
||
|
||
with open(MCP_CONFIG_PATH, "w", encoding="utf-8") as f:
|
||
json.dump(cfg, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f" ✓ 已更新 {MCP_CONFIG_PATH}")
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="创建一个新的 Dev Agent（从 luna 模板）")
|
||
parser.add_argument("dir", nargs="?", help="agent 目录名（如 claude）")
|
||
parser.add_argument("--model", default=None, help="模型名（如 opcode/claude-sonnet）")
|
||
parser.add_argument("--port", type=int, default=None, help="API 端口，默认自动取最大+1")
|
||
parser.add_argument("--display", default=None, help="MCP 描述，默认 '{Dir} Agent ({model}) 全栈开发助手'")
|
||
args = parser.parse_args()
|
||
|
||
d, model, port, display = interact(args.dir, args.model, args.port, args.display)
|
||
|
||
if not d or not model:
|
||
print("错误：目录名和模型名不能为空")
|
||
sys.exit(1)
|
||
|
||
agent = f"{d}_agent"
|
||
title = d.title()
|
||
db = f"sessions_{d}.db"
|
||
mcp_key = f"{d.replace('_', '-')}-agent"
|
||
target_dir = os.path.join(AGENTS_DIR, d)
|
||
server_file = os.path.join(MCP_DIR, f"{d}_server.py")
|
||
|
||
mapping = {"{AGENT}": agent, "{DIR}": d, "{TITLE}": title,
|
||
"{PORT}": str(port), "{MODEL}": model}
|
||
|
||
# 1. 校验目标目录不存在
|
||
if os.path.exists(target_dir):
|
||
print(f"错误：目录已存在 {target_dir}")
|
||
sys.exit(1)
|
||
if os.path.exists(server_file):
|
||
print(f"错误：MCP server 已存在 {server_file}")
|
||
sys.exit(1)
|
||
|
||
print(f"创建 Agent: {d} (agent={agent}, port={port}, model={model})")
|
||
print(f" 数据库: {db}")
|
||
|
||
# 2. 复制模板目录
|
||
copy_tree(TEMPLATE_DIR, target_dir)
|
||
# 3. 替换所有文件里的标识符
|
||
for root, _dirs, files in os.walk(target_dir):
|
||
for fn in files:
|
||
apply_replacements(os.path.join(root, fn), mapping)
|
||
# 4. 重写 __init__.py
|
||
with open(os.path.join(target_dir, "__init__.py"), "w", encoding="utf-8") as f:
|
||
f.write(f"# {d} package\nfrom . import agent\n")
|
||
print(f" ✓ 已生成 agents/{d}/")
|
||
|
||
# 5. 生成 MCP server 入口
|
||
shutil.copy2(os.path.join(MCP_DIR, "luna_server.py"), server_file)
|
||
apply_replacements(server_file, mapping)
|
||
print(f" ✓ 已生成 {server_file}")
|
||
|
||
# 6. 更新 .mcp.json
|
||
update_mcp_config(mcp_key, server_file, display)
|
||
|
||
print("\n完成！启动方式：")
|
||
print(f" cd agents/{d} && python api_server.py # 启动 API Server（自动注册到网关）")
|
||
print(f" cd agents/{d} && python chat.py # 命令行对话")
|
||
print(f" python agent_status.py --agent {agent} # 查看状态（需先手动加入 AGENTS 映射）")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |