Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
omniagents / omniagents / core / project / export.py
Size: Mime:
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")