The bot manager owns the interactive menu, local bot registry, artifact generation, and running-bot operations.
openenv.bots¶
openenv.bots
¶
Bot catalog and interactive menu helpers.
openenv.bots.manager¶
openenv.bots.manager
¶
Interactive bot catalog management.
AllBotsStackArtifacts
dataclass
¶
Result of generating the shared compose stack for every managed bot.
Source code in src/openenv/bots/manager.py
@dataclass(slots=True)
class AllBotsStackArtifacts:
"""Result of generating the shared compose stack for every managed bot."""
stack_path: Path
bot_artifacts: list[GeneratedArtifacts]
BotAnswers
dataclass
¶
Normalized answers collected from the interactive bot creation or edit flow.
Source code in src/openenv/bots/manager.py
@dataclass(slots=True)
class BotAnswers:
"""Normalized answers collected from the interactive bot creation or edit flow."""
display_name: str
role: str
skill_sources: list[str]
system_packages: list[str]
python_packages: list[str]
secret_names: list[str]
websites: list[str]
databases: list[str]
access_notes: list[str]
node_packages: list[str] = field(default_factory=list)
BotRecord
dataclass
¶
Managed bot discovered from the local bots/ catalog.
Source code in src/openenv/bots/manager.py
@dataclass(slots=True)
class BotRecord:
"""Managed bot discovered from the local `bots/` catalog."""
slug: str
manifest_path: Path
manifest: Manifest
@property
def display_name(self) -> str:
"""Return the human-facing bot name stored in the OpenClaw configuration."""
return self.manifest.openclaw.agent_name
@property
def role(self) -> str:
"""Return the role/description that summarizes what the bot is expected to do."""
return self.manifest.project.description
DocumentImprovementResult
dataclass
¶
Summary of markdown documents updated by the OpenRouter improvement flow.
Source code in src/openenv/bots/manager.py
@dataclass(slots=True)
class DocumentImprovementResult:
"""Summary of markdown documents updated by the OpenRouter improvement flow."""
bot: BotRecord
summary: str
updated_paths: list[Path]
GeneratedArtifacts
dataclass
¶
Paths and metadata produced when rendering one bot build bundle.
Source code in src/openenv/bots/manager.py
@dataclass(slots=True)
class GeneratedArtifacts:
"""Paths and metadata produced when rendering one bot build bundle."""
bot: BotRecord
lock_path: Path
dockerfile_path: Path
compose_path: Path
env_path: Path
image_tag: str
RunningBotRecord
dataclass
¶
Managed bot enriched with runtime details about its running Docker container.
Source code in src/openenv/bots/manager.py
@dataclass(slots=True)
class RunningBotRecord:
"""Managed bot enriched with runtime details about its running Docker container."""
bot: BotRecord
compose_path: Path
container_name: str
@property
def display_name(self) -> str:
"""Return the display name of the running bot."""
return self.bot.display_name
@property
def slug(self) -> str:
"""Return the managed slug of the running bot."""
return self.bot.slug
SkillSnapshotResult
dataclass
¶
Result of reconciling installed runtime skills back into the bot manifest.
Source code in src/openenv/bots/manager.py
@dataclass(slots=True)
class SkillSnapshotResult:
"""Result of reconciling installed runtime skills back into the bot manifest."""
bot: BotRecord
manifest_path: Path
lock_path: Path | None
added_skill_names: list[str]
hydrated_skill_names: list[str]
all_bots_compose_path(root)
¶
Return the shared compose path for all managed bots.
Source code in src/openenv/bots/manager.py
def all_bots_compose_path(root: str | Path) -> Path:
"""Return the shared compose path for all managed bots."""
return bots_root(root) / all_bots_compose_filename()
bots_root(root)
¶
Return the canonical directory that stores all managed bot folders.
Source code in src/openenv/bots/manager.py
def bots_root(root: str | Path) -> Path:
"""Return the canonical directory that stores all managed bot folders."""
return Path(root).resolve() / "bots"
build_bot_manifest(answers)
¶
Build a manifest from bot creation answers.
Source code in src/openenv/bots/manager.py
def build_bot_manifest(answers: BotAnswers) -> Manifest:
"""Build a manifest from bot creation answers."""
slug = slugify_name(answers.display_name)
system_packages = _unique_preserving_order(
[*DEFAULT_SYSTEM_PACKAGES, *answers.system_packages]
)
skill_sources = merge_mandatory_skill_sources(answers.skill_sources)
skills = [
build_catalog_skill(source, mandatory=source in MANDATORY_SKILL_SOURCES)
for source in skill_sources
]
tools_md = _render_tools_markdown(
skill_sources,
answers.websites,
answers.databases,
answers.access_notes,
)
memory_seed = [
f"Primary role: {answers.role}",
*[f"Website access: {website}" for website in answers.websites],
*[f"Database access: {database}" for database in answers.databases],
*answers.access_notes,
]
return Manifest(
schema_version=1,
project=ProjectConfig(
name=slug,
version="0.1.0",
description=answers.role,
runtime="openclaw",
),
runtime=RuntimeConfig(
base_image="python:3.12-slim",
python_version="3.12",
system_packages=system_packages,
python_packages=answers.python_packages,
node_packages=answers.node_packages,
env={"OPENCLAWENV_PROJECT": slug, "PYTHONUNBUFFERED": "1"},
user="root",
workdir="/workspace",
secret_refs=[],
),
agent=AgentConfig(
agents_md=(
"# Agent Contract\n\n"
f"- Primary role: {answers.role}\n"
"- Review SOUL.md, USER.md and memory.md before acting.\n"
"- Never expose secrets or credentials in output.\n"
"- Prefer reproducible, auditable commands.\n"
),
soul_md=f"# Soul\n\n{answers.role}\n",
user_md=(
"# User\n\n"
f"Bot `{answers.display_name}` supports the workspace and follows its role.\n"
),
identity_md=(
"# Identity\n\n"
f"You are {answers.display_name}.\n"
f"Your primary role is: {answers.role}\n"
),
tools_md=tools_md,
memory_seed=[item for item in memory_seed if item],
agents_md_ref=AGENT_DOC_FILENAMES["agents_md"],
soul_md_ref=AGENT_DOC_FILENAMES["soul_md"],
user_md_ref=AGENT_DOC_FILENAMES["user_md"],
identity_md_ref=AGENT_DOC_FILENAMES["identity_md"],
tools_md_ref=AGENT_DOC_FILENAMES["tools_md"],
memory_seed_ref=AGENT_DOC_FILENAMES["memory_seed"],
),
skills=skills,
openclaw=OpenClawConfig(
agent_id=slug,
agent_name=answers.display_name,
workspace="/opt/openclaw/workspace",
state_dir="/opt/openclaw",
tools_allow=["shell_command"],
tools_deny=[],
sandbox=SandboxConfig(mode="off"),
),
access=AccessConfig(
websites=answers.websites,
databases=answers.databases,
notes=answers.access_notes,
),
)
create_bot(root, answers)
¶
Create a new managed bot manifest from interactive answers.
Source code in src/openenv/bots/manager.py
def create_bot(root: str | Path, answers: BotAnswers) -> BotRecord:
"""Create a new managed bot manifest from interactive answers."""
slug = slugify_name(answers.display_name)
bot_dir = bots_root(root) / slug
if bot_dir.exists():
raise OpenEnvError(f"Bot `{slug}` already exists.")
bot_dir.mkdir(parents=True, exist_ok=False)
manifest = build_bot_manifest(answers)
_write_agent_docs(bot_dir, manifest.agent)
manifest_path = bot_dir / MANIFEST_FILENAME
manifest_path.write_text(render_manifest(manifest), encoding="utf-8")
write_secret_env(
secret_env_path(bot_dir),
answers.secret_names,
display_name=answers.display_name,
)
return load_bot(root, slug)
create_skill_snapshot(root, slug)
¶
Snapshot installed skills from a running bot and update the manifest.
Source code in src/openenv/bots/manager.py
def create_skill_snapshot(root: str | Path, slug: str) -> SkillSnapshotResult:
"""Snapshot installed skills from a running bot and update the manifest."""
running_bot = _load_running_bot(root, slug)
manifest = running_bot.bot.manifest
captured_skills = snapshot_installed_skills(
running_bot.container_name,
workspace=manifest.openclaw.workspace,
)
added_skill_names, hydrated_skill_names = _apply_skill_snapshot(
manifest,
captured_skills,
)
if not added_skill_names and not hydrated_skill_names:
return SkillSnapshotResult(
bot=running_bot.bot,
manifest_path=running_bot.bot.manifest_path,
lock_path=None,
added_skill_names=[],
hydrated_skill_names=[],
)
rendered_manifest = render_manifest(manifest)
running_bot.bot.manifest_path.write_text(rendered_manifest, encoding="utf-8")
lock_path = _preferred_lockfile_path(running_bot.bot.manifest_path.parent)
updated_lock_path: Path | None = None
if lock_path.exists():
existing_lock = load_lockfile(lock_path)
lockfile = build_lockfile(
manifest,
rendered_manifest,
resolver=lambda _: {
"digest": existing_lock.base_image["digest"],
"resolved_reference": existing_lock.base_image["resolved_reference"],
},
)
write_lockfile(lock_path, lockfile)
updated_lock_path = lock_path
return SkillSnapshotResult(
bot=load_bot(root, slug),
manifest_path=running_bot.bot.manifest_path,
lock_path=updated_lock_path,
added_skill_names=added_skill_names,
hydrated_skill_names=hydrated_skill_names,
)
delete_bot(root, slug)
¶
Delete all managed data for a bot.
Source code in src/openenv/bots/manager.py
def delete_bot(root: str | Path, slug: str) -> None:
"""Delete all managed data for a bot."""
target = bots_root(root) / slugify_name(slug)
if not target.exists():
raise OpenEnvError(f"Bot `{slug}` does not exist.")
shutil.rmtree(target, ignore_errors=False)
discover_bots(root)
¶
Discover managed bot manifests.
Source code in src/openenv/bots/manager.py
def discover_bots(root: str | Path) -> list[BotRecord]:
"""Discover managed bot manifests."""
records: list[BotRecord] = []
root_path = bots_root(root)
if not root_path.exists():
return records
for bot_dir in sorted(path for path in root_path.iterdir() if path.is_dir()):
manifest_path = _resolve_bot_manifest_path(bot_dir)
if manifest_path is None:
continue
try:
manifest, _ = load_manifest(manifest_path)
except OpenEnvError:
continue
records.append(
BotRecord(
slug=manifest_path.parent.name,
manifest_path=manifest_path,
manifest=manifest,
)
)
return records
discover_running_bots(root)
¶
Discover managed bots that currently have running Docker containers.
Source code in src/openenv/bots/manager.py
def discover_running_bots(root: str | Path) -> list[RunningBotRecord]:
"""Discover managed bots that currently have running Docker containers."""
running_containers = list_running_container_names()
records: list[RunningBotRecord] = []
for bot in discover_bots(root):
compose_path = _compose_path_for_bot(bot)
if not compose_path.exists():
continue
container_name = _container_name_for_bot(bot)
if container_name not in running_containers:
continue
records.append(
RunningBotRecord(
bot=bot,
compose_path=compose_path,
container_name=container_name,
)
)
return records
generate_all_bots_stack(root)
¶
Generate a shared compose stack with one gateway and all managed bots.
Source code in src/openenv/bots/manager.py
def generate_all_bots_stack(root: str | Path) -> AllBotsStackArtifacts:
"""Generate a shared compose stack with one gateway and all managed bots."""
bots = discover_bots(root)
if not bots:
raise OpenEnvError("No managed bots were found.")
bot_artifacts = [generate_bot_artifacts(root, bot.slug) for bot in bots]
specs = [
AllBotsComposeSpec(
slug=artifact.bot.slug,
manifest=artifact.bot.manifest,
image_tag=artifact.image_tag,
)
for artifact in bot_artifacts
]
required_shared_env_names = _materialize_all_bots_runtime(root, bot_artifacts)
shared_env_path = bots_root(root) / all_bots_env_filename()
shared_env_values = prepare_runtime_env_values(load_secret_values(shared_env_path))
_merge_required_shared_env_values(
shared_env_values,
bot_artifacts=bot_artifacts,
required_env_names=required_shared_env_names,
)
write_env_file(shared_env_path, render_all_bots_env_file(existing_values=shared_env_values))
stack_path = all_bots_compose_path(root)
write_compose(stack_path, render_all_bots_compose(specs))
return AllBotsStackArtifacts(
stack_path=stack_path,
bot_artifacts=bot_artifacts,
)
generate_bot_artifacts(root, slug)
¶
Generate lockfile, Dockerfile, compose, and env bundle for a bot.
Source code in src/openenv/bots/manager.py
def generate_bot_artifacts(root: str | Path, slug: str) -> GeneratedArtifacts:
"""Generate lockfile, Dockerfile, compose, and env bundle for a bot."""
bot = load_bot(root, slug)
manifest, raw_manifest_text = load_manifest(bot.manifest_path)
lockfile = build_lockfile(manifest, raw_manifest_text)
lock_path = _preferred_lockfile_path(bot.manifest_path.parent)
write_lockfile(lock_path, lockfile)
raw_lock_text = dump_lockfile(lockfile)
dockerfile_path = bot.manifest_path.with_name("Dockerfile")
dockerfile_path.write_text(
render_dockerfile(
manifest,
lockfile,
raw_manifest_text=raw_manifest_text,
raw_lock_text=raw_lock_text,
),
encoding="utf-8",
)
image_tag = default_image_tag(manifest.project.name, manifest.project.version)
compose_path = bot.manifest_path.parent / default_compose_filename(
manifest.openclaw.agent_name
)
write_compose(compose_path, render_compose(manifest, image_tag))
env_path = bot.manifest_path.parent / default_env_filename(manifest.openclaw.agent_name)
sidecar_env_path = secret_env_path(bot.manifest_path.parent)
existing_values = load_secret_values(env_path)
if sidecar_env_path.exists():
existing_values.update(load_secret_values(sidecar_env_path))
write_env_file(
env_path,
render_env_file(manifest, image_tag, existing_values=existing_values),
)
materialize_runtime_mount_tree(
bot.manifest_path.parent,
manifest,
lockfile,
raw_manifest_text=raw_manifest_text,
raw_lock_text=raw_lock_text,
)
return GeneratedArtifacts(
bot=bot,
lock_path=lock_path,
dockerfile_path=dockerfile_path,
compose_path=compose_path,
env_path=env_path,
image_tag=image_tag,
)
improve_bot_markdown_documents(root, slug, *, instruction, api_key)
¶
Improve bot markdown documents via OpenRouter tool calling.
Source code in src/openenv/bots/manager.py
def improve_bot_markdown_documents(
root: str | Path,
slug: str,
*,
instruction: str,
api_key: str,
) -> DocumentImprovementResult:
"""Improve bot markdown documents via OpenRouter tool calling."""
bot = _ensure_bot_agent_documents_materialized(load_bot(root, slug))
updated_paths: list[Path] = []
def write_document(relative_path: str, content: str) -> None:
"""Persist one markdown update produced by OpenRouter into the bot directory."""
target = bot.manifest_path.parent / relative_path
target.write_text(_normalize_markdown_content(content), encoding="utf-8")
updated_paths.append(target)
summary = improve_markdown_documents_with_openrouter(
api_key=api_key,
bot_name=bot.display_name,
context_payload=_bot_document_context(bot),
instruction=instruction,
write_document=write_document,
)
return DocumentImprovementResult(
bot=load_bot(root, bot.slug),
summary=summary,
updated_paths=_unique_paths(updated_paths),
)
interactive_menu(root, language=None)
¶
Run the interactive menu.
Source code in src/openenv/bots/manager.py
def interactive_menu(root: str | Path, language: str | None = None) -> int:
"""Run the interactive menu."""
base = Path(root).resolve()
lang = _select_language() if language is None else _require_language(language)
while True:
print(f"\n{_message(lang, 'menu_title')}")
print(_message(lang, "menu_list"))
print(_message(lang, "menu_add"))
print(_message(lang, "menu_edit"))
print(_message(lang, "menu_delete"))
print(_message(lang, "menu_running"))
print(_message(lang, "menu_exit"))
choice = input(_message(lang, "menu_prompt")).strip()
if choice == "1":
_interactive_browse_bots(base, lang)
continue
if choice == "2":
_interactive_add_bot(base, lang)
continue
if choice == "3":
_interactive_edit_bot(base, lang)
continue
if choice == "4":
_interactive_delete_bot(base, lang)
continue
if choice == "5":
_interactive_browse_running_bots(base, lang)
continue
if choice == "6":
print(_message(lang, "menu_exit_message"))
return 0
print(_message(lang, "menu_unknown"))
load_bot(root, slug)
¶
Load a single managed bot by slug.
Source code in src/openenv/bots/manager.py
def load_bot(root: str | Path, slug: str) -> BotRecord:
"""Load a single managed bot by slug."""
bot_dir = bots_root(root) / slugify_name(slug)
manifest_path = _resolve_bot_manifest_path(bot_dir)
if manifest_path is None:
raise OpenEnvError(f"Bot `{slug}` does not exist.")
manifest, _ = load_manifest(manifest_path)
return BotRecord(
slug=manifest_path.parent.name,
manifest_path=manifest_path,
manifest=manifest,
)
preview_running_bot_logs(root, slug, *, tail=120)
¶
Fetch recent logs for a running managed bot.
Source code in src/openenv/bots/manager.py
def preview_running_bot_logs(root: str | Path, slug: str, *, tail: int = 120) -> str:
"""Fetch recent logs for a running managed bot."""
running_bot = _load_running_bot(root, slug)
return fetch_container_logs(running_bot.container_name, tail=tail)
update_bot(root, existing_slug, answers)
¶
Update an existing managed bot manifest.
Source code in src/openenv/bots/manager.py
def update_bot(root: str | Path, existing_slug: str, answers: BotAnswers) -> BotRecord:
"""Update an existing managed bot manifest."""
current_slug = slugify_name(existing_slug)
current_dir = bots_root(root) / current_slug
if not current_dir.exists():
raise OpenEnvError(f"Bot `{existing_slug}` does not exist.")
new_slug = slugify_name(answers.display_name)
target_dir = bots_root(root) / new_slug
if new_slug != current_slug and target_dir.exists():
raise OpenEnvError(f"Bot `{new_slug}` already exists.")
existing_manifest = load_bot(root, current_slug).manifest
existing_secret_values = load_secret_values(secret_env_path(current_dir))
manifest = build_bot_manifest(answers)
manifest.openclaw.channels = deepcopy(existing_manifest.openclaw.channels)
if new_slug != current_slug:
current_dir.rename(target_dir)
else:
target_dir = current_dir
_write_agent_docs(target_dir, manifest.agent)
manifest_path = target_dir / MANIFEST_FILENAME
manifest_path.write_text(render_manifest(manifest), encoding="utf-8")
legacy_manifest_path = target_dir / LEGACY_MANIFEST_FILENAME
if legacy_manifest_path.exists():
legacy_manifest_path.unlink()
write_secret_env(
secret_env_path(target_dir),
answers.secret_names,
existing_values=existing_secret_values,
display_name=answers.display_name,
)
return load_bot(root, new_slug)