APM-Packages/omos/omos.py
Tobias J. Endres b897bb4b17 feat(omos): interactive init and setup commands
- omos init [--fill]: create user mapping file, interactively resolve
  PLACEHOLDER classes by picking from models discovered in
  opencode.json(c)
- omos setup: guided flow - show profile summary, create/fill missing
  mappings with explicit user choice, then render
- README quickstart updated to the two-command path
2026-08-23 17:10:35 +02:00

607 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""omos Adapter zwischen APM-Team-Profilen und oh-my-opencode-slim.
Liest ein team-profile.yaml aus einem APM-Package, löst Modellklassen gegen
eine lokale Mapping-Tabelle auf und generiert ein namespacetes Preset für
oh-my-opencode-slim.
Kommandos:
omos render Team-Profil in OMOS-Preset übersetzen und schreiben
omos check Validieren ohne zu schreiben (Dry-Run)
Beispiele:
python3 omos.py render --package examples/package --mapping ~/.config/apm-team/model-mapping.yaml
python3 omos.py check --package examples/package
"""
from __future__ import annotations
import argparse
import copy
import json
import re
import sys
from pathlib import Path
import yaml
# Eingebaute OMOS-Agenten. Alles andere wird als Custom Agent behandelt.
BUILTIN_AGENTS = {
"orchestrator",
"oracle",
"librarian",
"explorer",
"fixer",
"designer",
"council",
"observer",
}
DEFAULT_MAPPING_PATH = Path.home() / ".config" / "apm-team" / "model-mapping.yaml"
DEFAULT_TEAM_FILE = "team-profile.yaml"
DEFAULT_OUTPUT = Path(".opencode") / "oh-my-opencode-slim.json"
PROVENANCE_SUFFIX = ".omos-provenance.json"
SUPPORTED_SCHEMA = ("corentic.team-profile/v1", "acme.team-profile/v1")
class OmosError(Exception):
"""Fehler mit nutzerlesbarer Ursache."""
# ---------------------------------------------------------------------------
# Laden
def load_yaml(path: Path) -> dict:
if not path.exists():
raise OmosError(f"Datei nicht gefunden: {path}")
try:
data = yaml.safe_load(path.read_text(encoding="utf-8"))
except yaml.YAMLError as exc:
raise OmosError(f"YAML-Fehler in {path}: {exc}") from exc
if not isinstance(data, dict):
raise OmosError(f"{path} enthält kein YAML-Mapping")
return data
def load_team_profile(package_dir: Path, team_file: str | None) -> tuple[dict, Path]:
candidates = (
[package_dir / team_file]
if team_file
else [package_dir / DEFAULT_TEAM_FILE, package_dir / "team.yaml"]
)
for candidate in candidates:
if candidate.exists():
profile = load_yaml(candidate)
schema = profile.get("schema")
if schema and schema not in SUPPORTED_SCHEMA:
print(
f"Warnung: Unbekanntes Schema '{schema}'. "
f"Unterstützt: {', '.join(SUPPORTED_SCHEMA)}. Fahre fort."
)
if not profile.get("id"):
raise OmosError(f"{candidate}: Feld 'id' fehlt")
roles = profile.get("roles")
if not isinstance(roles, list) or not roles:
raise OmosError(f"{candidate}: Keine Rollen definiert ('roles')")
return profile, candidate
searched = ", ".join(str(c) for c in candidates)
raise OmosError(f"Kein Team-Profil gefunden. Gesucht: {searched}")
def load_mapping(mapping_path: Path | None) -> dict:
path = mapping_path or DEFAULT_MAPPING_PATH
if not path.exists():
raise OmosError(
f"Modell-Mapping nicht gefunden: {path}\n"
"Lege die Datei an, z. B.:\n"
" model_classes:\n"
" fast-research: ollama/qwen3.5:9b\n"
" high-reasoning:\n"
" model: ollama/qwen3.6:35b-a3b-q4_K_M\n"
" variant: thinking"
)
data = load_yaml(path)
classes = data.get("model_classes")
if not isinstance(classes, dict) or not classes:
raise OmosError(f"{path}: Sektion 'model_classes' fehlt oder ist leer")
return classes
def resolve_model(model_class_entry: object, model_class: str, role_id: str) -> dict:
"""Löst einen Mapping-Eintrag (string oder dict) zu Agent-Feldern auf."""
if isinstance(model_class_entry, str):
return {"model": model_class_entry}
if isinstance(model_class_entry, dict):
entry = {k: v for k, v in model_class_entry.items()}
if "model" not in entry:
raise OmosError(
f"Rolle '{role_id}': Mapping für '{model_class}' hat kein 'model'-Feld"
)
return entry
raise OmosError(f"Ungültiger Mapping-Eintrag für '{model_class}': {model_class_entry!r}")
def sanitize_preset_name(team_id: str) -> str:
return re.sub(r"[^a-zA-Z0-9_-]+", "-", team_id).strip("-").lower()
# ---------------------------------------------------------------------------
# Übersetzung
def build_preset(profile: dict, mapping: dict) -> tuple[dict, dict, list[str]]:
"""Erzeugt (preset, custom_agents, warnings)."""
preset: dict = {}
custom_agents: dict = {}
warnings: list[str] = []
for role in profile["roles"]:
role_id = role.get("id")
if not role_id:
raise OmosError("Rolle ohne 'id' gefunden")
model_class = role.get("model_class")
if not model_class:
raise OmosError(f"Rolle '{role_id}': 'model_class' fehlt")
if model_class not in mapping:
raise OmosError(
f"Rolle '{role_id}': Modellklasse '{model_class}' ist nicht gemappt.\n"
f"Ergänze in deinem Mapping:\n"
f" {model_class}: <provider/model>"
)
agent_fields = resolve_model(mapping[model_class], model_class, role_id)
capabilities = role.get("capabilities", {}) or {}
mcps = list(capabilities.get("mcps", []) or [])
skills = list(capabilities.get("skills", []) or [])
purpose = str(role.get("purpose", "")).strip()
omos_agent = role.get("omos_agent", role_id)
if omos_agent == "custom":
# Explizit als Custom Agent markiert → rolleneigener Name.
agent_key = role_id
else:
agent_key = omos_agent
preset[agent_key] = {
**agent_fields,
"mcps": mcps,
"skills": skills,
}
if agent_key in BUILTIN_AGENTS and agent_key != role_id:
preset[agent_key]["displayName"] = role_id
if agent_key not in BUILTIN_AGENTS or omos_agent == "custom":
if agent_key not in custom_agents:
prompt = (
purpose
if purpose
else f"Custom Agent '{role_id}' aus Team-Profil "
f"'{profile.get('id', 'unbekannt')}'."
)
custom_agents[agent_key] = {
"model": agent_fields["model"],
"description": purpose or f"Custom subagent '{role_id}'",
"prompt": prompt,
# Dem Orchestrator sagen, wann er delegieren soll.
"orchestratorPrompt": (
f"@{agent_key}\n- Rolle: {purpose}\n"
"- Delegiere Aufgaben dieser Rolle an diesen Agenten."
if purpose
else f"@{agent_key}"
),
}
else:
custom_agents[agent_key]["model"] = agent_fields["model"]
if not mcps and not skills:
warnings.append(
f"Rolle '{role_id}' ({agent_key}): keine MCPs/Skills zugewiesen "
"(rein koordinierend?)"
)
orchestrator_present = any(k == "orchestrator" for k in preset)
if not orchestrator_present:
warnings.append(
"Kein 'orchestrator' im Team. Ohne Orchestrator-Preset-Eintrag bleibt "
"dessen Modell unverändert."
)
return preset, custom_agents, warnings
def merge_into_config(config: dict, preset_name: str, preset: dict,
custom_agents: dict) -> dict:
merged = copy.deepcopy(config)
presets = merged.setdefault("presets", {})
existing = presets.get(preset_name)
if existing:
print(
f"Hinweis: Preset '{preset_name}' existierte bereits und wird ersetzt "
"(andere Presets bleiben unberührt)."
)
presets[preset_name] = preset
agents = merged.setdefault("agents", {})
agents.update(custom_agents)
return merged
# ---------------------------------------------------------------------------
# MCP-Verfügbarkeit prüfen (best effort)
def strip_jsonc(text: str) -> str:
text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
text = re.sub(r"(^|\s)//[^\n]*", r"\1", text)
return text
def configured_mcps() -> set[str]:
candidates = [
Path.home() / ".config" / "opencode" / "opencode.jsonc",
Path.home() / ".config" / "opencode" / "opencode.json",
]
for path in candidates:
if path.exists():
try:
data = json.loads(strip_jsonc(path.read_text(encoding="utf-8")))
return set((data.get("mcp") or {}).keys())
except (json.JSONDecodeError, OSError):
continue
return set()
def available_models() -> list[str]:
"""Listet alle provider/model-Kombinationen aus der opencode.json(c)."""
candidates = [
Path.home() / ".config" / "opencode" / "opencode.jsonc",
Path.home() / ".config" / "opencode" / "opencode.json",
]
models: list[str] = []
for path in candidates:
if not path.exists():
continue
try:
data = json.loads(strip_jsonc(path.read_text(encoding="utf-8")))
except (json.JSONDecodeError, OSError):
continue
for provider, pconf in (data.get("provider") or {}).items():
for model in (pconf.get("models") or {}):
models.append(f"{provider}/{model}")
break
return sorted(models)
INIT_TEMPLATE = """\
# Lokale Mapping-Tabelle des NUTZERS.
# Ordnet abstrakte Modellklassen konkrete Modell-IDs zu.
# Formate:
# klasse: provider/modell
# klasse:
# model: provider/modell
# variant: thinking # optional
# temperature: 0.7 # optional
model_classes:
strong-generalist: PLACEHOLDER
fast-research: PLACEHOLDER
strong-writing: PLACEHOLDER
high-reasoning: PLACEHOLDER
cheap-reliable: PLACEHOLDER
"""
def check_mcps(preset: dict, warnings: list[str]) -> None:
available = configured_mcps()
if not available:
warnings.append(
"Konnte opencode.json(c) nicht lesen MCP-Prüfung übersprungen."
)
return
needed: set[str] = set()
for agent in preset.values():
needed.update(agent.get("mcps", []) or [])
missing = sorted(m for m in needed if m not in available)
if missing:
warnings.append(
"MCPs im Team-Profil, aber nicht in opencode.json konfiguriert: "
+ ", ".join(missing)
)
# ---------------------------------------------------------------------------
# Kommandos
def package_output_path(package_dir: Path) -> Path:
return package_dir / DEFAULT_OUTPUT
def cmd_render(args: argparse.Namespace) -> int:
package_dir = Path(args.package).resolve()
profile, profile_path = load_team_profile(package_dir, args.team_file)
mapping = load_mapping(Path(args.mapping) if args.mapping else None)
preset, custom_agents, warnings = build_preset(profile, mapping)
output = Path(args.output) if args.output else package_dir / DEFAULT_OUTPUT
if output.exists():
try:
config = json.loads(output.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
raise OmosError(f"{output} ist kein gültiges JSON: {exc}")
else:
config = {}
config["$schema"] = (
"https://unpkg.com/oh-my-opencode-slim@latest/"
"oh-my-opencode-slim.schema.json"
)
merged = merge_into_config(
config, sanitize_preset_name(str(profile["id"])), preset, custom_agents
)
if args.dry_run:
print(json.dumps(merged, indent=2, ensure_ascii=False))
_report(warnings, dry_run=True)
return 0
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(merged, indent=2, ensure_ascii=False), encoding="utf-8")
print(f"Geschrieben: {output}")
provenance = {
"profile": profile["id"],
"profileFile": str(profile_path),
"teamId": profile["id"],
"description": profile.get("description", ""),
"presetName": sanitize_preset_name(str(profile["id"])),
"roles": {
r.get("id"): {
"omosAgent": r.get("omos_agent", r.get("id")),
"modelClass": r.get("model_class"),
"resolvedModel": mapping[r["model_class"]],
"mcps": (r.get("capabilities", {}) or {}).get("mcps", []),
"skills": (r.get("capabilities", {}) or {}).get("skills", []),
}
for r in profile["roles"]
},
}
provenance_path = output.with_suffix(PROVENANCE_SUFFIX)
provenance_path.write_text(
json.dumps(provenance, indent=2, ensure_ascii=False), encoding="utf-8"
)
print(f"Provenance: {provenance_path}")
_report(warnings)
print(
"\nNächste Schritte:\n"
" 1. opencode starten (Projekt-Config wird geladen)\n"
f" 2. /preset {sanitize_preset_name(str(profile['id']))}\n"
" 3. OpenCode neu laden → Team aktiv"
)
return 0
def cmd_check(args: argparse.Namespace) -> int:
args.dry_run = True
# Dry-Run: vorhandene Projekt-Config einbeziehen, aber nie schreiben.
existing = package_output_path(Path(args.package).resolve())
args.output = str(existing) if existing.exists() else None
try:
cmd_render(args)
except OmosError as exc:
print(f"Fehler: {exc}", file=sys.stderr)
return 1
return 0
def _report(warnings: list[str], dry_run: bool = False) -> None:
if dry_run:
print("\n--- Dry-Run: Es wurde nichts geschrieben ---")
if warnings:
print("\nWarnungen:")
for warning in warnings:
print(f"{warning}")
else:
print("\nKeine Warnungen.")
# ---------------------------------------------------------------------------
# Interaktive Assistenten
def _choose_model(model_class: str, models: list[str]) -> str:
"""Interaktive Auswahl eines Modells für eine Modellklasse."""
print(f"\nFür '{model_class}' ist kein Modell zugeordnet.")
if models:
print("Verfügbare Modelle (aus opencode.json(c)):")
for i, model in enumerate(models, 1):
print(f" {i}. {model}")
print(" 0. Modell-ID manuell eingeben")
while True:
choice = input("Auswahl: ").strip()
if choice == "0":
return input("Modell-ID (provider/model): ").strip()
if choice.isdigit() and 1 <= int(choice) <= len(models):
return models[int(choice) - 1]
print("Ungültige Auswahl.")
else:
print("Keine Modelle in opencode.json(c) gefunden.")
return input("Modell-ID (provider/model): ").strip()
def cmd_init(args: argparse.Namespace) -> int:
mapping_path = Path(args.mapping) if args.mapping else DEFAULT_MAPPING_PATH
if mapping_path.exists():
print(f"Mapping existiert bereits: {mapping_path}")
else:
mapping_path.parent.mkdir(parents=True, exist_ok=True)
mapping_path.write_text(INIT_TEMPLATE, encoding="utf-8")
print(f"Angelegt: {mapping_path}")
if args.fill:
missing = [k for k, v in load_mapping(mapping_path).items()
if v == "PLACEHOLDER"]
models = available_models()
if missing and not models:
raise OmosError(
"Keine Modelle in ~/.config/opencode/opencode.json(c) gefunden. "
"Trage die Modell-IDs manuell ein oder richte opencode ein."
)
data = load_yaml(mapping_path)
for model_class in missing:
data["model_classes"][model_class] = _choose_model(model_class, models)
mapping_path.write_text(
yaml.safe_dump(data, sort_keys=False, allow_unicode=True),
encoding="utf-8",
)
remaining = sum(1 for v in data["model_classes"].values()
if v == "PLACEHOLDER")
if remaining:
print(f"\n{remaining} Eintrag/Einträge bleiben PLACEHOLDER "
"bitte manuell ergänzen.")
print("\nNächster Schritt:")
print(f" omos.py check --package <package-dir> --mapping {mapping_path}")
return 0
def cmd_setup(args: argparse.Namespace) -> int:
"""Geführter Ablauf: Profil lesen → Mapping auffüllen → rendern."""
package_dir = Path(args.package).resolve()
profile, profile_path = load_team_profile(package_dir, args.team_file)
mapping_path = Path(args.mapping) if args.mapping else DEFAULT_MAPPING_PATH
print(f"Team-Profil: {profile['id']}")
if profile.get("description"):
print(f" {str(profile['description']).strip()}")
print("Rollen:")
for role in profile["roles"]:
mcps = (role.get("capabilities", {}) or {}).get("mcps", [])
skills = (role.get("capabilities", {}) or {}).get("skills", [])
print(f" - {role['id']}{role.get('model_class')} "
f"(MCPs: {', '.join(mcps) or 'keine'}; "
f"Skills: {', '.join(skills) or 'keine'})")
if not mapping_path.exists():
print(f"\nMapping fehlt lege an: {mapping_path}")
args_init = argparse.Namespace(mapping=str(mapping_path), fill=True)
cmd_init(args_init)
classes = load_mapping(mapping_path)
needed = sorted({r.get("model_class") for r in profile["roles"]
if r.get("model_class")} - set(classes))
models = available_models()
changed = False
for model_class in needed:
chosen = _choose_model(model_class, models)
classes[model_class] = {"model": chosen, "# selected_by": "user"}
changed = True
if changed:
mapping_path.write_text(
yaml.safe_dump(classes, sort_keys=False, allow_unicode=True),
encoding="utf-8",
)
print(f"\nMapping aktualisiert: {mapping_path}")
print("\nRendern …")
args_render = argparse.Namespace(
package=str(package_dir),
team_file=args.team_file,
mapping=str(mapping_path),
output=args.output,
dry_run=args.dry_run,
)
return cmd_render(args_render)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="omos",
description=(
"Adapter zwischen APM-Team-Profilen und oh-my-opencode-slim. "
"Übersetzt deterministisch: Team-Profil + lokales Modell-Mapping "
"-> OMOS-Preset."
),
)
sub = parser.add_subparsers(dest="command", required=True)
def common(p: argparse.ArgumentParser) -> None:
p.add_argument(
"--package",
default=".",
help="Pfad zum APM-Package (mit team-profile.yaml)",
)
p.add_argument(
"--team-file",
help="Alternativer Dateiname des Team-Profils (Default: team-profile.yaml)",
)
p.add_argument(
"--mapping",
help=f"Pfad zur Modell-Mapping-Datei (Default: {DEFAULT_MAPPING_PATH})",
)
p_render = sub.add_parser("render", help="Preset generieren und schreiben")
common(p_render)
p_render.add_argument(
"--output",
help=f"Zieldatei (Default: {DEFAULT_OUTPUT} relativ zum Package)",
)
p_render.add_argument(
"--dry-run",
action="store_true",
help="Nur anzeigen, nichts schreiben",
)
p_render.set_defaults(func=cmd_render)
p_check = sub.add_parser("check", help="Validieren ohne zu schreiben")
common(p_check)
p_check.set_defaults(func=cmd_check)
p_init = sub.add_parser(
"init",
help="Modell-Mapping-Datei anlegen und optional Platzhalter füllen",
)
p_init.add_argument(
"--mapping",
help=f"Zielpfad (Default: {DEFAULT_MAPPING_PATH})",
)
p_init.add_argument(
"--fill",
action="store_true",
help="Fehlende Modellklassen interaktiv aus opencode.json(c)-Modellen wählen",
)
p_init.set_defaults(func=cmd_init)
p_setup = sub.add_parser(
"setup",
help="Geführt: Profil lesen, Mapping auffüllen, rendern",
)
common(p_setup)
p_setup.add_argument(
"--output",
help=f"Zieldatei (Default: {DEFAULT_OUTPUT} relativ zum Package)",
)
p_setup.add_argument(
"--dry-run",
action="store_true",
help="Nur anzeigen, nichts schreiben",
)
p_setup.set_defaults(func=cmd_setup)
args = parser.parse_args(argv)
try:
return args.func(args)
except OmosError as exc:
print(f"Fehler: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())