mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-05-19 08:02:30 +00:00
feat(skills): add create-instance built-in skill
Add a skill that lets a running nanobot agent create new bot instances through a helper script. The agent collects instance name, channel type, and optional model from the user, then runs the script which: - Calls nanobot onboard to create config + workspace skeleton - Enables the target channel and sets workspace/model in config - Auto-assigns gateway/API ports if defaults are occupied - Validates config via Pydantic before saving - Reports required fields the user needs to fill in (e.g. bot token)
This commit is contained in:
parent
0ca0fe2221
commit
a7aeb1d2ea
50
nanobot/skills/create-instance/SKILL.md
Normal file
50
nanobot/skills/create-instance/SKILL.md
Normal file
@ -0,0 +1,50 @@
|
||||
---
|
||||
name: create-instance
|
||||
description: "Create a new nanobot instance with separate config and workspace. Use when the user wants to set up a new bot for a different channel, persona, or purpose."
|
||||
---
|
||||
|
||||
# Create Instance
|
||||
|
||||
Set up a new nanobot instance with its own config and workspace.
|
||||
|
||||
## When to Use
|
||||
|
||||
When the user wants to create a new bot instance — typically for a different channel (Telegram, Discord, WeChat, etc.) or with different settings.
|
||||
|
||||
## Steps
|
||||
|
||||
1. **Collect information from the user** (ask one at a time if not already provided):
|
||||
- **Instance name** (required): a short identifier like `telegram-bot`, `discord-bot`
|
||||
- **Channel type** (required): e.g. `telegram`, `discord`, `weixin`, `feishu`, `slack`
|
||||
- **Model** (optional): LLM model to use. Defaults to the same model as the current instance.
|
||||
|
||||
2. **Do NOT collect sensitive information** in the chat (API keys, bot tokens, secrets). API keys are automatically copied from the current instance. Channel-specific tokens (e.g. `telegram.token`) still need to be filled in manually.
|
||||
|
||||
3. **Run the creation script** using the exec tool — always pass `--inherit-config` with the current instance's config path so API keys are copied:
|
||||
|
||||
```bash
|
||||
python D:/path/to/nanobot/skills/create-instance/scripts/create_instance.py --name <name> --channel <channel> --inherit-config ~/.nanobot/config.json [--model <model>] [--config-dir <path>]
|
||||
```
|
||||
|
||||
**Path rules (critical on Windows):**
|
||||
- Use **forward-slash absolute paths** to the script, e.g. `D:/path/to/create_instance.py`
|
||||
- Do **NOT** wrap paths in quotes — the exec tool will mangle them
|
||||
- Do **NOT** use `cd` — the exec tool ignores it; working directory stays as workspace
|
||||
- Do **NOT** use backslash paths like `D:\path` — they will fail
|
||||
|
||||
Use `~/.nanobot/config.json` as the `--inherit-config` path unless the current instance uses a custom config location.
|
||||
|
||||
4. **Report results to the user**:
|
||||
- Where the config and workspace were created
|
||||
- Which fields they need to fill in (the script will list them)
|
||||
- The command to start the instance: `nanobot gateway --config <config-path>`
|
||||
|
||||
## Examples
|
||||
|
||||
User: "help me create a Telegram bot" (or similar request)
|
||||
|
||||
→ Ask for an instance name if not obvious from context
|
||||
→ Ask which model to use (optional, can skip if user doesn't care)
|
||||
→ Run: `python D:/path/to/nanobot/skills/create-instance/scripts/create_instance.py --name telegram-bot --channel telegram --inherit-config ~/.nanobot/config.json`
|
||||
(Replace `D:/path/to/nanobot` with the actual nanobot source directory)
|
||||
→ Tell user: config created at `~/.nanobot-telegram/config.json`, fill in `channels.telegram.token`, then start with `nanobot gateway --config ~/.nanobot-telegram/config.json`
|
||||
0
nanobot/skills/create-instance/scripts/__init__.py
Normal file
0
nanobot/skills/create-instance/scripts/__init__.py
Normal file
241
nanobot/skills/create-instance/scripts/create_instance.py
Normal file
241
nanobot/skills/create-instance/scripts/create_instance.py
Normal file
@ -0,0 +1,241 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Create a new nanobot instance with a dedicated config and workspace.
|
||||
|
||||
Usage:
|
||||
create_instance.py --name <name> --channel <channel> [--model <model>] [--config-dir <dir>]
|
||||
|
||||
Examples:
|
||||
create_instance.py --name telegram-bot --channel telegram
|
||||
create_instance.py --name discord-bot --channel discord --model deepseek/deepseek-chat
|
||||
create_instance.py --name my-bot --channel telegram --config-dir ~/.nanobot-custom
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import socket
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _validate_name(name: str) -> str:
|
||||
"""Normalize and validate instance name."""
|
||||
name = name.strip().lower()
|
||||
name = re.sub(r"[^a-z0-9-]", "-", name)
|
||||
name = re.sub(r"-{2,}", "-", name)
|
||||
name = name.strip("-")
|
||||
if not name:
|
||||
print("[ERROR] Instance name must contain at least one letter or digit.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
if len(name) > 64:
|
||||
print(f"[ERROR] Instance name too long ({len(name)} chars, max 64).", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
return name
|
||||
|
||||
|
||||
def _get_available_channels() -> list[str]:
|
||||
"""Get list of available channel names without importing channel classes."""
|
||||
from nanobot.channels.registry import discover_channel_names
|
||||
|
||||
return discover_channel_names()
|
||||
|
||||
|
||||
def _run_onboard(config_path: Path, workspace: Path) -> None:
|
||||
"""Create skeleton config + workspace using nanobot's programmatic API."""
|
||||
from nanobot.cli.commands import _onboard_plugins
|
||||
from nanobot.config.loader import save_config, set_config_path
|
||||
from nanobot.config.paths import get_workspace_path
|
||||
from nanobot.config.schema import Config
|
||||
from nanobot.utils.helpers import sync_workspace_templates
|
||||
|
||||
config = Config()
|
||||
config.agents.defaults.workspace = str(workspace)
|
||||
set_config_path(config_path)
|
||||
save_config(config, config_path)
|
||||
_onboard_plugins(config_path)
|
||||
|
||||
workspace_path = get_workspace_path(config.workspace_path)
|
||||
if not workspace_path.exists():
|
||||
workspace_path.mkdir(parents=True, exist_ok=True)
|
||||
sync_workspace_templates(workspace_path)
|
||||
|
||||
|
||||
def _patch_config(
|
||||
config_path: Path,
|
||||
*,
|
||||
channel: str,
|
||||
workspace: Path,
|
||||
model: str | None,
|
||||
inherit_config_path: Path | None = None,
|
||||
) -> dict:
|
||||
"""Patch the generated config: enable channel, set workspace, optionally set model."""
|
||||
data = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
|
||||
# Inherit providers (API keys, api_base, etc.) from current instance
|
||||
if inherit_config_path and inherit_config_path.exists():
|
||||
try:
|
||||
src = json.loads(inherit_config_path.read_text(encoding="utf-8"))
|
||||
src_providers = src.get("providers", {})
|
||||
if src_providers:
|
||||
data.setdefault("providers", {})
|
||||
for key, val in src_providers.items():
|
||||
if isinstance(val, dict) and val.get("apiKey"):
|
||||
data["providers"][key] = val
|
||||
except Exception as exc:
|
||||
print(f"[WARN] Could not inherit providers from {inherit_config_path}: {exc}", file=sys.stderr)
|
||||
|
||||
# Set workspace and model
|
||||
data.setdefault("agents", {}).setdefault("defaults", {})
|
||||
data["agents"]["defaults"]["workspace"] = str(workspace)
|
||||
if model:
|
||||
data["agents"]["defaults"]["model"] = model
|
||||
|
||||
# Enable the target channel
|
||||
channels = data.setdefault("channels", {})
|
||||
if channel in channels and isinstance(channels[channel], dict):
|
||||
channels[channel]["enabled"] = True
|
||||
else:
|
||||
channels[channel] = {"enabled": True}
|
||||
|
||||
# Auto-assign ports if defaults are already in use
|
||||
_assign_free_ports(data)
|
||||
|
||||
# Validate with Pydantic, then save
|
||||
from nanobot.config.schema import Config
|
||||
|
||||
Config.model_validate(data)
|
||||
config_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
return data
|
||||
|
||||
|
||||
def _is_port_in_use(port: int, host: str = "127.0.0.1") -> bool:
|
||||
"""Check if a port is already in use."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
try:
|
||||
s.bind((host, port))
|
||||
return False
|
||||
except OSError:
|
||||
return True
|
||||
|
||||
|
||||
def _find_free_port(start: int, host: str = "127.0.0.1", max_tries: int = 100) -> int:
|
||||
"""Find the first free port starting from `start`."""
|
||||
for port in range(start, start + max_tries):
|
||||
if not _is_port_in_use(port, host):
|
||||
return port
|
||||
# OS-level fallback: ask the kernel for an ephemeral port
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.bind((host, 0))
|
||||
return s.getsockname()[1]
|
||||
|
||||
|
||||
def _assign_free_ports(data: dict) -> None:
|
||||
"""If default gateway or API ports are in use, assign free ones."""
|
||||
from nanobot.config.schema import ApiConfig, GatewayConfig
|
||||
|
||||
defaults = [
|
||||
("gateway", GatewayConfig()),
|
||||
("api", ApiConfig()),
|
||||
]
|
||||
for key, default_cfg in defaults:
|
||||
section = data.setdefault(key, {})
|
||||
port = section.get("port", default_cfg.port)
|
||||
host = section.get("host", default_cfg.host)
|
||||
if _is_port_in_use(port, host):
|
||||
section["port"] = _find_free_port(port + 1, host)
|
||||
|
||||
|
||||
def _get_channel_required_fields(channel: str) -> list[str]:
|
||||
"""Inspect a channel's default config and list fields that are empty strings."""
|
||||
try:
|
||||
from nanobot.channels.registry import load_channel_class
|
||||
|
||||
cls = load_channel_class(channel)
|
||||
default = cls.default_config()
|
||||
return sorted(k for k, v in default.items() if isinstance(v, str) and v == "" and k != "enabled")
|
||||
except Exception as exc:
|
||||
print(f"[WARN] Could not inspect channel '{channel}' defaults: {exc}", file=sys.stderr)
|
||||
return []
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Create a new nanobot instance.",
|
||||
)
|
||||
parser.add_argument("--name", required=True, help="Instance name (e.g. telegram-bot)")
|
||||
parser.add_argument("--channel", required=True, help="Channel type (e.g. telegram, discord)")
|
||||
parser.add_argument("--model", default=None, help="LLM model (default: same as current instance)")
|
||||
parser.add_argument(
|
||||
"--config-dir",
|
||||
default=None,
|
||||
help="Config directory (default: ~/.nanobot-{name})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--inherit-config",
|
||||
default=None,
|
||||
help="Path to current instance's config.json to copy API keys from",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
# Validate name
|
||||
name = _validate_name(args.name)
|
||||
|
||||
# Validate channel
|
||||
available = _get_available_channels()
|
||||
if args.channel not in available:
|
||||
print(f"[ERROR] Unknown channel: {args.channel}", file=sys.stderr)
|
||||
print(f"Available channels: {', '.join(sorted(available))}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Resolve paths
|
||||
home = Path.home()
|
||||
config_dir = Path(args.config_dir).expanduser().resolve() if args.config_dir else home / f".nanobot-{name}"
|
||||
config_path = config_dir / "config.json"
|
||||
workspace = config_dir / "workspace"
|
||||
|
||||
# Check for duplicate
|
||||
if config_path.exists():
|
||||
print(f"[ERROR] Config already exists at {config_path}", file=sys.stderr)
|
||||
print("Delete it first or use a different --config-dir.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Creating instance '{name}'...")
|
||||
print(f" Config dir: {config_dir}")
|
||||
print(f" Workspace: {workspace}")
|
||||
print(f" Channel: {args.channel}")
|
||||
if args.model:
|
||||
print(f" Model: {args.model}")
|
||||
|
||||
# Run onboard
|
||||
_run_onboard(config_path, workspace)
|
||||
|
||||
# Patch config
|
||||
inherit_path = Path(args.inherit_config).expanduser().resolve() if args.inherit_config else None
|
||||
_patch_config(
|
||||
config_path,
|
||||
channel=args.channel,
|
||||
workspace=workspace,
|
||||
model=args.model,
|
||||
inherit_config_path=inherit_path,
|
||||
)
|
||||
|
||||
# Report
|
||||
print(f"\n[OK] Instance '{name}' created successfully.")
|
||||
print(f" Config: {config_path}")
|
||||
print(f" Workspace: {workspace}")
|
||||
|
||||
# List fields the user needs to fill in
|
||||
required_fields = _get_channel_required_fields(args.channel)
|
||||
if required_fields:
|
||||
print(f"\n[IMPORTANT] Edit {config_path} and fill in these fields:")
|
||||
for field in required_fields:
|
||||
print(f" - channels.{args.channel}.{field}")
|
||||
|
||||
print(f"\nTo start the instance:")
|
||||
print(f" nanobot gateway --config {config_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
168
tests/skills/test_create_instance.py
Normal file
168
tests/skills/test_create_instance.py
Normal file
@ -0,0 +1,168 @@
|
||||
"""Tests for nanobot/skills/create-instance/scripts/create_instance.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
SCRIPT = Path(__file__).parent.parent.parent / "nanobot" / "skills" / "create-instance" / "scripts" / "create_instance.py"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Point HOME at a temp dir so nanobot writes configs there."""
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.delenv("NANOBOT_CONFIG", raising=False)
|
||||
return tmp_path
|
||||
|
||||
|
||||
def _run_script(*args: str, cwd: Path | None = None) -> subprocess.CompletedProcess:
|
||||
"""Run create_instance.py as a subprocess."""
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT), *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
cwd=cwd,
|
||||
)
|
||||
|
||||
|
||||
class TestValidation:
|
||||
"""Argument validation tests."""
|
||||
|
||||
def test_missing_required_args_exits_with_error(self) -> None:
|
||||
result = _run_script()
|
||||
assert result.returncode != 0
|
||||
|
||||
def test_invalid_channel_exits_with_error(self, tmp_home: Path) -> None:
|
||||
result = _run_script("--name", "test", "--channel", "nonexistent_channel")
|
||||
assert result.returncode != 0
|
||||
assert "nonexistent_channel" in result.stderr or "nonexistent_channel" in result.stdout
|
||||
|
||||
|
||||
class TestCreateInstance:
|
||||
"""End-to-end instance creation tests."""
|
||||
|
||||
def test_creates_config_and_workspace(self, tmp_home: Path) -> None:
|
||||
config_dir = tmp_home / ".nanobot-test"
|
||||
result = _run_script(
|
||||
"--name", "test-bot",
|
||||
"--channel", "telegram",
|
||||
"--config-dir", str(config_dir),
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
config_path = config_dir / "config.json"
|
||||
assert config_path.exists(), f"Config not created at {config_path}"
|
||||
|
||||
workspace = config_dir / "workspace"
|
||||
assert workspace.exists(), f"Workspace not created at {workspace}"
|
||||
|
||||
def test_config_has_channel_enabled(self, tmp_home: Path) -> None:
|
||||
config_dir = tmp_home / ".nanobot-test"
|
||||
result = _run_script(
|
||||
"--name", "test-bot",
|
||||
"--channel", "telegram",
|
||||
"--config-dir", str(config_dir),
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
|
||||
assert data["channels"]["telegram"]["enabled"] is True
|
||||
|
||||
def test_config_workspace_path_set(self, tmp_home: Path) -> None:
|
||||
config_dir = tmp_home / ".nanobot-test"
|
||||
result = _run_script(
|
||||
"--name", "test-bot",
|
||||
"--channel", "telegram",
|
||||
"--config-dir", str(config_dir),
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
|
||||
ws = data["agents"]["defaults"]["workspace"]
|
||||
assert str(config_dir / "workspace") in ws or "workspace" in ws
|
||||
|
||||
def test_model_override(self, tmp_home: Path) -> None:
|
||||
config_dir = tmp_home / ".nanobot-test"
|
||||
result = _run_script(
|
||||
"--name", "test-bot",
|
||||
"--channel", "telegram",
|
||||
"--model", "deepseek/deepseek-chat",
|
||||
"--config-dir", str(config_dir),
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
|
||||
assert data["agents"]["defaults"]["model"] == "deepseek/deepseek-chat"
|
||||
|
||||
def test_rejects_duplicate_instance(self, tmp_home: Path) -> None:
|
||||
config_dir = tmp_home / ".nanobot-test"
|
||||
result1 = _run_script(
|
||||
"--name", "test-bot",
|
||||
"--channel", "telegram",
|
||||
"--config-dir", str(config_dir),
|
||||
)
|
||||
assert result1.returncode == 0
|
||||
|
||||
result2 = _run_script(
|
||||
"--name", "test-bot",
|
||||
"--channel", "telegram",
|
||||
"--config-dir", str(config_dir),
|
||||
)
|
||||
assert result2.returncode != 0
|
||||
|
||||
def test_port_reassigned_when_default_in_use(self, tmp_home: Path) -> None:
|
||||
"""When default gateway port is occupied, script should pick a different one."""
|
||||
config_dir = tmp_home / ".nanobot-test"
|
||||
|
||||
# Bind to the default gateway port to simulate a running instance
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as blocker:
|
||||
blocker.bind(("127.0.0.1", 18790))
|
||||
blocker.listen(1)
|
||||
|
||||
result = _run_script(
|
||||
"--name", "test-bot",
|
||||
"--channel", "telegram",
|
||||
"--config-dir", str(config_dir),
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
|
||||
assert data["gateway"]["port"] != 18790
|
||||
|
||||
def test_inherits_api_key_from_current_instance(self, tmp_home: Path) -> None:
|
||||
"""API keys from --inherit-config should be copied to new instance."""
|
||||
# Create a fake "current instance" config with an API key
|
||||
src_dir = tmp_home / ".nanobot-current"
|
||||
src_dir.mkdir()
|
||||
src_config = src_dir / "config.json"
|
||||
src_config.write_text(json.dumps({
|
||||
"providers": {
|
||||
"anthropic": {"apiKey": "sk-test-key-12345"},
|
||||
"deepseek": {"apiKey": "dsk-another-key"},
|
||||
"openai": {}, # no key, should not be copied
|
||||
},
|
||||
}), encoding="utf-8")
|
||||
|
||||
config_dir = tmp_home / ".nanobot-new"
|
||||
result = _run_script(
|
||||
"--name", "new-bot",
|
||||
"--channel", "telegram",
|
||||
"--config-dir", str(config_dir),
|
||||
"--inherit-config", str(src_config),
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
|
||||
providers = data.get("providers", {})
|
||||
assert providers.get("anthropic", {}).get("apiKey") == "sk-test-key-12345"
|
||||
assert providers.get("deepseek", {}).get("apiKey") == "dsk-another-key"
|
||||
# openai had no key, so it should not be in the new config's providers
|
||||
assert providers.get("openai", {}).get("apiKey") is None
|
||||
Loading…
x
Reference in New Issue
Block a user