@@ -29,6 +29,16 @@ TELEGRAM_SESSION_NAME=telegram_session
 # TELEGRAM_SESSION_STRING for the default account.
 # TELEGRAM_SESSION_STRINGS=<session A> <session B> <session C>
 
+# --- Session lock (optional) ---
+# Before connecting, each server process takes a per-session lock so a second
+# instance (e.g. a connector restart) never connects the same session while the
+# first is live. exclusive (default): refuse to start if another instance still
+# holds the session after TELEGRAM_LOCK_GRACE_SECONDS (default 20). shared: let
+# several instances on this host use one session -- safe only when they all
+# reach Telegram from the same IP; if unsure, use the pool above instead.
+# TELEGRAM_SESSION_LOCK=exclusive
+# TELEGRAM_LOCK_GRACE_SECONDS=20
+
 # --- Device identity (optional) ---
 # Controls how this client appears in Telegram > Settings > Devices. If unset,
 # Telethon falls back to the host platform (e.g. "arm64"). Set these so the
 
@@ -379,6 +379,25 @@ rather than reusing a session another client holds — reuse would make Telegram
 permanently invalidate that session for both clients.
 - "Send this from my work account to @example"
 
+### Sharing one session from one host
+
+Before connecting, every server process takes a per-session lock (see the
+`AuthKeyDuplicatedError` entry under Troubleshooting). By default it is
+exclusive: a second instance for the same session waits briefly for the first
+to exit and otherwise refuses to start. That lock can only see processes on the
+same host, and Telegram's rule is about IPs, not processes: instances on one
+host only collide when they reach Telegram from different IPs (dual-stack or
+split-tunnel VPN hosts). When they all share one egress IP — a laptop running
+several MCP clients — you can let them share the session instead of pooling:
+
+```env
+TELEGRAM_SESSION_LOCK=shared
+```
+
+Shared instances coexist with each other but never overlap an exclusive one
+(except on Windows, where a shared instance takes no lock). If you are not sure
+every client egresses from the same IP, use the pool.
+
 ## Device Identity
 
 These optional variables control how the client appears in Telegram under
@@ -645,7 +664,7 @@ Telegram messages, display names, chat titles, and button labels are untrusted c
   interactive phone-code login over stdio.
 - **Invalid API credentials:** verify `TELEGRAM_API_ID` and `TELEGRAM_API_HASH` at [my.telegram.org/apps](https://my.telegram.org/apps).
 - **Database is locked:** prefer string sessions, or make sure no other process is using the same file session.
-- **`AuthKeyDuplicatedError` / "Another telegram-mcp process is already connected with this session":** two processes tried to connect the same Telegram session at once (e.g. an MCP client restarted the connector before the old process exited), which Telegram rejects and can invalidate the session for both. The server now takes an exclusive lock per session before connecting; a second concurrent launch waits briefly (default 20s, override with `TELEGRAM_LOCK_GRACE_SECONDS`) for the first to release it and otherwise exits without ever calling `connect()`, instead of racing into a duplicate connection. Retry once only one instance is running.
+- **`AuthKeyDuplicatedError` / "Another telegram-mcp process is already connected with this session":** two processes tried to connect the same Telegram session at once (e.g. an MCP client restarted the connector before the old process exited), which Telegram rejects and can invalidate the session for both. The server now takes an exclusive lock per session before connecting; a second concurrent launch waits briefly (default 20s, override with `TELEGRAM_LOCK_GRACE_SECONDS`) for the first to release it and otherwise exits without ever calling `connect()`, instead of racing into a duplicate connection. Retry once only one instance is running — the refusal names the PID holding the lock. If several instances on this host are meant to share one session (all reaching Telegram from the same IP), set `TELEGRAM_SESSION_LOCK=shared`; see [Sharing one session from one host](#sharing-one-session-from-one-host).
 - **File tools are disabled:** pass allowed roots or configure MCP Roots in your client.
 - **Path rejected:** ensure the path is inside an allowed root and does not use traversal or wildcard patterns.
 - **Auth errors after password changes:** regenerate your session string.
 
@@ -35,13 +35,37 @@ def _lock_grace_seconds() -> float:
         return DEFAULT_GRACE_SECONDS
 
 
+_SESSION_LOCK_MODES = ("exclusive", "shared")
+
+
+def _session_lock_shared() -> bool:
+    """``TELEGRAM_SESSION_LOCK``: exclusive (default) | shared.
+
+    ``exclusive`` refuses to start while another instance on this host holds
+    the same session. ``shared`` takes the lock in shared mode instead, so any
+    number of instances on this host can use one session -- safe only when
+    they all reach Telegram from the same IP, since Telegram invalidates a
+    session it sees from two IPs at once. Shared and exclusive instances never
+    overlap, so the default keeps its guarantee.
+    """
+    raw = (os.getenv("TELEGRAM_SESSION_LOCK") or "exclusive").strip().lower()
+    if raw not in _SESSION_LOCK_MODES:
+        accepted = ", ".join(_SESSION_LOCK_MODES)
+        raise SystemExit(f"Invalid TELEGRAM_SESSION_LOCK '{raw}'. Expected one of: {accepted}.")
+    return raw == "shared"
+
+
 async def _connect_authorized_client(label, client) -> None:
-    # First, prevent our own duplicate-spawn case outright: an exclusive
-    # per-session lock means a second instance of this server never even
-    # attempts to connect while another instance already holds the same
-    # session (see telegram_mcp/singleton.py for why and how).
+    # First, prevent our own duplicate-spawn case outright: a per-session lock
+    # means a second instance of this server never even attempts to connect
+    # while another instance already holds the same session (see
+    # telegram_mcp/singleton.py for why and how). The lock only sees processes
+    # on this host; TELEGRAM_SESSION_LOCK=shared lets those share one session
+    # where they all reach Telegram from a single IP.
     lock = SessionLock(label, session_identity(client))
-    await asyncio.to_thread(lock.acquire, grace_seconds=_lock_grace_seconds())
+    await asyncio.to_thread(
+        lock.acquire, grace_seconds=_lock_grace_seconds(), shared=_session_lock_shared()
+    )
     _session_locks[label] = lock
 
     # Once we hold the lock, still tolerate a transient AuthKeyDuplicatedError
@@ -173,7 +197,10 @@ async def _warm_caches() -> None:
                 "session (e.g. the client restarted the connector without the old "
                 "process exiting yet). This instance is exiting instead of "
                 "connecting a second time, which would risk Telegram invalidating "
-                "the session for both. Retry once the other instance is gone.",
+                "the session for both. Retry once the other instance is gone, or "
+                "set TELEGRAM_SESSION_LOCK=shared if several instances on this "
+                "host are meant to share one session (safe when they all reach "
+                "Telegram from the same IP).",
                 file=sys.stderr,
             )
         sys.exit(1)
@@ -193,6 +220,7 @@ def main() -> None:
     _configure_allowed_roots_from_cli(sys.argv[1:])
     _runtime._apply_exposed_tools_mode()
     _transcription.validate_transcription_config()
+    _session_lock_shared()  # fail loudly at startup on a bad toggle
     asyncio.run(_main())
 
 
 
@@ -20,6 +20,16 @@
 "restart replaces the old process" case) and, if the lock is still held once
 the grace period elapses, raises :class:`SessionLockError` instead of ever
 calling ``connect()`` -- so the live session is never disturbed.
+
+The lock can only ever see processes on this host, and those only collide
+at Telegram when they reach it from different IPs (dual-stack or
+split-tunnel hosts). Where every instance shares one egress IP,
+``acquire(shared=True)`` takes the lock in shared mode instead: shared
+holders coexist with each other, but a shared holder and an exclusive
+holder never overlap, so an operator who never opted in keeps the
+exclusive guarantee. (On Windows, ``msvcrt`` byte-range locks have no
+shared mode; a shared holder there takes no lock at all.) The exclusive
+holder leaves its PID in the lock file so a refused contender can name it.
 """
 
 from __future__ import annotations
@@ -34,7 +44,9 @@
 if os.name == "nt":
     import msvcrt
 
-    def _try_lock(fh: IO) -> bool:
+    def _try_lock(fh: IO, shared: bool = False) -> bool:
+        if shared:
+            return True  # no shared byte-range locks on Windows; see module docstring
         try:
             fh.seek(0)
             msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
@@ -52,9 +64,10 @@ def _unlock(fh: IO) -> None:
 else:
     import fcntl
 
-    def _try_lock(fh: IO) -> bool:
+    def _try_lock(fh: IO, shared: bool = False) -> bool:
         try:
-            fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            mode = fcntl.LOCK_SH if shared else fcntl.LOCK_EX
+            fcntl.flock(fh.fileno(), mode | fcntl.LOCK_NB)
             return True
         except OSError:
             return False
@@ -97,35 +110,76 @@ def acquire(
         *,
         grace_seconds: float = DEFAULT_GRACE_SECONDS,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
+        shared: bool = False,
     ) -> None:
         """Block (up to ``grace_seconds``) until the lock is free, then take it.
 
-        Raises :class:`SessionLockError` if another live process still holds
-        the lock once the grace period elapses.
+        ``shared`` takes the lock in shared mode: any number of shared holders
+        coexist, but a shared holder and an exclusive holder never overlap.
+        Raises :class:`SessionLockError` if the lock is still held once the
+        grace period elapses.
         """
         fh = open(self.path, "a+")
         deadline = time.monotonic() + grace_seconds
         while True:
-            if _try_lock(fh):
+            if _try_lock(fh, shared=shared):
                 self._fh = fh
+                self._record_holder(shared)
                 return
             if time.monotonic() >= deadline:
                 fh.close()
+                holder = self.holder_pid()
+                where = (
+                    f"lock {self.path} held by PID {holder}"
+                    if holder
+                    else f"lock held: {self.path}"
+                )
                 raise SessionLockError(
                     "Another telegram-mcp process is already connected with this "
-                    f"session (lock held: {self.path}). Refusing to connect a "
-                    "second time to avoid Telegram's AuthKeyDuplicatedError. If "
-                    "that other process already exited, this lock will clear on "
-                    "its own -- retry."
+                    f"session ({where}). Refusing to connect a second time to "
+                    "avoid Telegram's AuthKeyDuplicatedError. If that other "
+                    "process already exited, this lock will clear on its own -- "
+                    "retry."
                 )
             time.sleep(poll_interval)
 
     def release(self) -> None:
         if self._fh is not None:
+            self._clear_holder()
             _unlock(self._fh)
             self._fh.close()
             self._fh = None
 
+    def holder_pid(self) -> Optional[int]:
+        """PID left in the lock file by the current exclusive holder, if any.
+
+        Best effort: Windows refuses to read a locked byte range, and a shared
+        holder records nothing (there may be several).
+        """
+        try:
+            text = self.path.read_text().strip()
+        except OSError:
+            return None
+        return int(text) if text.isdigit() else None
+
+    def _record_holder(self, shared: bool) -> None:
+        # Exclusive holders leave their PID for a refused contender to report;
+        # shared holders wipe whatever a previous exclusive holder left behind.
+        self._clear_holder()
+        if not shared:
+            try:
+                self._fh.write(str(os.getpid()))
+                self._fh.flush()
+            except OSError:
+                pass
+
+    def _clear_holder(self) -> None:
+        try:
+            self._fh.seek(0)
+            self._fh.truncate()
+        except OSError:
+            pass
+
 
 def session_identity(client: object) -> str:
     """Derive a stable identity string for a ``TelegramClient``'s session.
 
@@ -1,3 +1,5 @@
+import os
+
 import pytest
 
 from telegram_mcp import runner
@@ -43,6 +45,9 @@ def _init_with_tmp_dir(self, label, session_identity, *, lock_dir=tmp_path):
 
     monkeypatch.setattr(singleton_module.SessionLock, "__init__", _init_with_tmp_dir)
     monkeypatch.setattr(runner, "_lock_grace_seconds", lambda: 0.01)
+    # load_dotenv() may have pulled TELEGRAM_SESSION_LOCK from the developer's
+    # .env; tests assume the exclusive default unless they set it themselves.
+    monkeypatch.delenv("TELEGRAM_SESSION_LOCK", raising=False)
     yield
     runner._session_locks.clear()
 
@@ -96,6 +101,78 @@ async def test_connect_authorized_client_allows_different_sessions_concurrently(
     assert second.connected is True
 
 
+@pytest.mark.asyncio
+async def test_shared_lock_mode_lets_instances_share_a_session(monkeypatch):
+    monkeypatch.setenv("TELEGRAM_SESSION_LOCK", "shared")
+    first = _FakeClient(authorized=True, identity="shared-session")
+    second = _FakeClient(authorized=True, identity="shared-session")
+
+    await runner._connect_authorized_client("default", first)
+    first_lock = runner._session_locks["default"]
+    await runner._connect_authorized_client("default", second)
+
+    assert first.connected is True
+    assert second.connected is True
+
+    first_lock.release()
+    runner._session_locks["default"].release()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "first_mode, second_mode", [("exclusive", "shared"), ("shared", "exclusive")]
+)
+async def test_shared_and_exclusive_instances_never_overlap(monkeypatch, first_mode, second_mode):
+    first = _FakeClient(authorized=True, identity="shared-session")
+    second = _FakeClient(authorized=True, identity="shared-session")
+
+    monkeypatch.setenv("TELEGRAM_SESSION_LOCK", first_mode)
+    await runner._connect_authorized_client("default", first)
+    first_lock = runner._session_locks["default"]
+
+    monkeypatch.setenv("TELEGRAM_SESSION_LOCK", second_mode)
+    with pytest.raises(runner.SessionLockError, match="already connected"):
+        await runner._connect_authorized_client("default", second)
+
+    assert second.connected is False
+    first_lock.release()
+
+
+@pytest.mark.asyncio
+async def test_lock_error_names_the_exclusive_holder():
+    first = _FakeClient(authorized=True, identity="shared-session")
+    second = _FakeClient(authorized=True, identity="shared-session")
+
+    await runner._connect_authorized_client("default", first)
+    lock = runner._session_locks["default"]
+
+    with pytest.raises(runner.SessionLockError, match=f"held by PID {os.getpid()}"):
+        await runner._connect_authorized_client("default", second)
+
+    lock.release()
+    assert lock.path.read_text() == ""  # a released lock names nobody
+
+
+@pytest.mark.parametrize(
+    "value, shared",
+    [(None, False), ("", False), ("exclusive", False), ("shared", True), (" Shared ", True)],
+)
+def test_session_lock_mode_parsing(monkeypatch, value, shared):
+    if value is None:
+        monkeypatch.delenv("TELEGRAM_SESSION_LOCK", raising=False)
+    else:
+        monkeypatch.setenv("TELEGRAM_SESSION_LOCK", value)
+
+    assert runner._session_lock_shared() is shared
+
+
+def test_session_lock_mode_rejects_unknown_values(monkeypatch):
+    monkeypatch.setenv("TELEGRAM_SESSION_LOCK", "sometimes")
+
+    with pytest.raises(SystemExit, match="Invalid TELEGRAM_SESSION_LOCK 'sometimes'"):
+        runner._session_lock_shared()
+
+
 class _FakeSettings:
     def __init__(self):
         self.host = None