feat: replace env config with typer cli options

codex_telegram_bridge/pyproject.toml

@@ -8,6 +8,7 @@ dependencies = [
   "markdown-it-py",
   "sulguk",
   "tomli; python_version < '3.11'",
+  "typer",
 ]
 
 [project.scripts]

codex_telegram_bridge/readme.md

@@ -38,6 +38,12 @@ Run:
 uv run exec-bridge
 ```
 
+Optional flags:
+
+- `--progress-edit-every FLOAT` (default `2.5`)
+- `--progress-silent/--no-progress-silent` (default silent)
+- `--final-notify/--no-final-notify` (default notify via new message)
+
 ## Option 2: MCP server
 
 Run:

codex_telegram_bridge/src/codex_telegram_bridge/exec_bridge.py

@@ -14,6 +14,8 @@ import time
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any, Callable, Dict, Optional, Tuple
 
+import typer
+
 from .bridge_common import (
     TelegramClient,
     RouteStore,
@@ -260,7 +262,24 @@ class CodexExecRunner:
 # -------------------- Telegram loop --------------------
 
 
-def main() -> None:
+def run(
+    progress_edit_every_s: float = typer.Option(
+        2.5,
+        "--progress-edit-every",
+        help="Minimum seconds between progress message edits.",
+        min=0.1,
+    ),
+    progress_silent: bool = typer.Option(
+        True,
+        "--progress-silent/--no-progress-silent",
+        help="Send the progress message without sound/vibration.",
+    ),
+    final_notify: bool = typer.Option(
+        True,
+        "--final-notify/--no-final-notify",
+        help="Send the final response as a new message (not an edit).",
+    ),
+) -> None:
     config = load_telegram_config()
     token = config_get(config, "bot_token") or ""
     db_path = config_get(config, "bridge_db") or "./bridge_routes.sqlite3"
@@ -326,12 +345,9 @@ def main() -> None:
             "[handle] start "
             f"chat_id={chat_id} user_msg_id={user_msg_id} resume_session={resume_session!r}"
         )
-        try:
+        edit_every_s = progress_edit_every_s
-            edit_every_s = float(os.environ.get("TG_PROGRESS_EDIT_EVERY_S", "2.5"))
+        silent_progress = progress_silent
-        except ValueError:
+        loud_final = final_notify
-            edit_every_s = 2.5
-        silent_progress = os.environ.get("TG_PROGRESS_SILENT", "1") == "1"
-        loud_final = os.environ.get("TG_FINAL_NOTIFY", "1") == "1"
 
         typing_stop = threading.Event()
         typing_thread = threading.Thread(
@@ -527,5 +543,9 @@ def main() -> None:
             pool.submit(handle, chat_id, user_msg_id, text, resume_session)
 
 
+def main() -> None:
+    typer.run(run)
+
+
 if __name__ == "__main__":
     main()

codex_telegram_bridge/src/codex_telegram_bridge/mcp_bridge.py

@@ -13,6 +13,8 @@ import time
 from queue import Queue, Empty
 from typing import Any, Dict, List, Optional, Tuple
 
+import typer
+
 from .bridge_common import (
     TelegramClient,
     RouteStore,
@@ -253,7 +255,7 @@ class MCPStdioClient:
             pass
 
 
-def main() -> None:
+def run() -> None:
     config = load_telegram_config()
     token = config_get(config, "bot_token") or ""
     db_path = config_get(config, "bridge_db") or "./bridge_routes.sqlite3"
@@ -369,5 +371,9 @@ def main() -> None:
             work_q.put((chat_id, user_msg_id, prompt, conversation_id))
 
 
+def main() -> None:
+    typer.run(run)
+
+
 if __name__ == "__main__":
     main()

codex_telegram_bridge/src/codex_telegram_bridge/tmux_notify.py

@@ -5,50 +5,76 @@
 # ///
 from __future__ import annotations
 
-import argparse
 import sys
 from typing import Optional
 
+import typer
+
 from .bridge_common import TelegramClient, RouteStore, config_get, load_telegram_config
 
 
-def main() -> None:
+def run(
+    chat_id: Optional[int] = typer.Option(
+        None,
+        "--chat-id",
+        help="Telegram chat id (defaults to chat_id in ~/.codex/telegram.toml).",
+    ),
+    tmux_target: str = typer.Option(
+        ...,
+        "--tmux-target",
+        help='tmux target, e.g. "codex1:0.0" or "codex1"',
+    ),
+    db: Optional[str] = typer.Option(
+        None,
+        "--db",
+        help="Path to the routing database.",
+    ),
+    reply_to: Optional[int] = typer.Option(
+        None,
+        "--reply-to",
+        help="Optional Telegram message_id to reply to.",
+    ),
+    text: Optional[str] = typer.Option(
+        None,
+        "--text",
+        help="Message text. If omitted, read stdin.",
+    ),
+) -> None:
     config = load_telegram_config()
     default_chat_id = config_get(config, "chat_id")
     if isinstance(default_chat_id, str):
         default_chat_id = int(default_chat_id) if default_chat_id.strip() else None
     elif not isinstance(default_chat_id, int):
         default_chat_id = None
-
+    if chat_id is None:
-    ap = argparse.ArgumentParser()
+        chat_id = default_chat_id
-    ap.add_argument("--chat-id", type=int, default=default_chat_id, required=default_chat_id is None)
+    if chat_id is None:
-    ap.add_argument("--tmux-target", type=str, required=True, help='tmux target, e.g. "codex1:0.0" or "codex1"')
+        raise typer.BadParameter(
-    ap.add_argument(
+            "chat_id is required (pass --chat-id or set chat_id in ~/.codex/telegram.toml)."
-        "--db",
-        type=str,
-        default=config_get(config, "bridge_db") or "./bridge_routes.sqlite3",
         )
-    ap.add_argument("--reply-to", type=int, default=None, help="Optional Telegram message_id to reply to")
+    if db is None:
-    ap.add_argument("--text", type=str, default=None, help="Message text. If omitted, read stdin.")
+        db = config_get(config, "bridge_db") or "./bridge_routes.sqlite3"
-    args = ap.parse_args()
 
     token = config_get(config, "bot_token") or ""
     bot = TelegramClient(token)
-    store = RouteStore(args.db)
+    store = RouteStore(db)
 
-    text = args.text
     if text is None:
         text = sys.stdin.read()
 
     sent = bot.send_message_markdown_chunked(
-        chat_id=args.chat_id,
+        chat_id=chat_id,
         text=text,
-        reply_to_message_id=args.reply_to,
+        reply_to_message_id=reply_to,
     )
 
     # Store mapping for every chunk so user can reply to any chunk.
     for m in sent:
-        store.link(args.chat_id, m["message_id"], "tmux", args.tmux_target, meta={})
+        store.link(chat_id, m["message_id"], "tmux", tmux_target, meta={})
+
+
+def main() -> None:
+    typer.run(run)
 
 
 if __name__ == "__main__":

codex_telegram_bridge/src/codex_telegram_bridge/tmux_reply_bot.py

@@ -9,6 +9,8 @@ import subprocess
 import time
 from typing import Optional
 
+import typer
+
 from .bridge_common import (
     TelegramClient,
     RouteStore,
@@ -30,7 +32,7 @@ def tmux_send_text(target: str, text: str, press_enter: bool = True) -> None:
         subprocess.check_call(["tmux", "send-keys", "-t", target, "Enter"])
 
 
-def main() -> None:
+def run() -> None:
     config = load_telegram_config()
     token = config_get(config, "bot_token") or ""
     db_path = config_get(config, "bridge_db") or "./bridge_routes.sqlite3"
@@ -100,5 +102,9 @@ def main() -> None:
                 )
 
 
+def main() -> None:
+    typer.run(run)
+
+
 if __name__ == "__main__":
     main()