Skip to content

Native integrations

octomate_cli.tentacles.hooks

What every agent's hook installer shares: the check that the credential the hooks will carry — a server-issued API token — actually resolves, and the scripts the installed hooks run.

EMIT_SCRIPT module-attribute 

EMIT_SCRIPT = Path(__file__).parent.parent / 'emit.py'

LAUNCH_SCRIPT module-attribute 

LAUNCH_SCRIPT = Path(__file__).parent.parent / 'launch.py'

announce_secret 

announce_secret() -> None

Warn when hooks were just installed against a credential that resolves to nothing: the install reports success, and every turn after it 401s.

Source code in cli/octomate_cli/tentacles/hooks.py 
24
25
26
27
28
29
30
31
32
33
34
def announce_secret() -> None:
    """Warn when hooks were just installed against a credential that resolves to
    nothing: the install reports success, and every turn after it 401s."""
    if cli_settings().token is None:
        typer.secho(
            f"\nNo credential found — neither ./.octomate/cli.toml nor "
            f"{user_config_path()} holds one, and ${CLISettings.env('token')} is "
            "unset. Run `octomate configure --token <api-token>`, or every hook will be refused.",
            fg=typer.colors.YELLOW,
            err=True,
        )

octomate_cli.tentacles.mcp

Octomate's static MCP client config — the three facts every octomate <runtime> mcp install writes: where the server is, the credential, and which runtime the calls are from.

Owned here so the runtimes' installers cannot drift apart. The server holds the same literals (octomate.mcp.server, octomate.mcp.gateway, octomate.types.threads); the CLI cannot import that half, so the tests are what hold the two together. Unlike the hook commands — whose scripts resolve the address and credential from the environment when each hook fires — a static MCP entry is read by the runtime itself, so both are resolved once, at install time, and written into the file: the file holds the literal credential, and rotating it means re-running install.

OCTOMATE_MCP_PATH module-attribute 

OCTOMATE_MCP_PATH = '/octomate/mcp'

OCTOMATE_SERVER_KEY module-attribute 

OCTOMATE_SERVER_KEY = 'octomate'

CLIENT_HEADER module-attribute 

CLIENT_HEADER = 'X-Octomate-Client'

CLAUDE_NATIVE_CLIENT module-attribute 

CLAUDE_NATIVE_CLIENT = 'claude-native'

CODEX_NATIVE_CLIENT module-attribute 

CODEX_NATIVE_CLIENT = 'codex-native'

DEEPSEEK_NATIVE_CLIENT module-attribute 

DEEPSEEK_NATIVE_CLIENT = 'deepseek-native'

octomate_url 

octomate_url(url: str | None) -> str

The full /octomate/mcp URL an install writes, from the pinned base or the client's own resolution — refused when nothing names an address, since a static entry pointing nowhere would fail every session's tool listing.

Source code in cli/octomate_cli/tentacles/mcp.py 
41
42
43
44
45
46
47
48
49
50
51
52
def octomate_url(url: str | None) -> str:
    """The full `/octomate/mcp` URL an install writes, from the pinned base or the
    client's own resolution — refused when nothing names an address, since a
    static entry pointing nowhere would fail every session's tool listing."""
    base = url if url is not None else cli_settings().url
    if base is None:
        raise typer.BadParameter(
            f"no --url given, {CLISettings.env('url')} is unset, and no cli.toml names "
            "a url — a static MCP entry needs a concrete address; run "
            "`octomate configure --url http://<host>:<port>`"
        )
    return base.rstrip("/") + OCTOMATE_MCP_PATH

octomate_secret 

octomate_secret() -> str

The credential an install embeds in the entry's Authorization header — refused when nothing resolves, since the entry would 401 on every call.

Source code in cli/octomate_cli/tentacles/mcp.py 
55
56
57
58
59
60
61
62
63
64
65
def octomate_secret() -> str:
    """The credential an install embeds in the entry's Authorization header —
    refused when nothing resolves, since the entry would 401 on every call."""
    token = cli_settings().token
    if token is None:
        raise typer.BadParameter(
            f"no credential resolves — {CLISettings.env('token')} is unset and no "
            "cli.toml holds one; run `octomate configure --token <api-token>` first. The entry embeds the "
            "literal credential, so installing without one would only 401."
        )
    return token

octomate_cli.tentacles.claude.hooks

Install and manage the claude tentacle's native hooks.

HandledHookEvent module-attribute 

HandledHookEvent = Literal[
    "UserPromptSubmit",
    "Stop",
    "SessionEnd",
    "SubagentStart",
    "SubagentStop",
]

HANDLED_HOOK_EVENTS module-attribute 

HANDLED_HOOK_EVENTS: tuple[HandledHookEvent, ...] = (
    "UserPromptSubmit",
    "Stop",
    "SessionEnd",
    "SubagentStart",
    "SubagentStop",
)

TELEPORT_TOOL_MATCHER module-attribute 

TELEPORT_TOOL_MATCHER = 'mcp__.*__gateway_teleport'

HOOK_TIMEOUT module-attribute 

HOOK_TIMEOUT = 10

CLAUDE_HOOK_PATH module-attribute 

CLAUDE_HOOK_PATH = '/hooks/claude'

CLAUDE_STREAM_PATH module-attribute 

CLAUDE_STREAM_PATH = '/hooks/claude/stream'

hooks_typer module-attribute 

hooks_typer = typer.Typer(
    help="Manage the Claude Code hook pipe into Octomate.",
    no_args_is_help=True,
)

ScopeOption module-attribute 

ScopeOption = Annotated[
    Scope,
    typer.Option(
        help="Which settings file to touch: 'user' (~/.claude) or 'project' (./.claude)."
    ),
]

SettingsOption module-attribute 

SettingsOption = Annotated[
    Path | None,
    typer.Option(
        help="Explicit settings.json path; overrides --scope when given."
    ),
]

Scope

Bases: str, Enum

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
77
78
79
class Scope(str, Enum):  # noqa: UP042
    user = "user"
    project = "project"

user class-attribute instance-attribute 

user = 'user'

project class-attribute instance-attribute 

project = 'project'

settings_file 

settings_file(scope: Scope, settings: Path | None) -> Path
Source code in cli/octomate_cli/tentacles/claude/hooks.py 
 96
 97
 98
 99
100
def settings_file(scope: Scope, settings: Path | None) -> Path:
    if settings is not None:
        return settings
    root = Path.home() if scope is Scope.user else Path.cwd()
    return root / ".claude" / "settings.json"

claude_emit_handler 

claude_emit_handler(url: str | None) -> CommandHook

One forwarding command hook: emit.py carries the event body from stdin to the hook router, reading the credential — and, unless url pins one, the router's address (OCTOMATE_CLI_URL) — from the environment at fire time. A command rather than a native http handler so the settings file stays free of hosts and credentials both: the same install serves whichever server the environment names. Synchronous either way, which is what guarantees delivery before a short-lived claude -p exits.

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def claude_emit_handler(url: str | None) -> CommandHook:
    """One forwarding `command` hook: `emit.py` carries the event body from stdin to
    the hook router, reading the credential — and, unless `url` pins one, the router's
    address (`OCTOMATE_CLI_URL`) — from the environment at fire time. A command rather
    than a native `http` handler so the settings file stays free of hosts and
    credentials both: the same install serves whichever server the environment names.
    Synchronous either way, which is what guarantees delivery before a short-lived
    `claude -p` exits."""
    command = [sys.executable, str(EMIT_SCRIPT), "--path", CLAUDE_HOOK_PATH]
    if url is not None:
        command += ["--url", url]
    return CommandHook(
        type="command", command=shlex.join(command), timeout=HOOK_TIMEOUT
    )

stream_url_for 

stream_url_for(hook_url: str) -> str

The stream endpoint the hook URL implies: same host, ws(s) for http(s).

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
119
120
121
122
123
def stream_url_for(hook_url: str) -> str:
    """The stream endpoint the hook URL implies: same host, ws(s) for http(s)."""
    base = hook_url.removesuffix(CLAUDE_HOOK_PATH)
    scheme, _, rest = base.partition("://")
    return f"{'wss' if scheme == 'https' else 'ws'}://{rest}{CLAUDE_STREAM_PATH}"

claude_launch_handler 

claude_launch_handler(hook_url: str | None) -> CommandHook

The launcher command hook: spawns octomate claude tail for the session, detached (launch.py) — the forwarding hooks reach Octomate but can start nothing on this machine, and the stream needs a local process. The command pins this installer's own interpreter and octomate script by absolute path, so it works from whatever shell Claude runs hooks in; the stream address is pinned only when the install pinned --url, and otherwise resolved from OCTOMATE_CLI_URL at fire time, like the credential always is.

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def claude_launch_handler(hook_url: str | None) -> CommandHook:
    """The launcher `command` hook: spawns `octomate claude tail` for the session,
    detached (`launch.py`) — the forwarding hooks reach Octomate but can start nothing
    on this machine, and the stream needs a local process. The command pins this
    installer's own interpreter and octomate script by absolute path, so it works from
    whatever shell Claude runs hooks in; the stream address is pinned only when the
    install pinned `--url`, and otherwise resolved from `OCTOMATE_CLI_URL` at fire time,
    like the credential always is."""
    command = [
        sys.executable,
        str(LAUNCH_SCRIPT),
        "--path",
        CLAUDE_HOOK_PATH,
        "--octomate",
        str(Path(sys.argv[0]).resolve()),
    ]
    if hook_url is not None:
        command += ["--url", stream_url_for(hook_url)]
    return CommandHook(
        type="command", command=shlex.join(command), timeout=HOOK_TIMEOUT
    )

is_octomate_hook 

is_octomate_hook(hook: Hook) -> bool

A handler this installer wrote: a command carrying Octomate's hook path (which the stream path extends, so pinned launchers of every age match too), or the http handler an older install pointed at it. Matched by path, not the exact command, so a re-install replaces a stale handler whatever its host, port, interpreter, or script location — every generation back to the http ones.

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
149
150
151
152
153
154
155
156
157
158
159
def is_octomate_hook(hook: Hook) -> bool:
    """A handler this installer wrote: a `command` carrying Octomate's hook path
    (which the stream path extends, so pinned launchers of every age match too), or
    the `http` handler an older install pointed at it. Matched by path, not the exact
    command, so a re-install replaces a stale handler whatever its host, port,
    interpreter, or script location — every generation back to the http ones."""
    if isinstance(hook, HttpHook):
        return hook.url.endswith(CLAUDE_HOOK_PATH)
    if isinstance(hook, CommandHook):
        return CLAUDE_HOOK_PATH in hook.command
    return False

install 

install(url: Annotated[str | None, Option(help=f'Full hook URL to pin. Without it, hooks resolve ${env('url')} from each session's environment at fire time.')] = None, scope: ScopeOption = Scope.user, settings: SettingsOption = None) -> None

Point native Claude Code sessions at Octomate's hook router.

The transcript-stream launcher (a command hook on UserPromptSubmit) installs unconditionally, the machine Octomate itself runs on included: the stream is the only assembler — the server never reads a transcript from disk — so a session without a tail keeps only the hooks' sketch. Merges into the settings file: other hooks are preserved, and re-running replaces a stale Octomate handler in place rather than stacking another.

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
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
@hooks_typer.command("install")
def install(
    url: Annotated[
        str | None,
        typer.Option(
            help="Full hook URL to pin. Without it, hooks resolve "
            f"${CLISettings.env('url')} from each session's environment at fire time."
        ),
    ] = None,
    scope: ScopeOption = Scope.user,
    settings: SettingsOption = None,
) -> None:
    """Point native Claude Code sessions at Octomate's hook router.

    The transcript-stream launcher (a command hook on UserPromptSubmit) installs
    unconditionally, the machine Octomate itself runs on included: the stream is the
    only assembler — the server never reads a transcript from disk — so a session
    without a tail keeps only the hooks' sketch. Merges into the settings file:
    other hooks are preserved, and re-running replaces a stale Octomate handler in
    place rather than stacking another.
    """
    path = settings_file(scope, settings)
    document = HookSettings.load(path)
    hooks = document.hooks

    group = HookGroup(hooks=[claude_emit_handler(url)])
    # The transcript-stream launcher rides the same event the ledger's first write
    # does: by the time it fires, the forwarding hook has already created the session
    # server-side, and the tail it spawns is deduplicated per session.
    launcher_group = HookGroup(hooks=[claude_launch_handler(url)])
    # Every event present, not just the handled ones: an event Octomate once registered
    # and no longer does (`SessionStart`) would otherwise keep a stale handler forever.
    for event in {*hooks, *HANDLED_HOOK_EVENTS}:
        document.remove_handlers(event, is_octomate_hook)
        if event in HANDLED_HOOK_EVENTS:
            hooks.setdefault(event, []).append(group)
        if event == "UserPromptSubmit":
            hooks[event].append(launcher_group)
    hooks.setdefault("PreToolUse", []).append(
        HookGroup.model_validate(
            {"matcher": TELEPORT_TOOL_MATCHER, "hooks": [claude_emit_handler(url)]}
        )
    )
    document.hooks = hooks
    document.write(path)

    target = url if url is not None else f"${CLISettings.env('url')} at fire time"
    typer.echo(f"Installed Octomate hooks → {target}")
    typer.echo(f"  events:   {', '.join(HANDLED_HOOK_EVENTS)}")
    typer.echo(f"  teleport: PreToolUse on {TELEPORT_TOOL_MATCHER}")
    stream = (
        stream_url_for(url)
        if url is not None
        else f"derived from ${CLISettings.env('url')}"
    )
    typer.echo(f"  stream:   {stream} (via {LAUNCH_SCRIPT.name})")
    typer.echo(f"  settings: {path}")
    typer.echo(
        f"  auth:     Bearer ${{{CLISettings.env('token')}}} from the environment"
    )
    announce_secret()

uninstall 

uninstall(
    scope: ScopeOption = Scope.user,
    settings: SettingsOption = None,
) -> None

Remove Octomate's hook handlers from a Claude settings file, leaving any other hooks untouched.

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@hooks_typer.command("uninstall")
def uninstall(scope: ScopeOption = Scope.user, settings: SettingsOption = None) -> None:
    """Remove Octomate's hook handlers from a Claude settings file, leaving any other
    hooks untouched."""
    path = settings_file(scope, settings)
    document = HookSettings.load(path)
    if not document.hooks:
        typer.echo(f"No Octomate hooks in {path}")
        raise typer.Exit()

    for event in list(document.hooks):
        document.remove_handlers(event, is_octomate_hook)
    document.write(path)
    typer.echo(f"Removed Octomate hooks from {path}")

show 

show(
    scope: ScopeOption = Scope.user,
    settings: SettingsOption = None,
) -> None

Show the Octomate hook handlers currently installed in a Claude settings file.

Source code in cli/octomate_cli/tentacles/claude/hooks.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
@hooks_typer.command("show")
def show(scope: ScopeOption = Scope.user, settings: SettingsOption = None) -> None:
    """Show the Octomate hook handlers currently installed in a Claude settings file."""
    path = settings_file(scope, settings)
    document = HookSettings.load(path)
    found = [
        (event, hook)
        for event, groups in document.hooks.items()
        for group in groups
        for hook in group.hooks
        if isinstance(hook, CommandHook | HttpHook) and is_octomate_hook(hook)
    ]
    if not found:
        typer.echo(f"No Octomate hooks in {path}")
        raise typer.Exit()

    typer.echo(f"Octomate hooks in {path}:")
    for event, hook in found:
        target = hook.url if isinstance(hook, HttpHook) else hook.command
        typer.echo(f"  {event}: {target} (timeout {hook.timeout}s)")

octomate_cli.tentacles.claude.mcp

Install and manage the claude tentacle's MCP configuration.

mcp_typer module-attribute 

mcp_typer = typer.Typer(
    help="Manage the Octomate MCP entry for native Claude Code sessions.",
    no_args_is_help=True,
)

McpScopeOption module-attribute 

McpScopeOption = Annotated[
    McpScope,
    typer.Option(
        help="Where the entry lives: 'local' (~/.claude.json, this project only), 'user' (~/.claude.json, every project), or 'project' (./.mcp.json)."
    ),
]

McpFileOption module-attribute 

McpFileOption = Annotated[
    Path | None,
    typer.Option(
        "--file",
        help="Explicit config path; replaces the file --scope implies, while the scope keeps deciding where in it the entry lives.",
    ),
]

McpScope

Bases: str, Enum

Claude's three MCP config placements. local — the default, since the entry embeds a credential — is Claude's per-project slot inside ~/.claude.json (projects.<cwd>.mcpServers, the shape claude mcp add --scope local writes): scoped to this directory without putting a secret-bearing file in the repo.

Source code in cli/octomate_cli/tentacles/claude/mcp.py 
29
30
31
32
33
34
35
36
37
class McpScope(str, Enum):  # noqa: UP042
    """Claude's three MCP config placements. `local` — the default, since the entry
    embeds a credential — is Claude's per-project slot inside `~/.claude.json`
    (`projects.<cwd>.mcpServers`, the shape `claude mcp add --scope local` writes):
    scoped to this directory without putting a secret-bearing file in the repo."""

    local = "local"
    user = "user"
    project = "project"

local class-attribute instance-attribute 

local = 'local'

user class-attribute instance-attribute 

user = 'user'

project class-attribute instance-attribute 

project = 'project'

mcp_config_file 

mcp_config_file(scope: McpScope, file: Path | None) -> Path
Source code in cli/octomate_cli/tentacles/claude/mcp.py 
59
60
61
62
63
64
def mcp_config_file(scope: McpScope, file: Path | None) -> Path:
    if file is not None:
        return file
    if scope is McpScope.project:
        return Path.cwd() / ".mcp.json"
    return Path.home() / ".claude.json"

mcp_install 

mcp_install(url: Annotated[str | None, Option(help=f'Octomate's base URL (http://host:port) to write; defaults to ${env('url')}, then cli.toml.')] = None, scope: McpScopeOption = McpScope.local, file: McpFileOption = None) -> None

Point native Claude Code sessions at the served MCP server.

Writes the entry — the server's URL, the bearer, and the runtime attribution header — resolved once, now: unlike the hooks, a static entry is read by Claude itself, so the file holds the literal credential and rotating it means re-running install. Everything else in the file is kept, and re-running replaces the entry in place.

Source code in cli/octomate_cli/tentacles/claude/mcp.py 
 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
@mcp_typer.command("install")
def mcp_install(
    url: Annotated[
        str | None,
        typer.Option(
            help="Octomate's base URL (http://host:port) to write; defaults to "
            f"${CLISettings.env('url')}, then cli.toml."
        ),
    ] = None,
    scope: McpScopeOption = McpScope.local,
    file: McpFileOption = None,
) -> None:
    """Point native Claude Code sessions at the served MCP server.

    Writes the entry — the server's URL, the bearer, and the runtime
    attribution header — resolved once, now: unlike the hooks, a static entry is
    read by Claude itself, so the file holds the literal credential and rotating
    it means re-running install. Everything else in the file is kept, and
    re-running replaces the entry in place.
    """
    target = octomate_url(url)
    secret = octomate_secret()
    path = mcp_config_file(scope, file)
    document = load_settings(path)
    if scope is McpScope.local:
        container = document.projects.setdefault(str(Path.cwd()), McpContainer())
        document.model_fields_set.add("projects")
    else:
        container = document
    container.mcp_servers[OCTOMATE_SERVER_KEY] = McpServer(
        type="http",
        url=target,
        headers={
            "Authorization": f"Bearer {secret}",
            CLIENT_HEADER: CLAUDE_NATIVE_CLIENT,
        },
    )
    container.model_fields_set.add("mcp_servers")
    write_settings(path, document)
    typer.echo(f"Installed the Octomate MCP entry → {target}")
    typer.echo(f"  file:   {path}")
    if scope is McpScope.local:
        typer.echo(f"  scope:  local ({Path.cwd()})")
    typer.echo(f"  client: {CLAUDE_NATIVE_CLIENT}")
    typer.echo(
        "  auth:   embedded — the file holds the literal credential; rotation "
        "means re-running install"
    )

mcp_uninstall 

mcp_uninstall(
    scope: McpScopeOption = McpScope.local,
    file: McpFileOption = None,
) -> None

Remove the Octomate MCP entry, leaving every other server and setting.

Source code in cli/octomate_cli/tentacles/claude/mcp.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@mcp_typer.command("uninstall")
def mcp_uninstall(
    scope: McpScopeOption = McpScope.local, file: McpFileOption = None
) -> None:
    """Remove the Octomate MCP entry, leaving every other server and setting."""
    path = mcp_config_file(scope, file)
    document = load_settings(path)
    key = str(Path.cwd())
    container = document.projects.get(key) if scope is McpScope.local else document
    if container is None or OCTOMATE_SERVER_KEY not in container.mcp_servers:
        typer.echo(f"No Octomate MCP entry in {path}")
        raise typer.Exit()
    del container.mcp_servers[OCTOMATE_SERVER_KEY]
    if not container.mcp_servers:
        container.model_fields_set.discard("mcp_servers")
        if scope is McpScope.local and not container.model_dump(exclude_unset=True):
            del document.projects[key]
            if not document.projects:
                document.model_fields_set.discard("projects")
    write_settings(path, document)
    typer.echo(f"Removed the Octomate MCP entry from {path}")

mcp_show 

mcp_show(
    scope: McpScopeOption = McpScope.local,
    file: McpFileOption = None,
) -> None

Show the Octomate MCP entry, credential masked.

Source code in cli/octomate_cli/tentacles/claude/mcp.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@mcp_typer.command("show")
def mcp_show(
    scope: McpScopeOption = McpScope.local, file: McpFileOption = None
) -> None:
    """Show the Octomate MCP entry, credential masked."""
    path = mcp_config_file(scope, file)
    document = load_settings(path)
    container = (
        document.projects.get(str(Path.cwd())) if scope is McpScope.local else document
    )
    entry = (
        container.mcp_servers.get(OCTOMATE_SERVER_KEY)
        if container is not None
        else None
    )
    if entry is None:
        typer.echo(f"No Octomate MCP entry in {path}")
        raise typer.Exit()
    client = entry.headers.get(CLIENT_HEADER) if entry.headers is not None else None
    typer.echo(f"Octomate MCP entry in {path}:")
    typer.echo(f"  url:    {entry.url}")
    typer.echo(f"  client: {client}")
    typer.echo("  auth:   Bearer ***")

octomate_cli.tentacles.codex.hooks

Install and manage the codex tentacle's native hooks.

HOOK_TIMEOUT module-attribute 

HOOK_TIMEOUT = 10

CODEX_HOOK_PATH module-attribute 

CODEX_HOOK_PATH = '/hooks/codex'

CODEX_STREAM_PATH module-attribute 

CODEX_STREAM_PATH = '/hooks/codex/stream'

HANDLED_HOOK_EVENTS module-attribute 

HANDLED_HOOK_EVENTS = (
    "SessionStart",
    "UserPromptSubmit",
    "Stop",
    "SubagentStart",
    "SubagentStop",
)

LAUNCHER_HOOK_EVENTS module-attribute 

LAUNCHER_HOOK_EVENTS = (
    "SessionStart",
    "UserPromptSubmit",
    "SubagentStop",
)

hooks_typer module-attribute 

hooks_typer = typer.Typer(
    help="Manage the Codex hook pipe into Octomate.",
    no_args_is_help=True,
)

Scope

Bases: str, Enum

Source code in cli/octomate_cli/tentacles/codex/hooks.py 
60
61
62
class Scope(str, Enum):  # noqa: UP042
    user = "user"
    project = "project"

user class-attribute instance-attribute 

user = 'user'

project class-attribute instance-attribute 

project = 'project'

hooks_file 

hooks_file(scope: Scope, path: Path | None) -> Path
Source code in cli/octomate_cli/tentacles/codex/hooks.py 
65
66
67
68
69
def hooks_file(scope: Scope, path: Path | None) -> Path:
    if path is not None:
        return path
    root = Path.home() if scope is Scope.user else Path.cwd()
    return root / ".codex" / "hooks.json"

is_octomate_handler 

is_octomate_handler(value: Hook) -> bool

A command hook aimed at Octomate's Codex hook path (which the stream path extends, so pinned launchers match too) — matched by path, not the exact command, so a re-install replaces a stale handler whatever its host, port, or interpreter, including ones written by versions that ran codex hooks emit.

Source code in cli/octomate_cli/tentacles/codex/hooks.py 
72
73
74
75
76
77
def is_octomate_handler(value: Hook) -> bool:
    """A command hook aimed at Octomate's Codex hook path (which the stream path
    extends, so pinned launchers match too) — matched by path, not the exact command,
    so a re-install replaces a stale handler whatever its host, port, or interpreter,
    including ones written by versions that ran `codex hooks emit`."""
    return isinstance(value, CommandHook) and CODEX_HOOK_PATH in value.command

stream_url_for 

stream_url_for(hook_url: str) -> str

The stream endpoint the hook URL implies: same host, ws(s) for http(s).

Source code in cli/octomate_cli/tentacles/codex/hooks.py 
80
81
82
83
84
def stream_url_for(hook_url: str) -> str:
    """The stream endpoint the hook URL implies: same host, ws(s) for http(s)."""
    base = hook_url.removesuffix(CODEX_HOOK_PATH)
    scheme, _, rest = base.partition("://")
    return f"{'wss' if scheme == 'https' else 'ws'}://{rest}{CODEX_STREAM_PATH}"

codex_launch_handler 

codex_launch_handler(hook_url: str | None) -> CommandHook

The launcher command hook: spawns octomate codex tail for the session, detached (launch.py) — the forwarding hooks reach Octomate but can start nothing on this machine, and the stream needs a local process. The command pins this installer's own interpreter and octomate script by absolute path, so it works from whatever shell Codex runs hooks in; the stream address is pinned only when the install pinned --url, and otherwise resolved from OCTOMATE_CLI_URL at fire time, like the credential always is.

Source code in cli/octomate_cli/tentacles/codex/hooks.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def codex_launch_handler(hook_url: str | None) -> CommandHook:
    """The launcher `command` hook: spawns `octomate codex tail` for the session,
    detached (`launch.py`) — the forwarding hooks reach Octomate but can start
    nothing on this machine, and the stream needs a local process. The command pins
    this installer's own interpreter and octomate script by absolute path, so it
    works from whatever shell Codex runs hooks in; the stream address is pinned only
    when the install pinned `--url`, and otherwise resolved from `OCTOMATE_CLI_URL` at
    fire time, like the credential always is."""
    command = [
        sys.executable,
        str(LAUNCH_SCRIPT),
        "--path",
        CODEX_HOOK_PATH,
        "--agent",
        "codex",
        "--octomate",
        str(Path(sys.argv[0]).resolve()),
    ]
    if hook_url is not None:
        command += ["--url", stream_url_for(hook_url)]
    return CommandHook(
        type="command", command=shlex.join(command), timeout=HOOK_TIMEOUT
    )

install 

install(url: Annotated[str | None, Option(help=f'Full hook URL to pin. Without it, hooks resolve ${env('url')} from each session's environment at fire time.')] = None, scope: Annotated[Scope, Option()] = Scope.user, path: Annotated[Path | None, Option('--hooks-file')] = None) -> None

Install observing command hooks without replacing the operator's hooks.

The transcript-stream launcher installs unconditionally, the machine Octomate itself runs on included: the stream is the only assembler — the server never reads a rollout from disk — so a session without a tail keeps only the hooks' sketch. Re-running replaces a stale Octomate handler in place rather than stacking another.

Source code in cli/octomate_cli/tentacles/codex/hooks.py 
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
@hooks_typer.command("install")
def install(
    url: Annotated[
        str | None,
        typer.Option(
            help="Full hook URL to pin. Without it, hooks resolve "
            f"${CLISettings.env('url')} from each session's environment at fire time."
        ),
    ] = None,
    scope: Annotated[Scope, typer.Option()] = Scope.user,
    path: Annotated[Path | None, typer.Option("--hooks-file")] = None,
) -> None:
    """Install observing command hooks without replacing the operator's hooks.

    The transcript-stream launcher installs unconditionally, the machine Octomate
    itself runs on included: the stream is the only assembler — the server never
    reads a rollout from disk — so a session without a tail keeps only the hooks'
    sketch. Re-running replaces a stale Octomate handler in place rather than
    stacking another.
    """
    target = hooks_file(scope, path)
    document = HookSettings.load(target)
    hooks = document.hooks
    # By absolute path, not `-m octomate...`: importing the package costs ~1.9s, which
    # Codex would pay on every blocking hook. Through `sys.executable` so the hook runs
    # on this interpreter, not whichever `python` the session's PATH resolves.
    parts = [sys.executable, str(EMIT_SCRIPT), "--path", CODEX_HOOK_PATH]
    if url is not None:
        parts += ["--url", url]
    command = shlex.join(parts)
    group = HookGroup(
        hooks=[CommandHook(type="command", command=command, timeout=HOOK_TIMEOUT)]
    )
    launcher_group = HookGroup(hooks=[codex_launch_handler(url)])
    for event in HANDLED_HOOK_EVENTS:
        document.remove_handlers(event, is_octomate_handler)
        hooks.setdefault(event, []).append(group)
        if event in LAUNCHER_HOOK_EVENTS:
            hooks[event].append(launcher_group)
    document.hooks = hooks
    document.write(target)
    hook_target = url if url is not None else f"${CLISettings.env('url')} at fire time"
    typer.echo(f"Installed Octomate Codex hooks in {target} → {hook_target}")
    typer.echo(f"  events: {', '.join(HANDLED_HOOK_EVENTS)}")
    typer.echo(f"  emit:   {EMIT_SCRIPT}")
    stream = (
        stream_url_for(url)
        if url is not None
        else f"derived from ${CLISettings.env('url')}"
    )
    typer.echo(f"  stream: {stream} (via {LAUNCH_SCRIPT.name})")
    typer.echo("Open /hooks in Codex and trust the new command hooks.")
    announce_secret()

uninstall 

uninstall(
    scope: Annotated[Scope, Option()] = Scope.user,
    path: Annotated[
        Path | None, Option("--hooks-file")
    ] = None,
) -> None
Source code in cli/octomate_cli/tentacles/codex/hooks.py 
167
168
169
170
171
172
173
174
175
176
177
@hooks_typer.command("uninstall")
def uninstall(
    scope: Annotated[Scope, typer.Option()] = Scope.user,
    path: Annotated[Path | None, typer.Option("--hooks-file")] = None,
) -> None:
    target = hooks_file(scope, path)
    document = HookSettings.load(target)
    for event in list(document.hooks):
        document.remove_handlers(event, is_octomate_handler)
    document.write(target)
    typer.echo(f"Removed Octomate Codex hooks from {target}")

octomate_cli.tentacles.codex.mcp

Install and manage the codex tentacle's MCP configuration.

mcp_typer module-attribute 

mcp_typer = typer.Typer(
    help="Manage the Octomate MCP entry for native Codex sessions.",
    no_args_is_help=True,
)

McpConfigOption module-attribute 

McpConfigOption = Annotated[
    Path | None,
    typer.Option(
        "--config-file",
        help="Explicit config.toml path; defaults to ~/.codex/config.toml.",
    ),
]

mcp_config_file 

mcp_config_file(path: Path | None) -> Path
Source code in cli/octomate_cli/tentacles/codex/mcp.py 
40
41
def mcp_config_file(path: Path | None) -> Path:
    return path if path is not None else Path.home() / ".codex" / "config.toml"

load_toml 

load_toml(path: Path) -> tomlkit.TOMLDocument

The whole document, comments and all — tomlkit's round-trip is what keeps the operator's file theirs.

Source code in cli/octomate_cli/tentacles/codex/mcp.py 
44
45
46
47
def load_toml(path: Path) -> tomlkit.TOMLDocument:
    """The whole document, comments and all — tomlkit's round-trip is what keeps
    the operator's file theirs."""
    return tomlkit.parse(path.read_text()) if path.exists() else tomlkit.document()

mcp_install 

mcp_install(url: Annotated[str | None, Option(help=f'Octomate's base URL (http://host:port) to write; defaults to ${env('url')}, then cli.toml.')] = None, config_file: McpConfigOption = None) -> None

Point native Codex sessions at the served MCP server.

Writes the mcp_servers.octomate table — the server's URL, the bearer, the runtime attribution header, and teleport's approval requirement — preserving the operator's comments and every other table. The credential is embedded; rotating means re-running install. Driven sessions disable inherited MCP servers and supply their caller's connection separately.

Source code in cli/octomate_cli/tentacles/codex/mcp.py 
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
@mcp_typer.command("install")
def mcp_install(
    url: Annotated[
        str | None,
        typer.Option(
            help="Octomate's base URL (http://host:port) to write; defaults to "
            f"${CLISettings.env('url')}, then cli.toml."
        ),
    ] = None,
    config_file: McpConfigOption = None,
) -> None:
    """Point native Codex sessions at the served MCP server.

    Writes the `mcp_servers.octomate` table — the server's URL, the bearer,
    the runtime attribution header, and teleport's approval requirement — preserving
    the operator's comments and every other table. The credential is embedded;
    rotating means re-running install. Driven sessions disable inherited MCP servers
    and supply their caller's connection separately.
    """
    target = octomate_url(url)
    secret = octomate_secret()
    path = mcp_config_file(config_file)
    document = load_toml(path)
    servers = document.get("mcp_servers")
    if servers is None:
        servers = tomlkit.table(is_super_table=True)
        document["mcp_servers"] = servers
    if not isinstance(servers, Table):
        raise typer.BadParameter(f"{path} has a non-table 'mcp_servers' section")
    entry = tomlkit.table()
    entry["url"] = target
    headers = tomlkit.inline_table()
    headers["Authorization"] = f"Bearer {secret}"
    headers[CLIENT_HEADER] = CODEX_NATIVE_CLIENT
    entry["http_headers"] = headers
    entry["tools"] = {gateway_tool(GatewayTool.TELEPORT): {"approval_mode": "prompt"}}
    servers[OCTOMATE_SERVER_KEY] = entry
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(tomlkit.dumps(document))
    typer.echo(f"Installed the Octomate MCP entry → {target}")
    typer.echo(f"  file:   {path}")
    typer.echo(f"  client: {CODEX_NATIVE_CLIENT}")
    typer.echo(
        "  auth:   embedded — the file holds the literal credential; rotation "
        "means re-running install"
    )

mcp_uninstall 

mcp_uninstall(config_file: McpConfigOption = None) -> None

Remove the Octomate MCP entry, leaving the operator's comments and every other server.

Source code in cli/octomate_cli/tentacles/codex/mcp.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@mcp_typer.command("uninstall")
def mcp_uninstall(config_file: McpConfigOption = None) -> None:
    """Remove the Octomate MCP entry, leaving the operator's comments and every
    other server."""
    path = mcp_config_file(config_file)
    document = load_toml(path)
    servers = document.get("mcp_servers")
    if not isinstance(servers, Table) or OCTOMATE_SERVER_KEY not in servers:
        typer.echo(f"No Octomate MCP entry in {path}")
        raise typer.Exit()
    del servers[OCTOMATE_SERVER_KEY]
    if len(servers) == 0:
        del document["mcp_servers"]
    path.write_text(tomlkit.dumps(document))
    typer.echo(f"Removed the Octomate MCP entry from {path}")

mcp_show 

mcp_show(config_file: McpConfigOption = None) -> None

Show the Octomate MCP entry, credential masked.

Source code in cli/octomate_cli/tentacles/codex/mcp.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@mcp_typer.command("show")
def mcp_show(config_file: McpConfigOption = None) -> None:
    """Show the Octomate MCP entry, credential masked."""
    path = mcp_config_file(config_file)
    servers = load_toml(path).get("mcp_servers")
    entry = servers.get(OCTOMATE_SERVER_KEY) if isinstance(servers, Table) else None
    if not isinstance(entry, Table):
        typer.echo(f"No Octomate MCP entry in {path}")
        raise typer.Exit()
    headers = entry.get("http_headers")
    client = headers.get(CLIENT_HEADER) if isinstance(headers, dict) else None
    typer.echo(f"Octomate MCP entry in {path}:")
    typer.echo(f"  url:    {entry.get('url')}")
    typer.echo(f"  client: {client}")
    typer.echo("  auth:   Bearer ***")

octomate_cli.tentacles.deepseek.hooks

Install and manage the deepseek tentacle's native hooks.

HOOK_TIMEOUT module-attribute 

HOOK_TIMEOUT = 10

DEEPSEEK_HOOK_PATH module-attribute 

DEEPSEEK_HOOK_PATH = '/hooks/deepseek'

DEEPSEEK_STREAM_PATH module-attribute 

DEEPSEEK_STREAM_PATH = '/hooks/deepseek/stream'

HANDLED_HOOK_EVENTS module-attribute 

HANDLED_HOOK_EVENTS = ('UserPromptSubmit', 'Stop')

BRIDGE_PACKAGE module-attribute 

BRIDGE_PACKAGE = '@deepseek-ai/dsh-hooks-claude-code'

PATCH_ROW_ID module-attribute 

PATCH_ROW_ID = 'octomate-hooks'

hooks_typer module-attribute 

hooks_typer = typer.Typer(
    help="Manage the dsh hook pipe into Octomate.",
    no_args_is_help=True,
)

hooks_file 

hooks_file(home: Path) -> Path
Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
70
71
def hooks_file(home: Path) -> Path:
    return home / "octomate-hooks.json"

bridge_link(home: Path) -> Path
Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
74
75
def bridge_link(home: Path) -> Path:
    return home / "profiles" / "node_modules" / BRIDGE_PACKAGE

emit_handler 

emit_handler(url: str | None) -> CommandHook

The forwarding command hook: emit.py carries the event body from stdin to the hook router, reading the credential — and, unless url pins one, the router's address — from the environment at fire time.

Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
78
79
80
81
82
83
84
85
86
87
88
89
def emit_handler(url: str | None) -> CommandHook:
    """The forwarding `command` hook: `emit.py` carries the event body from
    stdin to the hook router, reading the credential — and, unless `url` pins
    one, the router's address — from the environment at fire time."""
    command = [sys.executable, str(EMIT_SCRIPT), "--path", DEEPSEEK_HOOK_PATH]
    if url is not None:
        command += ["--url", url]
    return {
        "type": "command",
        "command": shlex.join(command),
        "timeout": HOOK_TIMEOUT,
    }

stream_url_for 

stream_url_for(hook_url: str) -> str

The stream endpoint the hook URL implies: same host, ws(s) for http(s).

Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
92
93
94
95
96
def stream_url_for(hook_url: str) -> str:
    """The stream endpoint the hook URL implies: same host, ws(s) for http(s)."""
    base = hook_url.removesuffix(DEEPSEEK_HOOK_PATH)
    scheme, _, rest = base.partition("://")
    return f"{'wss' if scheme == 'https' else 'ws'}://{rest}{DEEPSEEK_STREAM_PATH}"

launch_handler 

launch_handler(hook_url: str | None) -> CommandHook

The launcher command hook: spawns octomate deepseek tail for the session, detached (launch.py) — the forwarding hook reaches Octomate but can start nothing on this machine, and the stream needs a local process to read the gateway. Pins this installer's own interpreter and octomate script by absolute path; the stream address is pinned only when the install pinned --url, and otherwise resolved from OCTOMATE_CLI_URL at fire time, like the credential always is.

Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def launch_handler(hook_url: str | None) -> CommandHook:
    """The launcher `command` hook: spawns `octomate deepseek tail` for the
    session, detached (`launch.py`) — the forwarding hook reaches Octomate but
    can start nothing on this machine, and the stream needs a local process to
    read the gateway. Pins this installer's own interpreter and octomate
    script by absolute path; the stream address is pinned only when the
    install pinned `--url`, and otherwise resolved from `OCTOMATE_CLI_URL` at fire
    time, like the credential always is."""
    command = [
        sys.executable,
        str(LAUNCH_SCRIPT),
        "--path",
        DEEPSEEK_HOOK_PATH,
        "--agent",
        "deepseek",
        "--octomate",
        str(Path(sys.argv[0]).resolve()),
    ]
    if hook_url is not None:
        command += ["--url", stream_url_for(hook_url)]
    return {"type": "command", "command": shlex.join(command), "timeout": HOOK_TIMEOUT}

hooks_document 

hooks_document(url: str | None) -> HookSettings
Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def hooks_document(url: str | None) -> HookSettings:
    handler = emit_handler(url)
    # The launcher rides the prompt event only: by the time it fires, the emit
    # hook on the same event has already created the session server-side, and
    # the tail it spawns deduplicates itself per session.
    return HookSettings(
        hooks={
            "UserPromptSubmit": [
                {"hooks": [handler]},
                {"hooks": [launch_handler(url)]},
            ],
            "Stop": [{"hooks": [handler]}],
        }
    )

patch_block 

patch_block(config_path: Path) -> str

The marker-delimited patch entry mounting the bridge over our hooks file. Textual rather than parsed YAML: the user's patch file carries their comments and !!js expressions, which a parse-and-rewrite would destroy.

Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
def patch_block(config_path: Path) -> str:
    """The marker-delimited patch entry mounting the bridge over our hooks
    file. Textual rather than parsed YAML: the user's patch file carries their
    comments and `!!js` expressions, which a parse-and-rewrite would destroy."""
    return (
        f"{MARK_BEGIN}\n"
        f"- insert:\n"
        f"    - id: {PATCH_ROW_ID}\n"
        f"      name: '{BRIDGE_PACKAGE}'\n"
        f"      config:\n"
        f"        configPath: {json.dumps(str(config_path))}\n"
        f"{MARK_END}\n"
    )

link_bridge(home: Path, bridge: Path) -> Path

Symlink the bridge package into dsh's flat module fallback, where every profile resolves plugins from. Validated first: the wrong directory here would fail every dsh boot, not just the hooks.

Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
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
def link_bridge(home: Path, bridge: Path) -> Path:
    """Symlink the bridge package into dsh's flat module fallback, where every
    profile resolves plugins from. Validated first: the wrong directory here
    would fail every dsh boot, not just the hooks."""
    manifest = bridge / "package.json"
    try:
        name = BridgeManifest.model_validate_json(manifest.read_bytes()).name
    except (OSError, ValidationError):
        raise typer.BadParameter(f"{bridge} holds no readable package.json") from None
    if name != BRIDGE_PACKAGE:
        raise typer.BadParameter(f"{bridge} is {name!r}, not {BRIDGE_PACKAGE!r}")
    if not (bridge / "lib" / "index.js").exists():
        raise typer.BadParameter(
            f"{bridge} has no lib/index.js — build the harness first"
        )
    link = bridge_link(home)
    link.parent.mkdir(parents=True, exist_ok=True)
    if link.is_symlink():
        link.unlink()
    elif link.exists():
        raise typer.BadParameter(
            f"{link} exists and is not a symlink; remove it so the installer "
            "can manage the bridge link"
        )
    link.symlink_to(bridge.resolve())
    return link

install 

install(url: Annotated[str | None, Option(help=f'Full hook URL to pin. Without it, hooks resolve ${env('url')} from each session's environment at fire time.')] = None, home: DshHomeOption = None, bridge: Annotated[Path | None, Option(help="Path of the dsh-hooks-claude-code package to link into dsh's module fallback (e.g. <harness checkout>/packages/experimental/hooks-claude-code). The bridge ships outside dsh's own bundle, so the first install needs this once.")] = None) -> None

Point native dsh sessions at Octomate's hook router.

Writes Octomate's own hooks file (never a merge — an operator's dsh hooks live in their own file, mounted by their own patch row) and a marker-delimited row in $DSH_HOME/cordis.patch.yml mounting the dsh-hooks-claude-code bridge over it. Re-running replaces both in place. Restart dsh processes for the composition change to take.

Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
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
@hooks_typer.command("install")
def install(
    url: Annotated[
        str | None,
        typer.Option(
            help="Full hook URL to pin. Without it, hooks resolve "
            f"${CLISettings.env('url')} from each session's environment at fire time."
        ),
    ] = None,
    home: DshHomeOption = None,
    bridge: Annotated[
        Path | None,
        typer.Option(
            help="Path of the dsh-hooks-claude-code package to link into dsh's "
            "module fallback (e.g. <harness checkout>/packages/experimental/"
            "hooks-claude-code). The bridge ships outside dsh's own bundle, so "
            "the first install needs this once."
        ),
    ] = None,
) -> None:
    """Point native dsh sessions at Octomate's hook router.

    Writes Octomate's own hooks file (never a merge — an operator's dsh hooks
    live in their own file, mounted by their own patch row) and a
    marker-delimited row in $DSH_HOME/cordis.patch.yml mounting the
    `dsh-hooks-claude-code` bridge over it. Re-running replaces both in place.
    Restart dsh processes for the composition change to take.
    """
    target_home = dsh_home(home)
    config_path = hooks_file(target_home)
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(hooks_document(url).model_dump_json(indent=2) + "\n")

    patch = patch_file(target_home)
    text = patch.read_text() if patch.exists() else "[]\n"
    patch.write_text(patch_text_with_block(text, patch_block(config_path)))

    if bridge is not None:
        link = link_bridge(target_home, bridge)
        typer.echo(f"Linked {BRIDGE_PACKAGE} → {link.readlink()}")
    elif not bridge_link(target_home).exists():
        typer.secho(
            f"\n{BRIDGE_PACKAGE} does not resolve from {bridge_link(target_home)} — "
            "dsh will fail to mount the hooks row until it does. Re-run with "
            "--bridge <harness checkout>/packages/experimental/hooks-claude-code.",
            fg=typer.colors.YELLOW,
            err=True,
        )

    hook_target = url if url is not None else f"${CLISettings.env('url')} at fire time"
    typer.echo(f"Installed Octomate dsh hooks → {hook_target}")
    typer.echo(f"  events: {', '.join(HANDLED_HOOK_EVENTS)}")
    typer.echo(f"  hooks:  {config_path}")
    typer.echo(f"  patch:  {patch} (row id {PATCH_ROW_ID!r})")
    stream = (
        stream_url_for(url)
        if url is not None
        else f"derived from ${CLISettings.env('url')}"
    )
    typer.echo(f"  stream: {stream} (via {LAUNCH_SCRIPT.name})")
    typer.echo("Restart dsh (the web daemon included) to load the bridge.")
    announce_secret()

uninstall 

uninstall(home: DshHomeOption = None) -> None

Remove Octomate's hooks file, patch row, and bridge link from a dsh home, leaving everything else in the patch file untouched.

Source code in cli/octomate_cli/tentacles/deepseek/hooks.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
@hooks_typer.command("uninstall")
def uninstall(home: DshHomeOption = None) -> None:
    """Remove Octomate's hooks file, patch row, and bridge link from a dsh
    home, leaving everything else in the patch file untouched."""
    target_home = dsh_home(home)
    patch = patch_file(target_home)
    if patch.exists():
        patch.write_text(patch_text_without_block(patch.read_text()))
    config_path = hooks_file(target_home)
    if config_path.exists():
        config_path.unlink()
    link = bridge_link(target_home)
    if link.is_symlink():
        link.unlink()
    typer.echo(f"Removed Octomate dsh hooks from {target_home}")

octomate_cli.tentacles.deepseek.mcp

Install and manage the deepseek tentacle's MCP configuration.

mcp_typer module-attribute 

mcp_typer = typer.Typer(
    help="Manage the Octomate MCP row for native dsh sessions.",
    no_args_is_help=True,
)

MCP_CLIENT_PACKAGE module-attribute 

MCP_CLIENT_PACKAGE = '@deepseek-ai/dsh-mcp-client'

MCP_ROW_ID module-attribute 

MCP_ROW_ID = 'octomate-mcp'

MCP_MARK_BEGIN module-attribute 

MCP_MARK_BEGIN = '# >>> octomate deepseek mcp >>>'

MCP_MARK_END module-attribute 

MCP_MARK_END = '# <<< octomate deepseek mcp <<<'

LEGACY_MCP_MARK_BEGIN module-attribute 

LEGACY_MCP_MARK_BEGIN = (
    "# >>> octomate deepseek gateway >>>"
)

LEGACY_MCP_MARK_END module-attribute 

LEGACY_MCP_MARK_END = '# <<< octomate deepseek gateway <<<'

mcp_patch_block 

mcp_patch_block(url: str, secret: str) -> str

The marker-delimited row mounting dsh's MCP client on the served server — serverName: octomate, so the model sees mcp__octomate__<tool>, the same names Claude and Codex read. Header values are JSON-quoted, which YAML reads as flow scalars: the credential is hand-written and need not be YAML-safe.

Source code in cli/octomate_cli/tentacles/deepseek/mcp.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def mcp_patch_block(url: str, secret: str) -> str:
    """The marker-delimited row mounting dsh's MCP client on the served server —
    `serverName: octomate`, so the model sees `mcp__octomate__<tool>`, the same
    names Claude and Codex read. Header values are JSON-quoted, which YAML reads
    as flow scalars: the credential is hand-written and need not be YAML-safe."""
    return (
        f"{MCP_MARK_BEGIN}\n"
        f"- insert:\n"
        f"    - id: {MCP_ROW_ID}\n"
        f"      name: '{MCP_CLIENT_PACKAGE}'\n"
        f"      config:\n"
        f"        serverName: {OCTOMATE_SERVER_KEY}\n"
        f"        transport: streamable-http\n"
        f"        url: {json.dumps(url)}\n"
        f"        headers:\n"
        f"          Authorization: {json.dumps(f'Bearer {secret}')}\n"
        f"          {CLIENT_HEADER}: {DEEPSEEK_NATIVE_CLIENT}\n"
        f"{MCP_MARK_END}\n"
    )

mcp_install 

mcp_install(url: Annotated[str | None, Option(help=f'Octomate's base URL (http://host:port) to write; defaults to ${env('url')}, then cli.toml.')] = None, home: DshHomeOption = None) -> None

Point native dsh sessions at the served MCP server.

Writes a marker-delimited row in $DSH_HOME/cordis.patch.yml mounting @deepseek-ai/dsh-mcp-client on the served MCP server, the credential and the runtime attribution embedded — resolved once, now: the file holds the literal credential, and rotating it means re-running install. Re-running replaces the row in place; restart dsh processes for the change to take.

Source code in cli/octomate_cli/tentacles/deepseek/mcp.py 
 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
@mcp_typer.command("install")
def mcp_install(
    url: Annotated[
        str | None,
        typer.Option(
            help="Octomate's base URL (http://host:port) to write; defaults to "
            f"${CLISettings.env('url')}, then cli.toml."
        ),
    ] = None,
    home: DshHomeOption = None,
) -> None:
    """Point native dsh sessions at the served MCP server.

    Writes a marker-delimited row in $DSH_HOME/cordis.patch.yml mounting
    `@deepseek-ai/dsh-mcp-client` on the served MCP server, the credential and the
    runtime attribution embedded — resolved once, now: the file holds the
    literal credential, and rotating it means re-running install. Re-running
    replaces the row in place; restart dsh processes for the change to take.
    """
    target = octomate_url(url)
    secret = octomate_secret()
    patch = patch_file(dsh_home(home))
    text = patch.read_text() if patch.exists() else "[]\n"
    text = patch_text_without_block(text, LEGACY_MCP_MARK_BEGIN, LEGACY_MCP_MARK_END)
    patch.parent.mkdir(parents=True, exist_ok=True)
    patch.write_text(
        patch_text_with_block(
            text,
            mcp_patch_block(target, secret),
            MCP_MARK_BEGIN,
            MCP_MARK_END,
        )
    )
    typer.echo(f"Installed the Octomate MCP row → {target}")
    typer.echo(f"  patch:  {patch} (row id {MCP_ROW_ID!r})")
    typer.echo(f"  client: {DEEPSEEK_NATIVE_CLIENT}")
    typer.echo(
        "  auth:   embedded — the file holds the literal credential; rotation "
        "means re-running install"
    )
    typer.echo("Restart dsh (the web daemon included) to load the row.")

mcp_uninstall 

mcp_uninstall(home: DshHomeOption = None) -> None

Remove the Octomate MCP row, leaving the hooks row and everything else in the patch file untouched.

Source code in cli/octomate_cli/tentacles/deepseek/mcp.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
@mcp_typer.command("uninstall")
def mcp_uninstall(home: DshHomeOption = None) -> None:
    """Remove the Octomate MCP row, leaving the hooks row and everything else in
    the patch file untouched."""
    patch = patch_file(dsh_home(home))
    if not patch.exists():
        typer.echo(f"No Octomate MCP row in {patch}")
        raise typer.Exit()
    text = patch_text_without_block(patch.read_text(), MCP_MARK_BEGIN, MCP_MARK_END)
    patch.write_text(
        patch_text_without_block(text, LEGACY_MCP_MARK_BEGIN, LEGACY_MCP_MARK_END)
    )
    typer.echo(f"Removed the Octomate MCP row from {patch}")

mcp_show 

mcp_show(home: DshHomeOption = None) -> None

Show the Octomate MCP row, credential masked.

Source code in cli/octomate_cli/tentacles/deepseek/mcp.py 
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
@mcp_typer.command("show")
def mcp_show(home: DshHomeOption = None) -> None:
    """Show the Octomate MCP row, credential masked."""
    patch = patch_file(dsh_home(home))
    text = patch.read_text() if patch.exists() else ""
    lines = text.splitlines()
    markers = (
        (MCP_MARK_BEGIN, MCP_MARK_END)
        if MCP_MARK_BEGIN in (line.strip() for line in lines)
        else (LEGACY_MCP_MARK_BEGIN, LEGACY_MCP_MARK_END)
    )
    if markers[0] not in (line.strip() for line in lines):
        typer.echo(f"No Octomate MCP row in {patch}")
        raise typer.Exit()
    typer.echo(f"Octomate MCP row in {patch}:")
    inside = False
    for line in lines:
        stripped = line.strip()
        if stripped == markers[0]:
            inside = True
            continue
        if stripped == markers[1]:
            break
        if inside:
            if stripped.startswith("Authorization:"):
                indent = line[: len(line) - len(line.lstrip())]
                line = f'{indent}Authorization: "Bearer ***"'
            typer.echo(f"  {line}")