Skip to content

Commands

octomate_cli.main

The root Typer app: client commands at the root, server commands under service.

Every group is mounted whatever the machine holds — server or client — so the surface stays discoverable; a command that needs the absent half says so when invoked rather than vanishing.

app module-attribute 

app = typer.Typer(
    help="Octomate operator CLI.", no_args_is_help=True
)

options 

options(
    show_version: Annotated[
        bool,
        Option(
            "--version",
            is_eager=True,
            help="Show installed package versions.",
        ),
    ] = False,
) -> None
Source code in cli/octomate_cli/main.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@app.callback(invoke_without_command=True)
def options(
    show_version: Annotated[
        bool,
        typer.Option(
            "--version", is_eager=True, help="Show installed package versions."
        ),
    ] = False,
) -> None:
    if show_version:
        for package in ("octomate-cli", "octomate-protocol", "octomate"):
            try:
                installed = version(package)
            except PackageNotFoundError:
                continue
            typer.echo(f"{package} {installed}")
        raise typer.Exit()

octomate_cli.config

The client's server addresses and credentials for hooks and native history reads.

A settings class, the way the deployment's own config is one, so the variables, the files and their precedence are declared in CLISettings rather than assembled by hand, and every command reads the one cli_settings() object rather than building its own. Owned here so a machine holding only octomate-cli never touches the server's config.

The stdlib hook scripts (emit.py, launch.py) cannot use it, and that is not an oversight to be tidied away later. They exist to stay off the import cost of a package on a hook the session blocks on, several times a turn: tomllib costs ~7ms against pydantic_settings' ~230ms, on a script whose whole budget is ~25ms. So they re-spell the variable names and the same file precedence in stdlib. Two tests in tests/cli/test_codex_emit.py are what stop them drifting — one holds the names against CLISettings.env, the other that both halves still pick the same file — so a renamed field fails a test rather than a session going unauthenticated.

console module-attribute 

console = Console(stderr=True)

USER_CONFIG module-attribute 

USER_CONFIG = '~/.config/octomate/cli.toml'

PROJECT_CONFIG module-attribute 

PROJECT_CONFIG = '.octomate/cli.toml'

Scope

Bases: StrEnum

Source code in cli/octomate_cli/config.py 
43
44
45
class Scope(StrEnum):
    user = "user"
    project = "project"

user class-attribute instance-attribute 

user = 'user'

project class-attribute instance-attribute 

project = 'project'

DeepseekSettings pydantic-model

Bases: BaseModel

The local gateway used to read native DeepSeek history.

Fields:

  • url (str | None)
Source code in cli/octomate_cli/config.py 
69
70
71
72
73
74
75
76
77
78
79
class DeepseekSettings(BaseModel):
    """The local gateway used to read native DeepSeek history."""

    url: str | None = Field(
        default=None,
        description=(
            "The local DeepSeek gateway's launch URL, including its authentication "
            "token. Store in cli.toml so native hooks can authenticate even when "
            "DeepSeek removes DSH_* environment variables."
        ),
    )

url pydantic-field 

url: str | None = None

The local DeepSeek gateway's launch URL, including its authentication token. Store in cli.toml so native hooks can authenticate even when DeepSeek removes DSH_* environment variables.

CLISettings

Bases: BaseSettings

What this machine's client half knows, resolved the same way everywhere.

Reached through cli_settings() rather than constructed: one object per process, which is what a command or a hook is.

Source code in cli/octomate_cli/config.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class CLISettings(BaseSettings):
    """What this machine's client half knows, resolved the same way everywhere.

    Reached through `cli_settings()` rather than constructed: one object per process,
    which is what a command or a hook is.
    """

    model_config = SettingsConfigDict(
        extra="ignore",
        # The reason there is a prefix: the server's settings read `OCTOMATE__` with
        # `__` between the levels, so a client variable under that prefix would read
        # as a deployment key — and one of them, the retired deployment secret,
        # literally was.
        env_prefix="OCTOMATE_CLI_",
        # An exported-but-empty variable means unset, and the files under it still
        # answer — what a shell that cleared one intends.
        env_ignore_empty=True,
        # Weakest first: within one TOML source the later file wins, so the project's
        # own beats the machine's. That is how one directory aims at a different
        # server (a debug instance) without touching the default.
        toml_file=(USER_CONFIG, PROJECT_CONFIG),
    )

    url: str | None = Field(
        default=None,
        description=(
            "Octomate's base URL (`http://host:port`), resolved when a hook fires "
            "rather than pinned at install time — so switching servers is a config "
            "switch, not a re-install. An explicit `--url` beats it."
        ),
    )
    token: str | None = Field(
        default=None,
        description=(
            "A server-issued API token with hooks and/or mcp scope, obtained through "
            "the authenticated account API."
        ),
    )
    deepseek: DeepseekSettings = Field(
        default_factory=DeepseekSettings,
        description="DeepSeek's native history gateway settings.",
    )

    @classmethod
    def env(cls, field: str) -> str:
        """`CLISettings.env("token")` — the variable that field reads.

        The class is asked every time rather than a constant standing beside it, so
        no variable name is ever spelled by hand: this composes from the prefix
        pydantic is holding, and a renamed field changes what every message says.

        A method and not `__class_getitem__`, which would read better at the ~20 call
        sites: a subscript on a class is a *type* to a type checker, so pyright reads
        `CLISettings["token"]` as specialising the model and calls the result
        `type[CLISettings]` whatever the override returns. Every site would be
        mistyped, and silently so wherever the value is only interpolated.

        `.get` with a default rather than a subscript because the key is optional in
        `SettingsConfigDict`, and an absent prefix is the empty one.
        """
        return f"{cls.model_config.get('env_prefix', '')}{field}".upper()

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        """Which kinds of source there are, and in what order — per key, following
        Claude's own settings pattern: an explicit value wins, then the environment,
        then the files. Which files is `toml_file`'s to say, above.

        The environment is ahead of the files for the cases with no home to write
        into, a container or a CI step. Everywhere else the files are the durable
        home and what `configure` writes: they survive launch paths the environment
        does not — a GUI-launched editor never sourced a shell profile — and they are
        per-user, where an exported credential is inherited by everything that shell
        starts.

        No dotenv and no secrets dir: `./.env` on a server machine is the
        *deployment's* environment, and reading it here would let a server's file
        decide a person's credential.
        """
        return (init_settings, env_settings, TomlConfigSettingsSource(settings_cls))

model_config class-attribute instance-attribute 

model_config = SettingsConfigDict(
    extra="ignore",
    env_prefix="OCTOMATE_CLI_",
    env_ignore_empty=True,
    toml_file=(USER_CONFIG, PROJECT_CONFIG),
)

url class-attribute instance-attribute 

url: str | None = Field(
    default=None,
    description="Octomate's base URL (`http://host:port`), resolved when a hook fires rather than pinned at install time — so switching servers is a config switch, not a re-install. An explicit `--url` beats it.",
)

token class-attribute instance-attribute 

token: str | None = Field(
    default=None,
    description="A server-issued API token with hooks and/or mcp scope, obtained through the authenticated account API.",
)

deepseek class-attribute instance-attribute 

deepseek: DeepseekSettings = Field(
    default_factory=DeepseekSettings,
    description="DeepSeek's native history gateway settings.",
)

env classmethod 

env(field: str) -> str

CLISettings.env("token") — the variable that field reads.

The class is asked every time rather than a constant standing beside it, so no variable name is ever spelled by hand: this composes from the prefix pydantic is holding, and a renamed field changes what every message says.

A method and not __class_getitem__, which would read better at the ~20 call sites: a subscript on a class is a type to a type checker, so pyright reads CLISettings["token"] as specialising the model and calls the result type[CLISettings] whatever the override returns. Every site would be mistyped, and silently so wherever the value is only interpolated.

.get with a default rather than a subscript because the key is optional in SettingsConfigDict, and an absent prefix is the empty one.

Source code in cli/octomate_cli/config.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@classmethod
def env(cls, field: str) -> str:
    """`CLISettings.env("token")` — the variable that field reads.

    The class is asked every time rather than a constant standing beside it, so
    no variable name is ever spelled by hand: this composes from the prefix
    pydantic is holding, and a renamed field changes what every message says.

    A method and not `__class_getitem__`, which would read better at the ~20 call
    sites: a subscript on a class is a *type* to a type checker, so pyright reads
    `CLISettings["token"]` as specialising the model and calls the result
    `type[CLISettings]` whatever the override returns. Every site would be
    mistyped, and silently so wherever the value is only interpolated.

    `.get` with a default rather than a subscript because the key is optional in
    `SettingsConfigDict`, and an absent prefix is the empty one.
    """
    return f"{cls.model_config.get('env_prefix', '')}{field}".upper()

settings_customise_sources classmethod 

settings_customise_sources(
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]

Which kinds of source there are, and in what order — per key, following Claude's own settings pattern: an explicit value wins, then the environment, then the files. Which files is toml_file's to say, above.

The environment is ahead of the files for the cases with no home to write into, a container or a CI step. Everywhere else the files are the durable home and what configure writes: they survive launch paths the environment does not — a GUI-launched editor never sourced a shell profile — and they are per-user, where an exported credential is inherited by everything that shell starts.

No dotenv and no secrets dir: ./.env on a server machine is the deployment's environment, and reading it here would let a server's file decide a person's credential.

Source code in cli/octomate_cli/config.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
@classmethod
def settings_customise_sources(
    cls,
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
    """Which kinds of source there are, and in what order — per key, following
    Claude's own settings pattern: an explicit value wins, then the environment,
    then the files. Which files is `toml_file`'s to say, above.

    The environment is ahead of the files for the cases with no home to write
    into, a container or a CI step. Everywhere else the files are the durable
    home and what `configure` writes: they survive launch paths the environment
    does not — a GUI-launched editor never sourced a shell profile — and they are
    per-user, where an exported credential is inherited by everything that shell
    starts.

    No dotenv and no secrets dir: `./.env` on a server machine is the
    *deployment's* environment, and reading it here would let a server's file
    decide a person's credential.
    """
    return (init_settings, env_settings, TomlConfigSettingsSource(settings_cls))

user_config_path 

user_config_path() -> Path

~/.config/octomate/cli.toml — where CLI tools keep their config, git and ruff alike.

Source code in cli/octomate_cli/config.py 
56
57
58
59
def user_config_path() -> Path:
    """`~/.config/octomate/cli.toml` — where CLI tools keep their config, git and
    ruff alike."""
    return Path(USER_CONFIG).expanduser()

project_config_path 

project_config_path() -> Path

./.octomate/cli.toml — beside the directory a session runs in, the way .claude/settings.json scopes Claude. On a server machine .octomate/ is also the data directory; the two uses share the name, not their files.

Source code in cli/octomate_cli/config.py 
62
63
64
65
66
def project_config_path() -> Path:
    """`./.octomate/cli.toml` — beside the directory a session runs in, the way
    `.claude/settings.json` scopes Claude. On a server machine `.octomate/` is also
    the data directory; the two uses share the name, not their files."""
    return Path.cwd() / PROJECT_CONFIG

cli_settings cached 

cli_settings() -> CLISettings

This process's client config — the one object every command reads.

Typer has nothing like FastAPI's Depends to hand a command its config, and its one shared-state channel, ctx.obj, is untyped and reaches only commands invoked through the root app. So the dependency is this function, and every call site asks it rather than building a settings object of its own.

Cached because a process is one CLI invocation or one hook, and neither moves its own environment or working directory while it runs. Not for speed: a settings object costs ~0.13ms to build and no invocation wants more than two. The suite is the one process that plays many, and tests/conftest.py clears this between them.

Source code in cli/octomate_cli/config.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@cache
def cli_settings() -> CLISettings:
    """This process's client config — the one object every command reads.

    Typer has nothing like FastAPI's `Depends` to hand a command its config, and its
    one shared-state channel, `ctx.obj`, is untyped and reaches only commands invoked
    through the root app. So the dependency is this function, and every call site
    asks it rather than building a settings object of its own.

    Cached because a process is one CLI invocation or one hook, and neither moves its
    own environment or working directory while it runs. Not for speed: a settings
    object costs ~0.13ms to build and no invocation wants more than two. The suite is
    the one process that plays many, and `tests/conftest.py` clears this between them.
    """
    return CLISettings()

load_config 

load_config(path: Path) -> dict[str, object]

One file's table, {} when absent — what configure reads back before rewriting it, so keys it does not own survive. A file that does not parse is an operator's mistake worth a precise error, not a silent fallback.

Source code in cli/octomate_cli/config.py 
188
189
190
191
192
193
194
195
196
197
198
199
def load_config(path: Path) -> dict[str, object]:
    """One file's table, `{}` when absent — what `configure` reads back before
    rewriting it, so keys it does not own survive. A file that does not parse is an
    operator's mistake worth a precise error, not a silent fallback."""
    try:
        text = path.read_text()
    except OSError:
        return {}
    try:
        return tomllib.loads(text)
    except tomllib.TOMLDecodeError as error:
        raise typer.BadParameter(f"{path}: {error}") from None

configure 

configure(
    url: Annotated[
        str | None,
        Option(
            help="Octomate's base URL (http://host:port)."
        ),
    ] = None,
    token: Annotated[
        str | None,
        Option(
            help="A server-issued API token. Omitted, the token already resolving is kept."
        ),
    ] = None,
    dsh_url: Annotated[
        str | None,
        Option(
            help="Save DeepSeek's gateway launch URL (including ?token=) under [deepseek]."
        ),
    ] = None,
    scope: Annotated[
        Scope,
        Option(
            help="Which file to write: 'user' (~/.config/octomate/cli.toml) or 'project' (./.octomate/cli.toml, resolved first)."
        ),
    ] = Scope.user,
) -> None

Write the client config every hook, tail and MCP entry on this machine reads.

Sign in through Trunkline and issue a token through the account API first.

The environment still overrides both scopes, and a --url pinned at install time beats everything — the files are the durable floor, so ingest works from any launch path on a machine where octomate-cli is the only thing installed.

Source code in cli/octomate_cli/config.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def configure(
    url: Annotated[
        str | None,
        typer.Option(help="Octomate's base URL (http://host:port)."),
    ] = None,
    token: Annotated[
        str | None,
        typer.Option(
            help="A server-issued API token. Omitted, the token already resolving is kept."
        ),
    ] = None,
    dsh_url: Annotated[
        str | None,
        typer.Option(
            help="Save DeepSeek's gateway launch URL (including ?token=) under [deepseek]."
        ),
    ] = None,
    scope: Annotated[
        Scope,
        typer.Option(
            help="Which file to write: 'user' (~/.config/octomate/cli.toml) or "
            "'project' (./.octomate/cli.toml, resolved first)."
        ),
    ] = Scope.user,
) -> None:
    """Write the client config every hook, tail and MCP entry on this machine reads.

    Sign in through Trunkline and issue a token through the account API first.

    The environment still overrides both scopes, and a `--url` pinned at install time
    beats everything — the files are the durable floor, so ingest works from any
    launch path on a machine where octomate-cli is the only thing installed.
    """
    path = project_config_path() if scope is Scope.project else user_config_path()
    current = load_config(path)
    if url is not None:
        current["url"] = url
    if dsh_url is not None:
        current["deepseek"] = DeepseekSettings(url=dsh_url).model_dump()
    if token is None:
        token = cli_settings().token
    if token is not None:
        current["token"] = token
    current.pop("secret", None)

    content = tomlkit.dumps(current)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)
    path.chmod(0o600)  # it holds a credential
    # The one command that moves what `cli_settings()` answers, so it is the one that
    # has to drop the cached answer: anything reading after this wants the file just
    # written, not the object built before it existed.
    cli_settings.cache_clear()

    address = current.get("url") or f"(unset — hooks need ${CLISettings.env('url')})"
    typer.echo(f"Wrote {path}")
    typer.echo(f"  url:    {address}")
    typer.echo(
        f"  token: {'saved' if token is not None else 'unset — provide --token'}"
    )
    # Prose is left for rich to wrap; lines meant to be copied are kept short enough
    # to survive intact, since a token split across two lines is worse than no help.
    steps = [
        "Point your runtimes at it. Each writes down what it resolves here, so "
        "these come after the credential exists — and again if it moves:",
        "",
        "  [cyan]octomate claude hooks install[/]   [cyan]octomate claude mcp install[/]",
        "  [cyan]octomate codex hooks install[/]    [cyan]octomate codex mcp install[/]",
    ]
    console.print(Panel("\n".join(steps), title="[bold]next[/]", padding=(1, 2)))

octomate_cli.cli

Manage the operator CLI independently from the deployed service.

upgrade 

upgrade() -> None

Upgrade a standalone uv tool installation of octomate-cli.

Source code in cli/octomate_cli/cli.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def upgrade() -> None:
    """Upgrade a standalone uv tool installation of octomate-cli."""
    console = Console(stderr=True, highlight=False, soft_wrap=not sys.stderr.isatty())
    try:
        distribution("octomate")
    except PackageNotFoundError:
        pass
    else:
        console.print(
            "This environment contains the Octomate service. Run the standalone "
            "CLI instead; install it with: uv tool install octomate-cli",
            style="red",
        )
        raise typer.Exit(1)

    installed = distribution("octomate-cli")
    installer = (installed.read_text("INSTALLER") or "unknown").strip()
    prefix = Path(sys.prefix).resolve()
    uv = shutil.which("uv")
    try:
        if installer == "uv" and uv is not None:
            result = subprocess.run(
                [uv, "tool", "dir"], capture_output=True, text=True, check=True
            )
            tool_root = Path(result.stdout.strip()).resolve() / "octomate-cli"
            if (
                prefix == tool_root.resolve()
                and Path(str(installed.locate_file("")))
                .resolve()
                .is_relative_to(prefix)
                and (prefix / "uv-receipt.toml").is_file()
            ):
                executable = prefix / "bin" / "octomate"
                if not executable.is_file():
                    console.print(
                        "The installed CLI executable is missing. Repair it with: "
                        "uv tool install --reinstall octomate-cli",
                        style="red",
                    )
                    raise typer.Exit(1)
                console.print("Upgrading the standalone CLI…", style="cyan")
                subprocess.run([uv, "tool", "upgrade", "octomate-cli"], check=True)
                result = subprocess.run(
                    [str(executable), "--version"],
                    capture_output=True,
                    text=True,
                    check=True,
                )
                console.print(result.stdout.strip(), markup=False)
                return

            result = subprocess.run(
                [uv, "cache", "dir"], capture_output=True, text=True, check=True
            )
            if prefix.is_relative_to(Path(result.stdout.strip()).resolve()):
                console.print(
                    "This CLI runs in a temporary uv environment. Refresh the "
                    "invocation with: uvx --refresh --from octomate-cli octomate\n"
                    "For a persistent CLI: uv tool install octomate-cli",
                    style="red",
                )
                raise typer.Exit(1)
    except (OSError, subprocess.CalledProcessError) as error:
        console.print(f"CLI upgrade failed: {error}", style="red", markup=False)
        raise typer.Exit(1) from error

    console.print(
        f"Cannot verify standalone uv tool ownership at {prefix} "
        f"(installer: {installer}).",
        style="red",
        markup=False,
    )
    if installer == "pip":
        console.print(
            "Update this pip installation with: "
            f"{shlex.quote(sys.executable)} -m pip install --upgrade octomate-cli",
            markup=False,
        )
    else:
        console.print(
            "Use this installation's package manager, or install the standalone CLI "
            "with: uv tool install octomate-cli",
        )
    raise typer.Exit(1)

octomate_cli.service

Run the server and manage the desktop account's GUI service.

RELEASE_URL module-attribute 

RELEASE_URL = "https://api.github.com/repos/kalynnka/octomate/releases"

console module-attribute 

console = Console(
    stderr=True, markup=False, highlight=False
)

service_typer module-attribute 

service_typer = typer.Typer(
    help="Prepare macOS, Linux or Docker deployments; run the server and manage its macOS GUI service."
)

Release pydantic-model

Bases: BaseModel

GitHub release metadata used to select a server upgrade.

Fields:

Source code in cli/octomate_cli/service.py 
39
40
41
42
43
44
class Release(BaseModel):
    """GitHub release metadata used to select a server upgrade."""

    tag_name: str
    draft: bool
    prerelease: bool

tag_name pydantic-field 

tag_name: str

draft pydantic-field 

draft: bool

prerelease pydantic-field 

prerelease: bool

PlistService pydantic-model

Bases: BaseModel

A launchd job and its runtime configuration, loaded from a service plist.

Config:

  • hide_input_in_errors: True

Fields:

Source code in cli/octomate_cli/service.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
class PlistService(BaseModel):
    """A launchd job and its runtime configuration, loaded from a service plist."""

    model_config = ConfigDict(hide_input_in_errors=True)

    label: str = Field(alias="Label", pattern=r"^[a-zA-Z0-9_.-]+$")
    directory: Path = Field(alias="WorkingDirectory")
    user: None = Field(default=None, alias="UserName")
    group: None = Field(default=None, alias="GroupName")
    arguments: list[str] = Field(alias="ProgramArguments")
    environment: dict[str, str] = Field(alias="EnvironmentVariables", repr=False)
    stdout: Path = Field(alias="StandardOutPath")
    stderr: Path = Field(alias="StandardErrorPath")
    session_type: Literal["Aqua"] = Field(alias="LimitLoadToSessionType")
    program: None = Field(default=None, alias="Program")
    keep_alive: Literal[True] = Field(alias="KeepAlive")
    abandon_process_group: Literal[False] = Field(
        default=False, alias="AbandonProcessGroup"
    )

    @staticmethod
    def path() -> Path:
        return Path.home() / "Library/LaunchAgents/io.octomate.server.plist"

    @classmethod
    def load(cls) -> PlistService:
        if sys.platform != "darwin":
            raise typer.BadParameter("Service commands require macOS launchd.")
        # pwd is unavailable on Windows; client CLI imports must remain portable.
        import pwd

        if os.getuid() == 0:
            raise typer.BadParameter("Run as the desktop account, without sudo.")
        path = cls.path()
        try:
            if path.stat().st_uid != os.getuid():
                raise ValueError("The service definition must belong to this account.")
            with path.open("rb") as source:
                service = cls.model_validate(plistlib.load(source))
        except (OSError, ValueError) as error:
            raise typer.BadParameter(
                f"Invalid GUI service definition: {error}"
            ) from None
        account = pwd.getpwuid(os.getuid())
        if service.label != "io.octomate.server":
            raise typer.BadParameter("Unexpected service label.")
        if (
            len(service.arguments) != 3
            or service.arguments[1:] != ["service", "serve"]
            or not Path(service.arguments[0]).is_absolute()
            or Path(service.arguments[0]).parts[-3:] != (".venv", "bin", "octomate")
        ):
            raise typer.BadParameter(
                "The service must run <checkout>/.venv/bin/octomate service serve."
            )
        if not all(
            path.is_absolute()
            for path in (service.directory, service.stdout, service.stderr)
        ):
            raise typer.BadParameter(
                "Working directory and log paths must be absolute."
            )
        if not all(
            service.environment.get(key)
            for key in ("PATH", "OCTOMATE_HOME", "OCTOMATE_DB_URL")
        ):
            raise typer.BadParameter(
                "The service must set PATH, OCTOMATE_HOME and OCTOMATE_DB_URL."
            )
        if service.environment.get("HOME") != account.pw_dir or any(
            service.environment.get(key) != account.pw_name
            for key in ("USER", "LOGNAME")
        ):
            raise typer.BadParameter(
                "The service must use this desktop account's HOME, USER and LOGNAME."
            )
        return service

    @property
    def checkout(self) -> Path:
        return Path(self.arguments[0]).parents[2]

    @property
    def root(self) -> Path:
        return self.checkout.parent

    @property
    def domain(self) -> str:
        return f"gui/{os.getuid()}"

    @property
    def target(self) -> str:
        """The desktop launchd job identified by the plist's Label."""
        return f"{self.domain}/{self.label}"

    def require_gui(self) -> None:
        result = subprocess.run(
            ["/bin/launchctl", "print", self.domain], capture_output=True, text=True
        )
        if result.returncode:
            raise ValueError(
                f"Desktop session {self.domain} is unavailable. Log into the desktop first."
            )

    def loaded(self) -> bool:
        """Check whether the launchd job defined by this plist is loaded."""
        result = subprocess.run(
            ["/bin/launchctl", "print", self.target], capture_output=True, text=True
        )
        if result.returncode == 113:
            return False
        result.check_returncode()
        return True

    def launchctl(self, action: str, *arguments: str) -> None:
        """Apply an unprivileged action in the desktop launchd domain."""
        subprocess.run(["/bin/launchctl", action, *arguments], check=True)

    def stop(self) -> None:
        """Disable the job and wait for its process group to exit."""
        self.launchctl("disable", self.target)
        if not self.loaded():
            return
        result = subprocess.run(
            ["/bin/launchctl", "print", self.target],
            capture_output=True,
            text=True,
            check=True,
        )
        match = re.search(r"^\s*pid = ([0-9]+)\s*$", result.stdout, re.MULTILINE)
        group: int | None = None
        if match is not None:
            try:
                group = os.getpgid(int(match[1]))
            except ProcessLookupError:
                pass
        self.launchctl("bootout", self.target)
        deadline = time.monotonic() + 30
        while group is not None:
            try:
                os.killpg(group, 0)
            except ProcessLookupError:
                break
            except PermissionError:
                # macOS reports EPERM while a group contains only unreaped zombies.
                # Keep waiting for ESRCH under the same shutdown deadline.
                pass
            if time.monotonic() >= deadline:
                raise TimeoutError(
                    "The service process group did not stop within 30 seconds."
                )
            time.sleep(0.25)

    def maintenance(self, action: str, backup: DatabaseBackup | None = None) -> str:
        """Run maintenance with the plist's working directory and environment."""
        result = subprocess.run(
            [
                str(self.checkout / ".venv/bin/python"),
                "-m",
                "octomate_cli.deployment",
                action,
            ],
            cwd=self.directory,
            env={
                **self.environment,
                "OCTOMATE_DEPLOYMENT_ROOT": str(self.root),
            },
            input=backup.model_dump_json() if backup is not None else None,
            stdout=subprocess.PIPE,
            text=True,
            check=True,
        )
        return result.stdout.strip()

    def report(self, message: str) -> None:
        """Print service status and log it beside the plist's configured checkout."""
        console.print(message)
        logs = self.root / "logs"
        logs.mkdir(parents=True, exist_ok=True)
        with (logs / "server.log").open("a") as output:
            output.write(f"{datetime.now(UTC).isoformat()} {message}\n")

model_config class-attribute instance-attribute 

model_config = ConfigDict(hide_input_in_errors=True)

label pydantic-field 

label: str

directory pydantic-field 

directory: Path

user pydantic-field 

user: None = None

group pydantic-field 

group: None = None

arguments pydantic-field 

arguments: list[str]

environment pydantic-field 

environment: dict[str, str]

stdout pydantic-field 

stdout: Path

stderr pydantic-field 

stderr: Path

session_type pydantic-field 

session_type: Literal['Aqua']

program pydantic-field 

program: None = None

keep_alive pydantic-field 

keep_alive: Literal[True]

abandon_process_group pydantic-field 

abandon_process_group: Literal[False] = False

checkout property 

checkout: Path

root property 

root: Path

domain property 

domain: str

target property 

target: str

The desktop launchd job identified by the plist's Label.

path staticmethod 

path() -> Path
Source code in cli/octomate_cli/service.py 
97
98
99
@staticmethod
def path() -> Path:
    return Path.home() / "Library/LaunchAgents/io.octomate.server.plist"

load classmethod 

load() -> PlistService
Source code in cli/octomate_cli/service.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@classmethod
def load(cls) -> PlistService:
    if sys.platform != "darwin":
        raise typer.BadParameter("Service commands require macOS launchd.")
    # pwd is unavailable on Windows; client CLI imports must remain portable.
    import pwd

    if os.getuid() == 0:
        raise typer.BadParameter("Run as the desktop account, without sudo.")
    path = cls.path()
    try:
        if path.stat().st_uid != os.getuid():
            raise ValueError("The service definition must belong to this account.")
        with path.open("rb") as source:
            service = cls.model_validate(plistlib.load(source))
    except (OSError, ValueError) as error:
        raise typer.BadParameter(
            f"Invalid GUI service definition: {error}"
        ) from None
    account = pwd.getpwuid(os.getuid())
    if service.label != "io.octomate.server":
        raise typer.BadParameter("Unexpected service label.")
    if (
        len(service.arguments) != 3
        or service.arguments[1:] != ["service", "serve"]
        or not Path(service.arguments[0]).is_absolute()
        or Path(service.arguments[0]).parts[-3:] != (".venv", "bin", "octomate")
    ):
        raise typer.BadParameter(
            "The service must run <checkout>/.venv/bin/octomate service serve."
        )
    if not all(
        path.is_absolute()
        for path in (service.directory, service.stdout, service.stderr)
    ):
        raise typer.BadParameter(
            "Working directory and log paths must be absolute."
        )
    if not all(
        service.environment.get(key)
        for key in ("PATH", "OCTOMATE_HOME", "OCTOMATE_DB_URL")
    ):
        raise typer.BadParameter(
            "The service must set PATH, OCTOMATE_HOME and OCTOMATE_DB_URL."
        )
    if service.environment.get("HOME") != account.pw_dir or any(
        service.environment.get(key) != account.pw_name
        for key in ("USER", "LOGNAME")
    ):
        raise typer.BadParameter(
            "The service must use this desktop account's HOME, USER and LOGNAME."
        )
    return service

require_gui 

require_gui() -> None
Source code in cli/octomate_cli/service.py 
172
173
174
175
176
177
178
179
def require_gui(self) -> None:
    result = subprocess.run(
        ["/bin/launchctl", "print", self.domain], capture_output=True, text=True
    )
    if result.returncode:
        raise ValueError(
            f"Desktop session {self.domain} is unavailable. Log into the desktop first."
        )

loaded 

loaded() -> bool

Check whether the launchd job defined by this plist is loaded.

Source code in cli/octomate_cli/service.py 
181
182
183
184
185
186
187
188
189
def loaded(self) -> bool:
    """Check whether the launchd job defined by this plist is loaded."""
    result = subprocess.run(
        ["/bin/launchctl", "print", self.target], capture_output=True, text=True
    )
    if result.returncode == 113:
        return False
    result.check_returncode()
    return True

launchctl 

launchctl(action: str, *arguments: str) -> None

Apply an unprivileged action in the desktop launchd domain.

Source code in cli/octomate_cli/service.py 
191
192
193
def launchctl(self, action: str, *arguments: str) -> None:
    """Apply an unprivileged action in the desktop launchd domain."""
    subprocess.run(["/bin/launchctl", action, *arguments], check=True)

stop 

stop() -> None

Disable the job and wait for its process group to exit.

Source code in cli/octomate_cli/service.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def stop(self) -> None:
    """Disable the job and wait for its process group to exit."""
    self.launchctl("disable", self.target)
    if not self.loaded():
        return
    result = subprocess.run(
        ["/bin/launchctl", "print", self.target],
        capture_output=True,
        text=True,
        check=True,
    )
    match = re.search(r"^\s*pid = ([0-9]+)\s*$", result.stdout, re.MULTILINE)
    group: int | None = None
    if match is not None:
        try:
            group = os.getpgid(int(match[1]))
        except ProcessLookupError:
            pass
    self.launchctl("bootout", self.target)
    deadline = time.monotonic() + 30
    while group is not None:
        try:
            os.killpg(group, 0)
        except ProcessLookupError:
            break
        except PermissionError:
            # macOS reports EPERM while a group contains only unreaped zombies.
            # Keep waiting for ESRCH under the same shutdown deadline.
            pass
        if time.monotonic() >= deadline:
            raise TimeoutError(
                "The service process group did not stop within 30 seconds."
            )
        time.sleep(0.25)

maintenance 

maintenance(
    action: str, backup: DatabaseBackup | None = None
) -> str

Run maintenance with the plist's working directory and environment.

Source code in cli/octomate_cli/service.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def maintenance(self, action: str, backup: DatabaseBackup | None = None) -> str:
    """Run maintenance with the plist's working directory and environment."""
    result = subprocess.run(
        [
            str(self.checkout / ".venv/bin/python"),
            "-m",
            "octomate_cli.deployment",
            action,
        ],
        cwd=self.directory,
        env={
            **self.environment,
            "OCTOMATE_DEPLOYMENT_ROOT": str(self.root),
        },
        input=backup.model_dump_json() if backup is not None else None,
        stdout=subprocess.PIPE,
        text=True,
        check=True,
    )
    return result.stdout.strip()

report 

report(message: str) -> None

Print service status and log it beside the plist's configured checkout.

Source code in cli/octomate_cli/service.py 
251
252
253
254
255
256
257
def report(self, message: str) -> None:
    """Print service status and log it beside the plist's configured checkout."""
    console.print(message)
    logs = self.root / "logs"
    logs.mkdir(parents=True, exist_ok=True)
    with (logs / "server.log").open("a") as output:
        output.write(f"{datetime.now(UTC).isoformat()} {message}\n")

latest_server_release 

latest_server_release() -> Release

Find the highest stable server version across all release pages.

Source code in cli/octomate_cli/service.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def latest_server_release() -> Release:
    """Find the highest stable server version across all release pages."""
    latest: Release | None = None
    latest_version = (0, 0, 0)
    page = 1
    adapter = TypeAdapter(list[Release])
    while True:
        request = Request(
            f"{RELEASE_URL}?per_page=100&page={page}",
            headers={"Accept": "application/vnd.github+json"},
        )
        with urlopen(request, timeout=30) as response:
            releases = adapter.validate_json(response.read())
        for release in releases:
            match = re.fullmatch(
                r"octomate-v([0-9]+)\.([0-9]+)\.([0-9]+)", release.tag_name
            )
            if release.draft or release.prerelease or match is None:
                continue
            version = tuple(int(part) for part in match.groups())
            if latest is None or version > latest_version:
                latest, latest_version = release, version
        if len(releases) < 100:
            break
        page += 1
    if latest is None:
        raise ValueError("No stable octomate-vX.Y.Z server release is available.")
    return latest

manage_service 

manage_service(
    action: Literal["start", "stop", "restart", "upgrade"],
) -> None

Serialize service changes, preserving the installed release on ordinary starts.

Source code in cli/octomate_cli/service.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
def manage_service(action: Literal["start", "stop", "restart", "upgrade"]) -> None:
    """Serialize service changes, preserving the installed release on ordinary starts."""
    service = PlistService.load()
    if (
        action == "upgrade"
        and Path(sys.prefix).resolve() == (service.checkout / ".venv").resolve()
    ):
        raise typer.BadParameter(
            "Run service upgrade from the standalone CLI, not the service environment. "
            "Install it with: uv tool install octomate-cli"
        )
    # fcntl is unavailable on Windows; client CLI imports must remain portable.
    import fcntl

    upgrade = action == "upgrade"
    control = service.root / "control"
    control.mkdir(parents=True, exist_ok=True)
    with (control / "server.lock").open("a") as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise typer.BadParameter("Another server operation is running.") from None

        changed_service = False
        try:
            service.require_gui()
            if action == "stop":
                service.stop()
                service.report("Service stopped and disabled.")
                return
            service.maintenance("check" if upgrade else "ready")
            loaded = service.loaded()
            if loaded and action == "start":
                service.report(service.maintenance("verify"))
                service.report("The server is already loaded; no migration was run.")
                return
            if upgrade:
                dirty = subprocess.check_output(
                    ["git", "status", "--porcelain", "--untracked-files=no"],
                    cwd=service.checkout,
                    env=service.environment,
                    text=True,
                ).strip()
                if dirty:
                    raise ValueError("Upgrade requires no tracked local changes.")
                previous = subprocess.check_output(
                    ["git", "rev-parse", "HEAD"],
                    cwd=service.checkout,
                    env=service.environment,
                    text=True,
                ).strip()
                release = latest_server_release()
                subprocess.run(
                    ["git", "fetch", "origin", f"refs/tags/{release.tag_name}"],
                    cwd=service.checkout,
                    env=service.environment,
                    check=True,
                )
                revision = subprocess.check_output(
                    ["git", "rev-parse", "FETCH_HEAD^{commit}"],
                    cwd=service.checkout,
                    env=service.environment,
                    text=True,
                ).strip()
                if previous == revision:
                    service.report(f"Already at {release.tag_name}; no update needed.")
                    return
                ancestor = subprocess.check_output(
                    ["git", "merge-base", previous, revision],
                    cwd=service.checkout,
                    env=service.environment,
                    text=True,
                ).strip()
                if ancestor != previous:
                    raise ValueError(
                        "The latest release does not contain the installed commit; "
                        "refusing a downgrade or divergent history."
                    )
                service.report(
                    f"Upgrading from {previous} to {release.tag_name} ({revision})."
                )

            changed_service = True
            service.stop()
            if upgrade:
                backup = DatabaseBackup.model_validate_json(
                    service.maintenance("backup")
                )
                service.report(f"Database: {backup.database}; backup: {backup.backup}.")
                subprocess.run(
                    ["git", "checkout", "--detach", revision],
                    cwd=service.checkout,
                    env=service.environment,
                    check=True,
                )
                subprocess.run(
                    [
                        "uv",
                        "sync",
                        "--locked",
                        "--no-dev",
                        "--project",
                        str(service.checkout),
                    ],
                    cwd=service.checkout,
                    env={
                        **service.environment,
                        "UV_PROJECT_ENVIRONMENT": str(service.checkout / ".venv"),
                    },
                    check=True,
                )
                service.report(service.maintenance("migrate", backup))
            else:
                service.maintenance("stopped")
            service.launchctl("enable", service.target)
            service.launchctl("bootstrap", service.domain, str(service.path()))
            service.report(service.maintenance("verify"))
            service.report(
                "Server started. Check service logs for agent/channel startup."
            )
        except (
            OSError,
            ValueError,
            subprocess.CalledProcessError,
            KeyboardInterrupt,
        ) as error:
            if changed_service:
                service.launchctl("disable", service.target)
                if service.loaded():
                    service.launchctl("bootout", service.target)
                service.report(
                    "Operation failed; the server remains disabled until recovery."
                )
            service.report(str(error))
            raise typer.Exit(
                130 if isinstance(error, KeyboardInterrupt) else 1
            ) from error

upgrade 

upgrade() -> None

Update the GUI service's release, dependencies and database, then restart.

Source code in cli/octomate_cli/service.py 
399
400
401
402
@service_typer.command()
def upgrade() -> None:
    """Update the GUI service's release, dependencies and database, then restart."""
    manage_service("upgrade")

start 

start() -> None

Ask macOS launchd to start and manage the installed GUI service.

Does not update code or migrate data. Use service serve to run the server directly in the foreground for Docker or development.

Source code in cli/octomate_cli/service.py 
405
406
407
408
409
410
411
412
@service_typer.command()
def start() -> None:
    """Ask macOS launchd to start and manage the installed GUI service.

    Does not update code or migrate data. Use `service serve` to run the server
    directly in the foreground for Docker or development.
    """
    manage_service("start")

stop 

stop() -> None

Stop the GUI service and keep it disabled across desktop logins.

Source code in cli/octomate_cli/service.py 
415
416
417
418
@service_typer.command()
def stop() -> None:
    """Stop the GUI service and keep it disabled across desktop logins."""
    manage_service("stop")

restart 

restart() -> None

Restart the installed GUI service without updating code or migrating data.

Source code in cli/octomate_cli/service.py 
421
422
423
424
@service_typer.command()
def restart() -> None:
    """Restart the installed GUI service without updating code or migrating data."""
    manage_service("restart")

verify 

verify() -> None

Check schema and protected HTTP routes; no model or connector requests.

Source code in cli/octomate_cli/service.py 
427
428
429
430
431
432
433
434
435
436
437
438
439
440
@service_typer.command()
def verify() -> None:
    """Check schema and protected HTTP routes; no model or connector requests."""
    service = PlistService.load()
    try:
        service.require_gui()
        if not service.loaded():
            raise ValueError("The service is not loaded. Run octomate service start.")
        service.maintenance("ready")
        console.print(service.maintenance("verify"))
        console.print("Not checked: Claude login, connectors and plugins.")
    except (OSError, ValueError, subprocess.CalledProcessError) as error:
        console.print(str(error), style="red")
        raise typer.Exit(1) from error

status 

status() -> None

Show the installed GUI job and its configured paths without modifying it.

Source code in cli/octomate_cli/service.py 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
@service_typer.command()
def status() -> None:
    """Show the installed GUI job and its configured paths without modifying it."""
    service = PlistService.load()
    try:
        service.require_gui()
        result = subprocess.run(
            ["/bin/launchctl", "print", service.target], capture_output=True, text=True
        )
        if result.returncode != 113:
            result.check_returncode()
        pid = re.search(r"^\s*pid = ([0-9]+)\s*$", result.stdout, re.MULTILINE)
        disabled = subprocess.check_output(
            ["/bin/launchctl", "print-disabled", service.domain], text=True
        )
        explicitly_disabled = (
            re.search(rf'"{re.escape(service.label)}"\s*=>\s*true', disabled)
            is not None
        )
        revision = subprocess.check_output(
            ["git", "rev-parse", "HEAD"],
            cwd=service.checkout,
            env=service.environment,
            text=True,
        ).strip()
        table = Table("Service", "Value", box=None)
        for name, value in (
            ("Target", service.target),
            ("Enabled", "No" if explicitly_disabled else "Yes"),
            ("Loaded", "Yes" if result.returncode == 0 else "No"),
            ("PID", pid[1] if pid else "Not running"),
            ("Revision", revision),
            ("Directory", str(service.directory)),
            ("Checkout", str(service.checkout)),
            ("Config", service.environment["OCTOMATE_HOME"]),
            ("Stdout", str(service.stdout)),
            ("Stderr", str(service.stderr)),
        ):
            table.add_row(name, value)
        console.print(table)
        console.print("Requires a desktop login; logout stops the service.")
    except (OSError, ValueError, subprocess.CalledProcessError) as error:
        console.print(str(error), style="red")
        raise typer.Exit(1) from error

logs 

logs(
    follow: Annotated[
        bool, Option(help="Follow new service log output.")
    ] = False,
) -> None

Read the configured stdout and stderr logs.

Source code in cli/octomate_cli/service.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
@service_typer.command()
def logs(
    follow: Annotated[
        bool, typer.Option(help="Follow new service log output.")
    ] = False,
) -> None:
    """Read the configured stdout and stderr logs."""
    service = PlistService.load()
    paths = list(dict.fromkeys((str(service.stdout), str(service.stderr))))
    try:
        subprocess.run(
            ["/usr/bin/tail", "-n", "100", *(["-f"] if follow else []), "--", *paths],
            check=True,
        )
    except KeyboardInterrupt:
        raise typer.Exit(130) from None
    except (OSError, subprocess.CalledProcessError) as error:
        console.print(str(error), style="red")
        raise typer.Exit(1) from error

invite 

invite(
    url: Annotated[
        str | None,
        Option(
            help="Print a registration link for this URL instead of a code."
        ),
    ] = None,
) -> None

Issue an anonymous, single-use registration invitation.

Source code in cli/octomate_cli/service.py 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
@service_typer.command()
def invite(
    url: Annotated[
        str | None,
        typer.Option(help="Print a registration link for this URL instead of a code."),
    ] = None,
) -> None:
    """Issue an anonymous, single-use registration invitation."""
    if find_spec("octomate") is None:
        raise typer.BadParameter("Invitations require the Octomate server package")

    # Server imports stay here so the standalone client CLI remains usable.
    from octomate.config import OctomateConfig
    from octomate.managers.auth import AuthManager
    from octomate.schemas.base import sqlalchemy_materia

    base_url = TypeAdapter(HttpUrl).validate_python(url) if url is not None else None
    config = OctomateConfig()
    if config.auth is None:
        raise typer.BadParameter("Configure auth.yaml before issuing invitations")
    manager = AuthManager(config.auth)

    async def issue() -> str:
        with sqlalchemy_materia():
            token = await manager.invite()
        return token.get_secret_value()

    token = asyncio.run(issue())
    if base_url is None:
        typer.echo(token)
    else:
        fragment = urlencode({"invitation": token})
        typer.echo(f"{str(base_url).rstrip('/')}/#{fragment}")

serve 

serve(
    host: Annotated[
        str | None,
        Option(
            help="Address to bind. Defaults to octomate.host (127.0.0.1)."
        ),
    ] = None,
    port: Annotated[
        int | None,
        Option(
            help="Port to bind. Defaults to octomate.port (8000)."
        ),
    ] = None,
    reload: Annotated[
        bool,
        Option(
            help="Restart when files under octomate/ change."
        ),
    ] = False,
    tmux: Annotated[
        bool,
        Option(
            help="Serve inside a detached tmux session and attach to it."
        ),
    ] = False,
    session: Annotated[
        str, Option(help="tmux session name.")
    ] = "octomate",
) -> None

Run the server directly in the foreground for Docker or development.

Use service start to have macOS launchd start and manage the installed GUI service. Use --tmux to create or attach to a persistent terminal session.

Source code in cli/octomate_cli/service.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
@service_typer.command()
def serve(
    host: Annotated[
        str | None,
        typer.Option(help="Address to bind. Defaults to octomate.host (127.0.0.1)."),
    ] = None,
    port: Annotated[
        int | None,
        typer.Option(help="Port to bind. Defaults to octomate.port (8000)."),
    ] = None,
    reload: Annotated[
        bool,
        typer.Option(help="Restart when files under octomate/ change."),
    ] = False,
    tmux: Annotated[
        bool,
        typer.Option(help="Serve inside a detached tmux session and attach to it."),
    ] = False,
    session: Annotated[
        str,
        typer.Option(help="tmux session name."),
    ] = "octomate",
) -> None:
    """Run the server directly in the foreground for Docker or development.

    Use `service start` to have macOS launchd start and manage the installed GUI
    service. Use --tmux to create or attach to a persistent terminal session.
    """
    if find_spec("octomate") is None:
        raise typer.BadParameter("Serving requires the Octomate server package")

    # Server imports stay here so the standalone client CLI remains usable.
    import uvicorn

    from octomate.config import OctomateConfig
    from octomate.server import run

    if tmux:
        if shutil.which("tmux") is None:
            typer.secho("tmux is not installed.", fg=typer.colors.RED, err=True)
            raise typer.Exit(1)
        serving = subprocess.run(
            ["tmux", "has-session", "-t", session], capture_output=True
        )
        if serving.returncode != 0:
            command = [
                # Keep the active virtualenv when tmux starts outside this shell.
                sys.executable,
                "-m",
                "octomate_cli.main",
                "service",
                "serve",
            ]
            if host is not None:
                command += ["--host", host]
            if port is not None:
                command += ["--port", str(port)]
            if reload:
                command.append("--reload")
            subprocess.run(
                [
                    "tmux",
                    "new-session",
                    "-d",
                    "-s",
                    session,
                    "-c",
                    os.getcwd(),
                    *command,
                ],
                check=True,
            )
            typer.secho(f"Serving in tmux session {session!r}.", fg=typer.colors.GREEN)
        # tmux refuses a nested attach, and switching is what the request means
        # when it comes from inside a pane.
        attach = "switch-client" if os.environ.get("TMUX") else "attach-session"
        subprocess.run(["tmux", attach, "-t", session], check=True)
        return

    if port is not None:
        # The factory reads OctomateConfig() itself; export the override so the
        # config the app is built from — the Octomate MCP URL driven runtimes are
        # wired with included — agrees with the bind.
        os.environ["OCTOMATE__PORT"] = str(port)
    config = OctomateConfig()
    run(
        uvicorn.Config(
            "octomate.app:create_app",
            factory=True,
            lifespan="on",
            host=str(config.host) if host is None else host,
            port=config.port if port is None else port,
            reload=reload,
            # Watch application code without watching the deployment's mutable data.
            reload_dirs=[str(Path(inspect.getfile(OctomateConfig)).parent.parent)]
            if reload
            else None,
            log_level=config.logging.level.lower(),
        )
    )

octomate_cli.init

Interactive service setup: installation, network, agents, channels and preparation.

init 

init(
    prepare: Annotated[
        bool,
        Option(
            "--prepare",
            help="Prepare files only; do not initialize data or start services.",
        ),
    ] = False,
    root: Annotated[
        Path | None,
        Option(
            help="Service installation directory; prompts when omitted."
        ),
    ] = None,
    source: Annotated[
        Path | None,
        Option(
            exists=True,
            file_okay=False,
            help="Copy a local Git working tree, including uncommitted files, for development.",
        ),
    ] = None,
    port: Annotated[
        int | None,
        Option(
            min=1,
            max=65535,
            help="Loopback port for the service.",
        ),
    ] = None,
    agent: Annotated[
        list[str] | None,
        Option(
            "--agent",
            help="Agent template: claude, codex or deepseek (DSH; experimental). Repeat for multiple agents.",
        ),
    ] = None,
    channel: Annotated[
        list[str] | None,
        Option(
            "--channel",
            help="Channel template: slack, lark, discord, trunkline or none. Repeat for multiple channels; fill credentials in config later.",
        ),
    ] = None,
    target: Annotated[
        DeploymentTarget | None,
        Option(
            help="Deployment target; defaults to launchd on macOS and systemd on Linux."
        ),
    ] = None,
    yes: Annotated[
        bool,
        Option(
            "--yes",
            help="Confirm preparation with explicit root, port, agent and channel choices; skip optional MCP setup.",
        ),
    ] = False,
) -> None

Prepare a macOS, Linux or Docker installation through the same wizard.

Currently requires --prepare. Creates a private installation and a reviewable service definition inside it. Database initialization, account creation, service activation and live agent verification remain explicit operator steps.

Source code in cli/octomate_cli/init.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def init(
    prepare: Annotated[
        bool,
        typer.Option(
            "--prepare",
            help="Prepare files only; do not initialize data or start services.",
        ),
    ] = False,
    root: Annotated[
        Path | None,
        typer.Option(help="Service installation directory; prompts when omitted."),
    ] = None,
    source: Annotated[
        Path | None,
        typer.Option(
            exists=True,
            file_okay=False,
            help="Copy a local Git working tree, including uncommitted files, for development.",
        ),
    ] = None,
    port: Annotated[
        int | None,
        typer.Option(min=1, max=65535, help="Loopback port for the service."),
    ] = None,
    agent: Annotated[
        list[str] | None,
        typer.Option(
            "--agent",
            help="Agent template: claude, codex or deepseek (DSH; experimental). Repeat for multiple agents.",
        ),
    ] = None,
    channel: Annotated[
        list[str] | None,
        typer.Option(
            "--channel",
            help="Channel template: slack, lark, discord, trunkline or none. Repeat for multiple channels; fill credentials in config later.",
        ),
    ] = None,
    target: Annotated[
        DeploymentTarget | None,
        typer.Option(
            help="Deployment target; defaults to launchd on macOS and systemd on Linux."
        ),
    ] = None,
    yes: Annotated[
        bool,
        typer.Option(
            "--yes",
            help="Confirm preparation with explicit root, port, agent and channel choices; skip optional MCP setup.",
        ),
    ] = False,
) -> None:
    """Prepare a macOS, Linux or Docker installation through the same wizard.

    Currently requires --prepare. Creates a private installation and a reviewable
    service definition inside it. Database initialization, account creation,
    service activation and live agent verification remain explicit operator steps.
    """
    if not prepare:
        raise typer.BadParameter(
            "Use --prepare; service activation is not implemented yet."
        )
    if sys.platform not in {"darwin", "linux"}:
        raise typer.BadParameter(
            "Deployment preparation supports macOS and Linux hosts. Windows support is deferred."
        )
    if target is None:
        target = (
            DeploymentTarget.launchd
            if sys.platform == "darwin"
            else DeploymentTarget.systemd
        )
    if (target == DeploymentTarget.launchd and sys.platform != "darwin") or (
        target == DeploymentTarget.systemd and sys.platform != "linux"
    ):
        raise typer.BadParameter(
            f"The {target} target cannot run on this host; choose docker or manual instead."
        )
    # pwd is unavailable on Windows; client CLI imports must remain portable.
    import pwd

    if os.getuid() == 0:
        raise typer.BadParameter(
            "Run as the account that will own the installation, without sudo."
        )
    account = pwd.getpwuid(os.getuid())
    console.print(
        Panel(
            "Generate config templates for an isolated service installation.\n"
            "Preparation does not start services or write databases.",
            title="Octomate init · prepare",
            border_style=brand_color,
        )
    )
    root = installation_step(root, account.pw_dir, yes)
    service = Installation(
        root, build_environment(root, account.pw_dir, account.pw_name, target), target
    )
    if service.draft.exists():
        if (
            source is not None
            or port is not None
            or agent is not None
            or channel is not None
        ):
            raise typer.BadParameter(
                "Prepared installations retain their settings. Omit --source, --port, --agent and --channel to validate again."
            )
        validate_existing(service, account.pw_dir)
        return
    if root.exists() and (not root.is_dir() or any(root.iterdir())):
        raise typer.BadParameter(
            "Choose an empty installation directory. Existing files will not be overwritten."
        )
    git = shutil.which("git")
    installer = shutil.which("docker" if target == DeploymentTarget.docker else "uv")
    if git is None or installer is None:
        raise typer.BadParameter(
            "Install Git and Docker (with Compose) for Docker, or Git and uv for a native service."
        )
    source, revision = select_source(root, source, git)
    if yes and (port is None or agent is None or channel is None):
        raise typer.BadParameter(
            "--yes requires --port, --agent and --channel (use none for no channels)."
        )
    port = network_step(port)
    tentacles = tentacles_step(agent, channel, interactive=not yes, console=console)
    if target == DeploymentTarget.docker and "deepseek" in tentacles.agents:
        raise typer.BadParameter(
            "The Docker image includes Claude and Codex. DSH needs a custom image; add it after preparation."
        )
    review_step(
        service,
        source,
        revision,
        port,
        tentacles.agents,
        tentacles.channels,
        tentacles.mcps,
        yes,
    )
    prepare_step(
        service,
        source,
        revision,
        git,
        installer,
        port,
        tentacles.agents,
        tentacles.channels,
        tentacles.mcps,
    )

installation_step 

installation_step(
    root: Path | None, home: str, yes: bool
) -> Path
Source code in cli/octomate_cli/init.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def installation_step(root: Path | None, home: str, yes: bool) -> Path:
    console.print("1/6 · Installation", style=f"bold {brand_color}")
    if root is None:
        if yes:
            raise typer.BadParameter("--yes requires --root.")
        root = Path(
            Prompt.ask(
                "Installation directory",
                default=str(
                    Path(home)
                    / (
                        "Library/Application Support/Octomate"
                        if sys.platform == "darwin"
                        else ".local/share/octomate-server"
                    )
                ),
                console=console,
            )
        )
    root = root.expanduser()
    if root.is_symlink():
        raise typer.BadParameter("The installation directory must not be a symlink.")
    root = root.resolve()
    return root

validate_existing 

validate_existing(
    installation: Installation, home: str
) -> None
Source code in cli/octomate_cli/init.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def validate_existing(installation: Installation, home: str) -> None:
    # Imported here because service owns this model and registers the init command.
    from octomate_cli.service import PlistService

    root = installation.directory
    draft = installation.draft
    try:
        if draft.is_symlink():
            raise ValueError("The prepared definition must not be a symlink.")
        if installation.target != DeploymentTarget.launchd:
            subprocess.run(
                installation.maintenance_command("check"),
                cwd=root,
                env=installation.environment,
                check=True,
            )
            console.print(
                "Configuration structure is valid; files and credentials are unchanged.",
                style="green",
            )
            return
        service = PlistService.model_validate(plistlib.loads(draft.read_bytes()))
        if (
            service.directory != root
            or service.arguments
            != [str(root / "app/.venv/bin/octomate"), "service", "serve"]
            or service.environment.get("OCTOMATE_HOME") != str(root / "config")
            or service.environment.get("OCTOMATE_DB_URL")
            != f"sqlite+aiosqlite:///{root / 'octomate.db'}"
            or service.environment.get("HOME") != home
        ):
            raise ValueError(
                "The prepared definition does not match this installation and account."
            )
        service.maintenance("check")
    except (OSError, ValueError, subprocess.CalledProcessError) as error:
        raise typer.BadParameter(
            f"Prepared configuration validation failed: {error}"
        ) from None
    console.print(
        "Configuration structure is valid; files and credentials are unchanged.",
        style="green",
    )
    console.print(
        "Follow CONFIGURATION.md to complete configuration. Service remains uninstalled; live authentication, connectors and plugins have not been verified."
    )
    return

select_source 

select_source(
    root: Path, source: Path | None, git: str
) -> tuple[Path | None, str]
Source code in cli/octomate_cli/init.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def select_source(root: Path, source: Path | None, git: str) -> tuple[Path | None, str]:
    # Deferred to avoid a cycle with the service command registration.
    from octomate_cli.service import latest_server_release

    if source is not None:
        source = source.resolve()
        if root.is_relative_to(source) or source.is_relative_to(root):
            raise typer.BadParameter(
                "The installation and source checkout must be separate directories."
            )
        try:
            source_root = Path(
                subprocess.check_output(
                    [git, "-C", str(source), "rev-parse", "--show-toplevel"], text=True
                ).strip()
            ).resolve()
        except subprocess.CalledProcessError:
            raise typer.BadParameter("--source must point to a Git checkout.") from None
        if source_root != source:
            raise typer.BadParameter(
                "--source must point to the root of the Git checkout."
            )
        revision = "Working tree snapshot (no release tag)"
        console.print(
            f"Source: {source} (selected by --source). No release tag is selected."
        )
    else:
        with console.status("Finding the latest stable service release…"):
            try:
                revision = latest_server_release().tag_name
            except (OSError, ValueError) as error:
                raise typer.BadParameter(f"Release lookup failed: {error}") from None
        console.print(
            f"Selected {revision}: highest stable octomate-vX.Y.Z GitHub release."
        )
    return source, revision

network_step 

network_step(port: int | None) -> int
Source code in cli/octomate_cli/init.py 
296
297
298
299
300
301
302
303
304
def network_step(port: int | None) -> int:
    console.print("2/6 · Network", style=f"bold {brand_color}")
    while port is None:
        value = IntPrompt.ask("Loopback port", default=8000, console=console)
        if 1 <= value <= 65535:
            port = value
        else:
            console.print("Choose a port from 1 to 65535.", style="yellow")
    return port

build_environment 

build_environment(
    root: Path,
    home: str,
    username: str,
    target: DeploymentTarget,
) -> dict[str, str]
Source code in cli/octomate_cli/init.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def build_environment(
    root: Path, home: str, username: str, target: DeploymentTarget
) -> dict[str, str]:
    environment = {
        "HOME": home,
        "USER": username,
        "LOGNAME": username,
        "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
        "OCTOMATE_HOME": str(root / "config"),
        "OCTOMATE_DB_URL": f"sqlite+aiosqlite:///{root / 'octomate.db'}",
    }
    forwarded = (
        (
            "CLAUDE_CONFIG_DIR",
            "CODEX_HOME",
            "HTTPS_PROXY",
            "HTTP_PROXY",
            "ALL_PROXY",
            "NO_PROXY",
            "SSL_CERT_FILE",
            "SSL_CERT_DIR",
            "REQUESTS_CA_BUNDLE",
        )
        if target != DeploymentTarget.docker
        else (
            "DOCKER_HOST",
            "DOCKER_CONTEXT",
            "DOCKER_CONFIG",
            "DOCKER_TLS_VERIFY",
            "DOCKER_CERT_PATH",
        )
    )
    for key in forwarded:
        if key in os.environ:
            environment[key] = os.environ[key]
    for key in ("CLAUDE_CONFIG_DIR", "CODEX_HOME"):
        if key in environment:
            environment[key] = str(Path(environment[key]).expanduser().resolve())
    return environment

review_step 

review_step(
    service: Installation,
    source: Path | None,
    revision: str,
    port: int,
    agent: list[str],
    channels: list[str],
    mcps: list[McpPreset],
    yes: bool,
) -> None
Source code in cli/octomate_cli/init.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
def review_step(
    service: Installation,
    source: Path | None,
    revision: str,
    port: int,
    agent: list[str],
    channels: list[str],
    mcps: list[McpPreset],
    yes: bool,
) -> None:
    root = service.directory
    environment = service.environment
    draft = service.draft
    console.print("5/6 · Review", style=f"bold {brand_color}")
    table = Table(box=None, header_style="bold")
    table.add_column("Setting", no_wrap=True, style="dim")
    table.add_column("Value", overflow="fold")
    for key, value in (
        ("Installation", str(root)),
        ("Version", revision),
        ("Source", str(source) if source else "github.com/kalynnka/octomate"),
        ("Bind", f"127.0.0.1:{port}"),
        (
            "Agents Tentacles",
            ", ".join(AGENTS[name].label for name in agent),
        ),
        (
            "Channels Tentacles",
            ", ".join(CHANNELS[name].label for name in channels) or "None",
        ),
        (
            "MCP Tentacles",
            ", ".join(
                f"{MCPS[mcp.provider].label} ({mcp.name}, {'read-only' if mcp.read_only else 'read/write'})"
                for mcp in mcps
            )
            or "None",
        ),
        (
            "Claude home",
            (
                str(root / "agent-home/.claude")
                if service.target == DeploymentTarget.docker
                else environment.get(
                    "CLAUDE_CONFIG_DIR", str(Path(environment["HOME"]) / ".claude")
                )
            )
            if "claude" in agent
            else "Not enabled",
        ),
        (
            "Codex home",
            (
                str(root / "agent-home/.codex")
                if service.target == DeploymentTarget.docker
                else environment.get(
                    "CODEX_HOME", str(Path(environment["HOME"]) / ".codex")
                )
            )
            if "codex" in agent
            else "Not enabled",
        ),
        ("Launch context", f"{environment['USER']} · {service.target}"),
        ("Configuration", str(service.state / "config")),
        ("Auth secrets", str(service.state / ".env")),
        ("Configuration checklist", str(service.state / "CONFIGURATION.md")),
        ("Database (not created)", str(service.state / "octomate.db")),
        ("Prepared definition", str(draft)),
    ):
        table.add_row(key, value)
    console.print(table)
    if service.target == DeploymentTarget.launchd:
        console.print(
            "The GUI service will require a desktop login; logging out stops it."
        )
    elif service.target == DeploymentTarget.docker:
        console.print(
            "Agent credentials belong in agent-home/ or explicit container environment variables. Host logins are not copied."
        )
    elif service.target == DeploymentTarget.systemd:
        console.print(
            "A systemd user unit will be drafted. Boot without a user login requires lingering."
        )
    if not yes and not Confirm.ask(
        "Prepare these files?", default=True, console=console
    ):
        raise typer.Abort()

prepare_step 

prepare_step(
    service: Installation,
    source: Path | None,
    revision: str,
    git: str,
    installer: str,
    port: int,
    agent: list[str],
    channels: list[str],
    mcps: list[McpPreset],
) -> None
Source code in cli/octomate_cli/init.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
def prepare_step(
    service: Installation,
    source: Path | None,
    revision: str,
    git: str,
    installer: str,
    port: int,
    agent: list[str],
    channels: list[str],
    mcps: list[McpPreset],
) -> None:
    root = service.directory
    environment = service.environment
    root.mkdir(mode=0o700, parents=True, exist_ok=True)
    root.chmod(0o700)
    for directory in ("control", "logs", "backups"):
        (root / directory).mkdir(mode=0o700)
    log_path = root / "logs/prepare.log"
    console.print("6/6 · Prepare and validate", style=f"bold {brand_color}")
    console.print(f"Build log: {log_path}", style="dim")
    checkout = root / "app"
    build_env = {**environment, "UV_PROJECT_ENVIRONMENT": str(checkout / ".venv")}
    if "UV_CACHE_DIR" in os.environ:
        build_env["UV_CACHE_DIR"] = os.environ["UV_CACHE_DIR"]
    try:
        with log_path.open("x") as log:
            log_path.chmod(0o600)
            with console.status(
                "Installing the selected service and locked dependencies…"
            ):
                subprocess.run(
                    [git, "clone", "--no-hardlinks", str(source), str(checkout)]
                    if source
                    else [
                        git,
                        "clone",
                        "--depth",
                        "1",
                        "--branch",
                        revision,
                        "https://github.com/kalynnka/octomate.git",
                        str(checkout),
                    ],
                    env=build_env,
                    stdout=log,
                    stderr=subprocess.STDOUT,
                    check=True,
                )
                if source is not None:
                    copy_working_tree(source, checkout, git)
                if service.target == DeploymentTarget.docker:
                    service.state.mkdir(mode=0o700)
                    (root / "agent-home").mkdir(mode=0o700)
                    service.write_definition(port)
                    install_command = [installer, "compose", "build"]
                else:
                    install_command = [
                        installer,
                        "sync",
                        "--locked",
                        "--no-default-groups",
                        "--project",
                        str(checkout),
                    ]
                subprocess.run(
                    install_command,
                    cwd=root,
                    env=build_env,
                    stdout=log,
                    stderr=subprocess.STDOUT,
                    check=True,
                )
            with console.status("Generating private configuration and validating it…"):
                subprocess.run(
                    [
                        *service.maintenance_command("prepare"),
                        "--port",
                        str(port),
                        *(argument for name in agent for argument in ("--agent", name)),
                        *(
                            argument
                            for name in channels
                            for argument in ("--channel", name)
                        ),
                        *(["--mcp-presets"] if mcps else []),
                        *(
                            ["--target", service.target.value]
                            if service.target != DeploymentTarget.launchd
                            else []
                        ),
                    ],
                    input=TypeAdapter(list[McpPreset]).dump_json(mcps)
                    if mcps
                    else None,
                    cwd=root,
                    env=environment,
                    stdout=log,
                    stderr=subprocess.STDOUT,
                    check=True,
                )
                subprocess.run(
                    service.maintenance_command("check"),
                    cwd=root,
                    env=environment,
                    check=True,
                )
        if service.target != DeploymentTarget.docker:
            service.write_definition(port)
    except (OSError, ValueError, subprocess.CalledProcessError) as error:
        console.print(f"Preparation failed: {error}", style="red")
        console.print(
            f"Inspect {log_path}. No service was activated; this directory may contain incomplete preparation files."
        )
        raise typer.Exit(1) from error
    console.print(
        Panel(
            Text.assemble(
                ("Installation\n", "bold"),
                str(root),
                ("\n\nNext steps\n", f"bold {brand_color}"),
                "Paths below are relative to the installation.\n",
                "1. Complete ",
                (str((service.state / "CONFIGURATION.md").relative_to(root)), "bold"),
                " to finish your configuration.\n",
                "2. Review ",
                (str(service.draft.relative_to(root)), "bold"),
                ".\n",
                ("\nStill pending\n", "bold"),
                "Database and account creation; service installation and startup.\n",
                "Live agent, connector and plugin verification.",
            ),
            title="Preparation complete",
            border_style="green",
        )
    )

copy_working_tree 

copy_working_tree(
    source: Path, checkout: Path, git: str
) -> None

Snapshot Git-visible working files without sharing ignored local config or data.

Source code in cli/octomate_cli/init.py 
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
def copy_working_tree(source: Path, checkout: Path, git: str) -> None:
    """Snapshot Git-visible working files without sharing ignored local config or data."""
    files = (
        subprocess.check_output(
            [
                git,
                "-C",
                str(source),
                "ls-files",
                "--cached",
                "--others",
                "--exclude-standard",
                "-z",
            ]
        )
        .decode()
        .split("\0")
    )
    deleted = (
        subprocess.check_output(
            [
                git,
                "-C",
                str(source),
                "diff",
                "--name-only",
                "--diff-filter=D",
                "HEAD",
                "-z",
            ]
        )
        .decode()
        .split("\0")
    )
    for name in filter(None, deleted):
        destination = checkout / name
        if not destination.parent.resolve().is_relative_to(checkout):
            raise ValueError(f"Destination must stay inside the checkout: {name}")
        destination.unlink(missing_ok=True)
    for name in filter(None, files):
        origin = source / name
        if origin.is_symlink() or not origin.resolve().is_relative_to(source):
            raise ValueError(f"Source file must stay inside the checkout: {name}")
        if not origin.exists():
            continue
        destination = checkout / name
        if destination.is_symlink() or not destination.resolve().is_relative_to(
            checkout
        ):
            raise ValueError(f"Destination must stay inside the checkout: {name}")
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(origin, destination)

octomate_cli.users

The octomate user commands: create an account from an invitation, reset a password.

user_typer module-attribute 

user_typer = typer.Typer(
    help="Manage local user accounts.", no_args_is_help=True
)

password_error 

password_error(
    error: ValidationError,
) -> typer.BadParameter
Source code in cli/octomate_cli/users.py 
14
15
16
17
18
19
20
21
22
23
def password_error(error: ValidationError) -> typer.BadParameter:
    issue = error.errors(include_input=False, include_url=False)[0]
    context = issue.get("ctx", {})
    if issue["type"] == "too_short":
        message = f"Use at least {context['min_length']} characters."
    elif issue["type"] == "too_long":
        message = f"Use at most {context['max_length']} characters."
    else:
        message = str(context.get("error", issue["msg"]))
    return typer.BadParameter(message, param_hint="--password")

create 

create(
    username: Annotated[
        str, Option(help="Username for the new account.")
    ],
    password: Annotated[
        str, Option(help="Password for the new account.")
    ],
    invitationcode: Annotated[
        str,
        Option(
            help="An unused registration invitation code."
        ),
    ],
    name: Annotated[
        str | None,
        Option(
            help="Display name; defaults to the username."
        ),
    ] = None,
) -> None

Register a user with an invitation using the server's config and database.

Source code in cli/octomate_cli/users.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@user_typer.command("create")
def create(
    username: Annotated[str, typer.Option(help="Username for the new account.")],
    password: Annotated[str, typer.Option(help="Password for the new account.")],
    invitationcode: Annotated[
        str, typer.Option(help="An unused registration invitation code.")
    ],
    name: Annotated[
        str | None, typer.Option(help="Display name; defaults to the username.")
    ] = None,
) -> None:
    """Register a user with an invitation using the server's config and database."""
    if find_spec("octomate") is None:
        raise typer.BadParameter("Registration requires the Octomate server package")

    # Server imports stay here so the standalone client CLI remains usable.
    from octomate.config import OctomateConfig
    from octomate.managers.auth import AuthManager
    from octomate.schemas.base import sqlalchemy_materia

    config = OctomateConfig()
    if config.auth is None:
        raise typer.BadParameter("Configure auth.yaml before registering users")
    manager = AuthManager(config.auth)

    async def register() -> None:
        with sqlalchemy_materia():
            await manager.register(
                username,
                SecretStr(password),
                SecretStr(invitationcode),
                name=username if name is None else name,
            )

    try:
        asyncio.run(register())
    except ValidationError as error:
        raise password_error(error) from error
    except ValueError as error:
        raise typer.BadParameter(str(error)) from error
    typer.echo(f"Registered {username}. Sign in through Trunkline.")

reset_password 

reset_password(
    username: Annotated[
        str,
        Option(help="Username of the existing account."),
    ],
    password: Annotated[
        str,
        Option(
            help="New password; prompts securely when omitted.",
            prompt=True,
            hide_input=True,
            confirmation_prompt=True,
        ),
    ],
) -> None

Reset an account's password using local administrator access to the database.

Source code in cli/octomate_cli/users.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@user_typer.command("reset-password")
def reset_password(
    username: Annotated[str, typer.Option(help="Username of the existing account.")],
    password: Annotated[
        str,
        typer.Option(
            help="New password; prompts securely when omitted.",
            prompt=True,
            hide_input=True,
            confirmation_prompt=True,
        ),
    ],
) -> None:
    """Reset an account's password using local administrator access to the database."""
    if find_spec("octomate") is None:
        raise typer.BadParameter("Password reset requires the Octomate server package")

    # Server imports stay here so the standalone client CLI remains usable.
    from octomate.config import OctomateConfig
    from octomate.managers.auth import AuthManager
    from octomate.schemas.base import sqlalchemy_materia

    config = OctomateConfig()
    if config.auth is None:
        raise typer.BadParameter("Configure auth.yaml before resetting passwords")
    manager = AuthManager(config.auth)

    async def reset() -> None:
        with sqlalchemy_materia():
            await manager.set_password(username, SecretStr(password))

    try:
        asyncio.run(reset())
    except ValidationError as error:
        raise password_error(error) from error
    except ValueError as error:
        raise typer.BadParameter(str(error)) from error
    typer.echo(f"Password reset for {username}. Browser sessions have been signed out.")

octomate_cli.mcp

Generate local MCP tentacle configuration from provider presets.

mcp_typer module-attribute 

mcp_typer = typer.Typer(
    help="Configure MCP tentacles.", no_args_is_help=True
)

McpPreset pydantic-model

Bases: BaseModel

A provider preset rendered into an MCP tentacle's configuration.

Config:

  • str_strip_whitespace: True
  • extra: forbid

Fields:

Source code in cli/octomate_cli/mcp.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class McpPreset(BaseModel):
    """A provider preset rendered into an MCP tentacle's configuration."""

    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    provider: Literal["github"]
    name: str = Field(min_length=1)
    client_id: str = Field(min_length=1)
    read_only: bool = False

    def configuration(self) -> dict[str, JsonValue]:
        template = Template(
            files("octomate_cli")
            .joinpath("presets", f"{self.provider}.yaml")
            .read_text()
        )
        rendered = template.substitute(
            name=json.dumps(self.name),
            client_id=json.dumps(self.client_id),
            endpoint="readonly" if self.read_only else "",
        )
        return TypeAdapter(dict[str, dict[str, dict[str, JsonValue]]]).validate_python(
            yaml.safe_load(rendered)
        )["tentacles"][self.name]

model_config class-attribute instance-attribute 

model_config = ConfigDict(
    str_strip_whitespace=True, extra="forbid"
)

provider pydantic-field 

provider: Literal['github']

name pydantic-field 

name: str

client_id pydantic-field 

client_id: str

read_only pydantic-field 

read_only: bool = False

configuration 

configuration() -> dict[str, JsonValue]
Source code in cli/octomate_cli/mcp.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def configuration(self) -> dict[str, JsonValue]:
    template = Template(
        files("octomate_cli")
        .joinpath("presets", f"{self.provider}.yaml")
        .read_text()
    )
    rendered = template.substitute(
        name=json.dumps(self.name),
        client_id=json.dumps(self.client_id),
        endpoint="readonly" if self.read_only else "",
    )
    return TypeAdapter(dict[str, dict[str, dict[str, JsonValue]]]).validate_python(
        yaml.safe_load(rendered)
    )["tentacles"][self.name]

preset 

preset(
    provider: Annotated[
        Literal["github"], Argument(help="Provider preset.")
    ],
    client_id: Annotated[
        str | None,
        Option(
            help="OAuth application client ID; enable Device Flow in the app."
        ),
    ] = None,
    name: Annotated[
        str | None, Option(help="Tentacle ID.")
    ] = None,
    output: Annotated[
        Path | None,
        Option(
            help="Destination YAML; defaults to tentacles.yaml in the config home."
        ),
    ] = None,
    read_only: Annotated[
        bool,
        Option(
            help="Use GitHub's read-only MCP endpoint. OAuth scopes still permit repository writes."
        ),
    ] = False,
) -> None

Save an MCP tentacle preset with endpoints and scopes filled in.

Source code in cli/octomate_cli/mcp.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@mcp_typer.command()
def preset(
    provider: Annotated[Literal["github"], typer.Argument(help="Provider preset.")],
    client_id: Annotated[
        str | None,
        typer.Option(
            help="OAuth application client ID; enable Device Flow in the app."
        ),
    ] = None,
    name: Annotated[str | None, typer.Option(help="Tentacle ID.")] = None,
    output: Annotated[
        Path | None,
        typer.Option(
            help="Destination YAML; defaults to tentacles.yaml in the config home."
        ),
    ] = None,
    read_only: Annotated[
        bool,
        typer.Option(
            help="Use GitHub's read-only MCP endpoint. OAuth scopes still permit repository writes."
        ),
    ] = False,
) -> None:
    """Save an MCP tentacle preset with endpoints and scopes filled in."""
    if client_id is None:
        client_id = Prompt.ask(
            "OAuth application client ID", console=Console(stderr=True)
        )
    if not client_id.strip() or (name is not None and not name.strip()):
        raise typer.BadParameter("Client ID and tentacle name must not be empty.")
    selection = McpPreset(
        provider=provider,
        name=name or provider,
        client_id=client_id,
        read_only=read_only,
    )
    path = (
        output.expanduser() if output is not None else config_home() / "tentacles.yaml"
    )
    adapter = TypeAdapter(dict[str, JsonValue])
    existing = yaml.safe_load(path.read_text()) if path.exists() else None
    config = adapter.validate_python(existing if existing is not None else {})
    mcps = adapter.validate_python(config.get("tentacles", {}))
    tentacle_id = selection.name
    if tentacle_id in mcps:
        raise typer.BadParameter(
            f"MCP tentacle {tentacle_id!r} already exists in {path}."
        )
    mcps[tentacle_id] = selection.configuration()
    config["tentacles"] = mcps
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(yaml.safe_dump(config, sort_keys=False))
    typer.echo(f"Saved MCP tentacle {tentacle_id!r} to {path.resolve()}")
    typer.echo(
        f"Set OCTOMATE__TENTACLES__{tentacle_id.upper()}__CLIENT_SECRET in your .env "
        "and configure oauth.callback_base_uri. Register "
        f"<callback_base_uri>/oauth/{tentacle_id}/callback in the GitHub OAuth App."
    )