feat(ai): model profiles, capability discovery, and agentless /ai list|models

Make connecting any model a config step, not a code change:
- models.toml named profiles (api_key_env names an env var, never the key)
- providers gain available_models(); add preflight + --list-models/--check
- /ai list and /ai models in-room; client probes local Ollama for
  /ai models when no agent is running, and /ai list hints to summon one
- docs/providers.md provider guide + examples/echo_provider.py
- README: command table, AI section, layout updated

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
leetcrypt
2026-06-01 15:25:07 -07:00
parent 4fd1b70cb3
commit 65df12de9e
10 changed files with 592 additions and 21 deletions
+67 -7
View File
@@ -14,40 +14,100 @@ Examples
python -m cmd_chat.agent 127.0.0.1 3000 --provider openai \
--base-url https://api.groq.com/openai/v1 --model llama-3.1-70b --password hunter2
# a named profile from models.toml (provider + model + endpoint + key env)
python -m cmd_chat.agent 127.0.0.1 3000 --profile groq-llama --password hunter2
# a custom provider you wrote
python -m cmd_chat.agent 127.0.0.1 3000 --provider mypkg.mod:MyProvider
# discovery / preflight (no room join)
python -m cmd_chat.agent --profile groq-llama --list-models
python -m cmd_chat.agent --profile groq-llama --check
"""
from __future__ import annotations
import argparse
import sys
from .bridge import AgentBridge
from .providers import make_provider
from .profiles import load_profiles, provider_from_profile
from .providers import make_provider, preflight
def _build_provider(args, ap):
"""Resolve a Provider from either --profile or the explicit flags."""
if args.profile:
profiles = load_profiles(args.models_file)
if args.profile not in profiles:
known = ", ".join(profiles) or "(none — create models.toml)"
ap.error(f"unknown profile '{args.profile}'. known: {known}")
prof = profiles[args.profile]
provider = provider_from_profile(
prof, name=args.profile, model=args.model, base_url=args.base_url
)
# Profile may also supply non-provider defaults.
if args.system is None and prof.get("system"):
args.system = prof["system"]
if args.context_window == 12 and prof.get("context_window"):
args.context_window = int(prof["context_window"])
return provider
opts: dict = {}
if args.base_url and (args.provider == "openai" or ":" in args.provider):
opts["base_url"] = args.base_url
return make_provider(args.provider, model=args.model, **opts)
def main() -> None:
ap = argparse.ArgumentParser(
prog="cmd_chat.agent", description="hack-house AI agent bridge (PoC)"
)
ap.add_argument("server")
ap.add_argument("port", type=int)
ap.add_argument("server", nargs="?", help="room host (omit with --list-models/--check)")
ap.add_argument("port", type=int, nargs="?", help="room port")
ap.add_argument("--name", default="oracle", help="agent's room display name")
ap.add_argument("--password", default=None, help="room password")
ap.add_argument("--provider", default="ollama",
help="ollama | anthropic | openai | module:Class")
ap.add_argument("--profile", default=None,
help="named profile from models.toml (overrides --provider/--model)")
ap.add_argument("--models-file", default=None,
help="path to models.toml (default: $HH_MODELS_FILE, ./models.toml, ~/.config/hh/models.toml)")
ap.add_argument("--model", default=None, help="model name (provider default if omitted)")
ap.add_argument("--base-url", default=None, help="endpoint for openai-compatible providers")
ap.add_argument("--system", default=None, help="override the system prompt")
ap.add_argument("--context-window", type=int, default=12)
ap.add_argument("--list-models", action="store_true",
help="list models the backend can serve, then exit")
ap.add_argument("--check", action="store_true",
help="run a reachability/model preflight, then exit (0 ok, 1 fail)")
ap.add_argument("--insecure", action="store_true", help="skip TLS cert verification")
ap.add_argument("--no-tls", action="store_true", help="plain ws/http (local/Tailscale)")
args = ap.parse_args()
opts: dict = {}
if args.base_url and (args.provider == "openai" or ":" in args.provider):
opts["base_url"] = args.base_url
provider = make_provider(args.provider, model=args.model, **opts)
provider = _build_provider(args, ap)
# Discovery / preflight modes never join a room.
if args.list_models:
discover = getattr(provider, "available_models", None)
if discover is None:
ap.error(f"provider '{provider.name}' has no model discovery")
for m in discover():
print(m)
return
if args.check:
ok, msg = preflight(provider)
print(("ok: " if ok else "FAIL: ") + msg, file=sys.stderr if not ok else sys.stdout)
sys.exit(0 if ok else 1)
if args.server is None or args.port is None:
ap.error("server and port are required to join a room")
# Non-fatal preflight: warn early, but still try (discovery may be blocked
# while completion works).
ok, msg = preflight(provider)
if not ok:
print(f"⚠ preflight: {msg}", file=sys.stderr)
bridge = AgentBridge(
args.server, args.port, name=args.name, provider=provider,
+38 -1
View File
@@ -62,7 +62,44 @@ class AgentBridge(Client):
frame = json.dumps({"_ai": "typing", "name": self.name, "on": on})
await ws.send(self.room_fernet.encrypt(frame.encode()).decode())
async def _send_chat(self, ws, text: str) -> None:
await ws.send(self.room_fernet.encrypt(text.encode()).decode())
def _command_reply(self, question: str) -> str | None:
"""Canned reply for a reserved verb, else None.
Handled locally so it never spends a model call:
- ``list`` → this agent's roster line (who's here / what it runs). With
several agents present each answers for itself, forming the roster.
- ``models`` → what the configured backend can serve (in-room
--list-models)."""
verb = question.strip().lower()
if verb == "list":
return (f"{self.name} (ai) here — {self.provider.name}/"
f"{self.provider.model}, context {self.context_window}")
if verb != "models":
return None
discover = getattr(self.provider, "available_models", None)
if discover is None:
return f"{self.provider.name}: model discovery not supported."
try:
models = discover()
except Exception as e: # noqa: BLE001 — report unreachable backend in-room
return f"[ai error: cannot reach {self.provider.name}: {e}]"
if not models:
return f"{self.provider.name}: no models reported."
# One line: the TUI collapses embedded newlines, so bracket the active
# model instead of using a multi-line, marker-prefixed list.
mark = lambda m: f"[{m}]" if m == self.provider.model else m # noqa: E731
listing = ", ".join(mark(m) for m in models)
return f"{self.provider.name} models ([active]): {listing}"
async def _answer(self, ws, question: str, asker: str) -> None:
canned = self._command_reply(question)
if canned is not None:
await self._send_chat(ws, canned)
self.success(f"answered /ai {question.strip().lower()} for {asker}")
return
self.transcript.append(Msg("user", f"{asker}: {question}"))
await self._send_typing(ws, True)
try:
@@ -77,7 +114,7 @@ class AgentBridge(Client):
await self._send_typing(ws, False)
reply = reply.strip() or "[empty reply]"
self.transcript.append(Msg("assistant", reply))
await ws.send(self.room_fernet.encrypt(reply.encode()).decode())
await self._send_chat(ws, reply)
self.success(f"replied to {asker}")
async def run_async(self) -> None:
+102
View File
@@ -0,0 +1,102 @@
"""Named model profiles for the hack-house AI agent.
A *profile* maps a friendly name (``groq-llama``, ``local``, ``claude``) to a
provider + model + endpoint, so operators type ``--profile groq-llama`` instead
of remembering ``--provider openai --base-url … --model …``. This mirrors the
``models:`` list in Continue.dev and the ``model_list`` in a LiteLLM proxy:
each entry is ``{provider, model, base_url, api_key_env}``.
Secrets are **never** stored here — ``api_key_env`` names an environment
variable to read the key from, keeping the file safe to commit and share.
Lookup order (first hit wins):
1. ``$HH_MODELS_FILE``
2. ``./models.toml`` (cwd)
3. ``~/.config/hh/models.toml``
"""
from __future__ import annotations
import os
from pathlib import Path
try: # stdlib on 3.11+, falls back to the `tomli` backport on 3.10
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib # type: ignore[no-redef]
from .providers import Provider, make_provider
_RECOGNIZED = {"provider", "model", "base_url", "host", "api_key_env",
"system", "context_window"}
def _candidate_paths(explicit: str | None) -> list[Path]:
if explicit:
return [Path(explicit).expanduser()]
paths = []
env = os.environ.get("HH_MODELS_FILE")
if env:
paths.append(Path(env).expanduser())
paths.append(Path.cwd() / "models.toml")
paths.append(Path.home() / ".config" / "hh" / "models.toml")
return paths
def find_profiles_file(explicit: str | None = None) -> Path | None:
for p in _candidate_paths(explicit):
if p.is_file():
return p
return None
def load_profiles(explicit: str | None = None) -> dict[str, dict]:
"""Return ``{name: profile_dict}`` from the first models.toml found."""
path = find_profiles_file(explicit)
if path is None:
return {}
with path.open("rb") as fh:
data = tomllib.load(fh)
profiles: dict[str, dict] = {}
for name, body in data.items():
if not isinstance(body, dict) or "provider" not in body:
continue # skip non-profile tables / malformed entries
unknown = set(body) - _RECOGNIZED
if unknown:
raise ValueError(
f"profile '{name}': unknown key(s) {', '.join(sorted(unknown))}"
)
profiles[name] = body
return profiles
def provider_from_profile(prof: dict, *, name: str = "?",
model: str | None = None,
base_url: str | None = None) -> Provider:
"""Build a :class:`Provider` from a profile dict.
``model`` / ``base_url`` (CLI flags) override the profile when given. The
api key is read from ``$<api_key_env>`` and passed only to providers that
accept one, so an Ollama profile never sees a stray ``api_key`` kwarg.
"""
spec = prof["provider"]
custom = ":" in spec
opts: dict = {}
mdl = model or prof.get("model")
bu = base_url or prof.get("base_url")
if bu and (spec == "openai" or custom):
opts["base_url"] = bu
if spec == "ollama" and prof.get("host"):
opts["host"] = prof["host"]
key_env = prof.get("api_key_env")
if key_env and (spec in ("openai", "anthropic") or custom):
key = os.environ.get(key_env)
if not key:
raise SystemExit(
f"profile '{name}': ${key_env} is not set — export it first"
)
opts["api_key"] = key
return make_provider(spec, model=mdl, **opts)
+53
View File
@@ -30,6 +30,11 @@ class Provider(Protocol):
def complete(self, system: str, messages: list[Msg]) -> str:
...
# Optional: list models the backend can serve, for discovery/preflight.
# Providers that can't enumerate (e.g. a bespoke endpoint) may omit this.
def available_models(self) -> list[str]:
...
class OllamaProvider:
"""Local Ollama (default, recommended). No API key — privacy-preserving."""
@@ -52,6 +57,11 @@ class OllamaProvider:
r.raise_for_status()
return (r.json().get("message", {}).get("content") or "").strip()
def available_models(self) -> list[str]:
r = requests.get(f"{self.host}/api/tags", timeout=self.timeout)
r.raise_for_status()
return [m.get("name", "") for m in r.json().get("models", [])]
class AnthropicProvider:
"""Anthropic Messages API. Cloud — opt-in. Needs ANTHROPIC_API_KEY."""
@@ -92,6 +102,15 @@ class AnthropicProvider:
blocks = r.json().get("content", [])
return "".join(b.get("text", "") for b in blocks).strip()
def available_models(self) -> list[str]:
r = requests.get(
"https://api.anthropic.com/v1/models",
timeout=self.timeout,
headers={"x-api-key": self.api_key, "anthropic-version": "2023-06-01"},
)
r.raise_for_status()
return [m.get("id", "") for m in r.json().get("data", [])]
class OpenAICompatibleProvider:
"""OpenAI-style /chat/completions — OpenAI, Groq, Together, local vLLM, etc."""
@@ -120,6 +139,14 @@ class OpenAICompatibleProvider:
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"].strip()
def available_models(self) -> list[str]:
headers = {}
if self.api_key:
headers["authorization"] = f"Bearer {self.api_key}"
r = requests.get(f"{self.base_url}/models", headers=headers, timeout=self.timeout)
r.raise_for_status()
return [m.get("id", "") for m in r.json().get("data", [])]
_BUILTINS = {
"ollama": OllamaProvider,
@@ -144,3 +171,29 @@ def make_provider(spec: str, model: str | None = None, **opts) -> Provider:
if model is not None:
opts["model"] = model
return cls(**opts)
def preflight(provider: Provider) -> tuple[bool, str]:
"""Cheap reachability + model-presence check before joining a room.
Returns ``(ok, message)``. Lets ``/ai start`` fail fast with a clear reason
(backend down / model not pulled / key missing) instead of erroring on the
first question. Providers without ``available_models`` are assumed reachable.
"""
discover = getattr(provider, "available_models", None)
if discover is None:
return True, f"{provider.name}: no discovery endpoint — assuming reachable"
try:
models = discover()
except Exception as e: # noqa: BLE001 — any failure means "not reachable yet"
return False, f"{provider.name}: cannot reach backend ({e})"
if provider.model in models:
return True, f"{provider.name}/{provider.model}: reachable"
if models:
sample = ", ".join(models[:8])
more = "" if len(models) > 8 else ""
return False, (
f"{provider.name}: model '{provider.model}' not available. "
f"reachable models: {sample}{more}"
)
return True, f"{provider.name}: reachable (empty model list — skipping check)"