roster

roster

YAML-driven agent and tool roster.

Loads per-agent definition files from bundled package resources and optional user extensions, deserializes them into the existing dataclass types, and provides the same query API that headless_providers and auth expose today.

Directory layout::

resources/agents/claude.yaml      (bundled, shipped in wheel)
resources/agents/codex.yaml
...
~/.config/terok/agent/agents/      (user overrides / additions)

MountDef(host_dir, container_path, label) dataclass

A shared directory mount derived from the agent roster.

host_dir instance-attribute

Directory name under mounts_dir() (e.g. "_codex-config").

container_path instance-attribute

Mount point inside the container (e.g. "/home/dev/.codex").

label instance-attribute

Human-readable label (e.g. "Codex config").

CredentialProxyRoute(provider, route_prefix, upstream, auth_header='Authorization', auth_prefix='Bearer ', credential_type='api_key', credential_file='', phantom_env=dict(), oauth_phantom_env=dict(), base_url_env='', socket_path='', socket_env='', shared_config_patch=None, oauth_refresh=None) dataclass

Proxy route config parsed from a credential_proxy: YAML section.

Used to generate the routes.json that the credential proxy server reads.

provider instance-attribute

Agent/tool name (e.g. "claude").

route_prefix instance-attribute

Path prefix in the proxy (e.g. "claude" → /claude/v1/...).

upstream instance-attribute

Upstream API base URL (e.g. "https://api.anthropic.com").

auth_header = 'Authorization' class-attribute instance-attribute

HTTP header name for the real credential.

auth_prefix = 'Bearer ' class-attribute instance-attribute

Prefix before the token value in the auth header.

credential_type = 'api_key' class-attribute instance-attribute

Type of credential: "oauth", "api_key", "oauth_token", "pat".

credential_file = '' class-attribute instance-attribute

Credential file path relative to the auth mount.

phantom_env = field(default_factory=dict) class-attribute instance-attribute

Phantom env vars for API-key credentials (e.g. {"ANTHROPIC_API_KEY": true}).

oauth_phantom_env = field(default_factory=dict) class-attribute instance-attribute

Phantom env vars for OAuth credentials (e.g. {"CLAUDE_CODE_OAUTH_TOKEN": true}).

When the stored credential type is "oauth" and this is non-empty, these env vars are injected instead of :attr:phantom_env.

base_url_env = '' class-attribute instance-attribute

Env var to override with proxy URL (e.g. "ANTHROPIC_BASE_URL").

socket_path = '' class-attribute instance-attribute

Unix socket path for socat bridge (e.g. "/tmp/terok-claude-proxy.sock").

socket_env = '' class-attribute instance-attribute

Env var that receives :attr:socket_path (e.g. "ANTHROPIC_UNIX_SOCKET").

shared_config_patch = None class-attribute instance-attribute

Optional shared config patch applied after auth (e.g. Vibe's config.toml).

oauth_refresh = None class-attribute instance-attribute

OAuth refresh config: {token_url, client_id, scope}.

SidecarSpec(tool_name, env_map=dict()) dataclass

Sidecar container configuration parsed from a sidecar: YAML section.

Tools with sidecar specs run in a separate lightweight L1 image (no agent CLIs) and receive the real API key instead of phantom tokens.

tool_name instance-attribute

Tool identifier used to select the Jinja2 install block in the template.

env_map = field(default_factory=dict) class-attribute instance-attribute

Maps container env var names to credential dict keys.

Example: {"CODERABBIT_API_KEY": "key"} reads cred["key"] and injects it as CODERABBIT_API_KEY.

AgentRoster(_providers=dict(), _auth_providers=dict(), _proxy_routes=dict(), _sidecar_specs=dict(), _mounts=(), _agent_names=(), _all_names=()) dataclass

Loaded roster of agents and tools from YAML definitions.

Provides the same query API as the legacy hardcoded dicts.

providers property

All headless agent providers (kind: agent only).

auth_providers property

All auth providers (agents + tools with auth: section).

agent_names property

Names of kind: agent entries (for CLI completion).

all_names property

Names of all entries (agents + tools).

mounts property

All shared directory mounts (auth dirs + explicit mounts: sections).

Deduplicated by host_dir — if auth and mounts define the same directory, only one entry is returned.

sidecar_specs property

All sidecar tool specs, keyed by tool name.

proxy_routes property

All credential proxy routes, keyed by provider name.

get_provider(name, *, default_agent=None)

Resolve a provider name to a HeadlessProvider.

Falls back to default_agent, then "claude". Raises SystemExit if the resolved name is unknown.

Source code in src/terok_agent/roster.py

def get_provider(
    self, name: str | None, *, default_agent: str | None = None
) -> HeadlessProvider:
    """Resolve a provider name to a ``HeadlessProvider``.

    Falls back to *default_agent*, then ``"claude"``.
    Raises ``SystemExit`` if the resolved name is unknown.
    """
    resolved = name or default_agent or "claude"
    provider = self._providers.get(resolved)
    if provider is None:
        valid = ", ".join(sorted(self._providers))
        raise SystemExit(f"Unknown headless provider {resolved!r}. Valid providers: {valid}")
    return provider

get_auth_provider(name)

Look up an auth provider by name.

Raises SystemExit if the name is unknown.

Source code in src/terok_agent/roster.py

def get_auth_provider(self, name: str) -> AuthProvider:
    """Look up an auth provider by name.

    Raises ``SystemExit`` if the name is unknown.
    """
    info = self._auth_providers.get(name)
    if info is None:
        available = ", ".join(sorted(self._auth_providers))
        raise SystemExit(f"Unknown auth provider: {name!r}. Available: {available}")
    return info

get_sidecar_spec(name)

Look up a sidecar spec by tool name.

Raises SystemExit if the name has no sidecar configuration.

Source code in src/terok_agent/roster.py

def get_sidecar_spec(self, name: str) -> SidecarSpec:
    """Look up a sidecar spec by tool name.

    Raises ``SystemExit`` if the name has no sidecar configuration.
    """
    spec = self._sidecar_specs.get(name)
    if spec is None:
        available = ", ".join(sorted(self._sidecar_specs)) or "(none)"
        raise SystemExit(f"No sidecar config for {name!r}. Available: {available}")
    return spec

generate_routes_json()

Generate the routes.json content for the credential proxy server.

Returns a JSON string mapping route prefixes to upstream config.

Source code in src/terok_agent/roster.py

def generate_routes_json(self) -> str:
    """Generate the ``routes.json`` content for the credential proxy server.

    Returns a JSON string mapping route prefixes to upstream config.
    """
    import json

    routes: dict[str, dict[str, object]] = {}
    prefix_owners: dict[str, str] = {}
    for route in self._proxy_routes.values():
        existing = prefix_owners.get(route.route_prefix)
        if existing is not None:
            raise ValueError(
                f"Duplicate route prefix {route.route_prefix!r}: "
                f"providers {existing!r} and {route.provider!r}"
            )
        prefix_owners[route.route_prefix] = route.provider
        entry: dict[str, object] = {
            "upstream": route.upstream,
            "auth_header": route.auth_header,
            "auth_prefix": route.auth_prefix,
        }
        if route.oauth_refresh:
            entry["oauth_refresh"] = route.oauth_refresh
        routes[route.provider] = entry
    return json.dumps(routes, indent=2)

collect_all_auto_approve_env()

Merge auto_approve.env from all providers into one dict.

Source code in src/terok_agent/roster.py

def collect_all_auto_approve_env(self) -> dict[str, str]:
    """Merge ``auto_approve.env`` from all providers into one dict."""
    merged: dict[str, str] = {}
    for p in self._providers.values():
        for key, value in p.auto_approve_env.items():
            if key in merged and merged[key] != value:
                raise ValueError(
                    f"Conflicting auto_approve_env for {key!r}: "
                    f"{merged[key]!r} vs {value!r} (provider {p.name!r})"
                )
            merged[key] = value
    return merged

collect_opencode_provider_env()

Collect env vars for all OpenCode-based providers.

Source code in src/terok_agent/roster.py

def collect_opencode_provider_env(self) -> dict[str, str]:
    """Collect env vars for all OpenCode-based providers."""
    env: dict[str, str] = {}
    for p in self._providers.values():
        if p.opencode_config is not None:
            env.update(p.opencode_config.to_env(p.name))
    return env

load_roster()

Load the agent roster from bundled YAML + user overrides.

Bundled agents in resources/agents/*.yaml are loaded first, then user files in ~/.config/terok/agent/agents/*.yaml are deep-merged on top (allowing field-level overrides or entirely new agents).

Source code in src/terok_agent/roster.py

def load_roster() -> AgentRoster:
    """Load the agent roster from bundled YAML + user overrides.

    Bundled agents in ``resources/agents/*.yaml`` are loaded first, then
    user files in ``~/.config/terok/agent/agents/*.yaml`` are deep-merged
    on top (allowing field-level overrides or entirely new agents).
    """
    raw = _load_bundled_agents()

    # Deep-merge user overrides on top of bundled definitions
    for name, user_data in _load_user_agents().items():
        if name in raw:
            raw[name] = deep_merge(raw[name], user_data)
        else:
            raw[name] = user_data

    providers: dict[str, HeadlessProvider] = {}
    auth_providers: dict[str, AuthProvider] = {}
    proxy_routes: dict[str, CredentialProxyRoute] = {}
    sidecar_specs: dict[str, SidecarSpec] = {}
    agent_names: list[str] = []
    all_names: list[str] = []

    # Collect mounts from all entries — deduplicate by host_dir
    seen_mounts: dict[str, MountDef] = {}

    for name, data in sorted(raw.items()):
        kind = data.get("kind", "native")
        if kind != "runtime":
            all_names.append(name)

        # Agent kinds (native, opencode, bridge) get a HeadlessProvider;
        # tools and runtime entries only contribute auth/mounts.
        if kind not in ("tool", "runtime"):
            agent_names.append(name)
            providers[name] = _to_headless_provider(name, data)

        # Auth: explicit auth section, or auto-derived from opencode config
        auth_prov = _to_auth_provider(name, data)
        if auth_prov is not None:
            auth_providers[name] = auth_prov
            # Auth providers also contribute a mount
            if auth_prov.host_dir_name not in seen_mounts:
                seen_mounts[auth_prov.host_dir_name] = MountDef(
                    host_dir=auth_prov.host_dir_name,
                    container_path=auth_prov.container_mount,
                    label=f"{auth_prov.label} config",
                )
        elif kind not in ("tool", "runtime"):
            oc_auth = _derive_opencode_auth(name, data)
            if oc_auth is not None:
                auth_providers[name] = oc_auth
                if oc_auth.host_dir_name not in seen_mounts:
                    seen_mounts[oc_auth.host_dir_name] = MountDef(
                        host_dir=oc_auth.host_dir_name,
                        container_path=oc_auth.container_mount,
                        label=f"{oc_auth.label} config",
                    )

        # Explicit mounts section
        for m in data.get("mounts", ()):
            hd = m["host_dir"]
            if hd not in seen_mounts:
                seen_mounts[hd] = MountDef(
                    host_dir=hd,
                    container_path=m["container_path"],
                    label=m.get("label", name),
                )

        # Credential proxy route
        proxy_route = _to_proxy_route(name, data)
        if proxy_route is not None:
            proxy_routes[name] = proxy_route

        # Sidecar spec
        sidecar = _to_sidecar_spec(name, data)
        if sidecar is not None:
            sidecar_specs[name] = sidecar

    return AgentRoster(
        _providers=providers,
        _auth_providers=auth_providers,
        _proxy_routes=proxy_routes,
        _sidecar_specs=sidecar_specs,
        _mounts=tuple(seen_mounts.values()),
        _agent_names=tuple(agent_names),
        _all_names=tuple(all_names),
    )

get_roster() cached

Return the singleton roster instance (loaded once, cached).

Source code in src/terok_agent/roster.py

@lru_cache(maxsize=1)
def get_roster() -> AgentRoster:
    """Return the singleton roster instance (loaded once, cached)."""
    return load_roster()

ensure_proxy_routes(cfg=None)

Generate routes.json from the YAML roster and write it to disk.

The routes file is written to the path configured in :class:~terok_sandbox.SandboxConfig (typically ~/.local/share/terok/proxy/routes.json).

When cfg is None, falls back to standalone defaults.

Returns the path to the written file.

Source code in src/terok_agent/roster.py

def ensure_proxy_routes(cfg: SandboxConfig | None = None) -> Path:
    """Generate ``routes.json`` from the YAML roster and write it to disk.

    The routes file is written to the path configured in
    :class:`~terok_sandbox.SandboxConfig` (typically
    ``~/.local/share/terok/proxy/routes.json``).

    When *cfg* is ``None``, falls back to standalone defaults.

    Returns the path to the written file.
    """
    from terok_sandbox import SandboxConfig

    if cfg is None:
        cfg = SandboxConfig()
    path = cfg.proxy_routes_path
    import os
    import tempfile

    path.parent.mkdir(parents=True, exist_ok=True)
    content = get_roster().generate_routes_json() + "\n"
    fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    tmp = Path(tmp_name)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
            f.flush()
            os.fsync(f.fileno())
        tmp.replace(path)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise
    return path