Repository URL to install this package:
|
Version:
0.7.16 ▾
|
from __future__ import annotations
from pathlib import Path
import yaml
from omniagents.core.config.loader import load_agent_spec_from_yaml
def save_agent_as_py(
config_path: str, output_path: str, project_tool_dirs: list[Path] | None = None
):
spec = load_agent_spec_from_yaml(config_path)
tool_funcs = spec.available_tools
cfg_path = Path(config_path)
raw_cfg = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
instructions = ""
if raw_cfg.get("instructions"):
instructions = raw_cfg["instructions"]
elif raw_cfg.get("instructions_file"):
instr_file = cfg_path.parent / raw_cfg["instructions_file"]
if not instr_file.is_file():
alt = Path(raw_cfg["instructions_file"])
if alt.is_file():
instr_file = alt
if instr_file.is_file():
instructions = instr_file.read_text(encoding="utf-8")
instructions = instructions.replace("'''", "\\'\\'\\'")
tool_dirs = []
base = cfg_path.parent
default_tools = base / "tools"
if default_tools.is_dir():
tool_dirs.append(default_tools)
for td in project_tool_dirs or []:
try:
p = Path(td)
except Exception:
p = None
if p and p.is_dir():
tool_dirs.append(p)
for td in raw_cfg.get("additional_tools_dirs", []):
p = (base / td).resolve()
if p.is_dir():
tool_dirs.append(p)
tool_files = []
for d in tool_dirs:
for py in d.rglob("*.py"):
if py.name != "__init__.py":
tool_files.append(py)
lines: list[str] = []
lines.append("# Autogenerated standalone agent script\n")
for tf in tool_files:
rel = tf.relative_to(Path.cwd())
lines.append(f"# --- From {rel} ---")
lines.append(tf.read_text(encoding="utf-8"))
lines.append("\n")
lines.append("def get_agent_instructions():")
lines.append(" return '''")
lines.append(instructions)
lines.append("'''\n")
lines.append("from omniagents.core.runtime.runner import Runner")
lines.append(
"from omniagents.core.agents.specs import AgentSpec, SafeAgentOptions, SettingField, MCPServerConfig\n"
)
lines.append("spec = AgentSpec(")
lines.append(f" name={spec.name!r},")
try:
_lbl = getattr(spec, "label", None)
except Exception:
_lbl = None
try:
_desc = getattr(spec, "description", None)
except Exception:
_desc = None
lines.append(f" label={_lbl!r},")
lines.append(f" description={_desc!r},")
lines.append(f" welcome_text={spec.welcome_text!r},")
lines.append(" get_agent_instructions=get_agent_instructions,")
lines.append(" build_agent=None, # use default internal builder\n")
tool_names = [t.name for t in tool_funcs]
if not tool_names:
try:
raw_tools = raw_cfg.get("tools") or []
if isinstance(raw_tools, list):
tool_names = [str(x) for x in raw_tools]
except Exception:
pass
lines.append(" available_tools=[%s]," % ", ".join(tool_names))
lines.append(f" model_name={spec.model_name!r},")
lines.append(f" use_safe_agent={spec.use_safe_agent},")
so = spec.safe_agent_options
lines.append(
f" safe_agent_options=SafeAgentOptions(skip_approvals={so.skip_approvals}, halt_on_rejection={so.halt_on_rejection}, safe_tool_names={so.safe_tool_names!r}, safe_tool_patterns={so.safe_tool_patterns!r}),"
)
lines.append(" mcp_servers=[")
for server in getattr(spec, "mcp_servers", []) or []:
params_repr = repr(server.params)
options_repr = "None" if server.options is None else repr(server.options)
lines.append(
f" MCPServerConfig(name={server.name!r}, type={server.type!r}, params={params_repr}, options={options_repr}),"
)
lines.append(" ],")
_proj_slug = getattr(spec, "project_slug", None)
_agent_slug = getattr(spec, "agent_slug", None)
lines.append(f" project_slug={_proj_slug!r},")
lines.append(f" agent_slug={_agent_slug!r},")
lines.append(" settings_fields=[")
for field in spec.settings_fields:
lines.append(
f" SettingField(key={field.key!r}, label={field.label!r}, group={field.group!r}, default={field.default!r}, secret={field.secret}, hint={field.hint!r}),"
)
lines.append(" ],")
lines.append(")\n")
lines.append("if __name__ == '__main__':")
lines.append(" runner = Runner(spec=spec)")
lines.append(" runner.run(mode='gui')\n")
Path(output_path).write_text("\n".join(lines), encoding="utf-8")