Merge pull request modelcontextprotocol#22 from jkoelker/jk/auth

fix(auth): align max token age with reauth warning

Author: jkoelker

pyproject.toml

@@ -28,7 +28,7 @@ build-backend = "hatchling.build"
 [dependency-groups]
 dev = [
     "mcp[cli]>=1.17.0",
-    "pyright>=1.1.406",
+    "pyright>=1.1.407",
     "pytest>=8.4.2",
     "ruff>=0.14.0",
 ]

src/schwab_mcp/auth.py

@@ -12,6 +12,9 @@
 
 from schwab_mcp import tokens
 
+
+DEFAULT_MAX_TOKEN_AGE_SECONDS = 5 * 24 * 60 * 60
+
 if TYPE_CHECKING:
     from multiprocessing import Process as ProcessType, Queue as QueueType
 else:  # pragma: no cover - runtime fallback for multiprocess
@@ -29,7 +32,7 @@ def easy_client(
     token_manager: tokens.Manager,
     asyncio: bool = False,
     enforce_enums: bool = True,
-    max_token_age: int | None = 60 * 60 * 24 * 13 // 2,
+    max_token_age: int | None = DEFAULT_MAX_TOKEN_AGE_SECONDS,
     callback_timeout: float = 300.0,
     interactive: bool = True,
     requested_browser: str | None = None,

src/schwab_mcp/cli.py

@@ -15,6 +15,7 @@
 
 
 APP_NAME = "schwab-mcp"
+TOKEN_MAX_AGE_SECONDS = schwab_auth.DEFAULT_MAX_TOKEN_AGE_SECONDS
 
 
 @click.group()
@@ -68,6 +69,7 @@ def auth(
             client_secret=client_secret,
             callback_url=callback_url,
             token_manager=token_manager,
+            max_token_age=TOKEN_MAX_AGE_SECONDS,
         )
 
         # If we get here, the authentication was successful
@@ -162,6 +164,7 @@ def server(
             asyncio=True,
             interactive=False,
             enforce_enums=False,
+            max_token_age=TOKEN_MAX_AGE_SECONDS,
         )
 
         if not isinstance(client, AsyncClient):
@@ -180,7 +183,7 @@ def server(
         return 1
 
     # Check token age
-    if client.token_age() > 5 * 86400:
+    if client.token_age() >= TOKEN_MAX_AGE_SECONDS:
         send_error_response(
             "Token is older than 5 days. Please run 'schwab-mcp auth' to re-authenticate.",
             code=401,

tests/test_cli_auth.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from typing import Any
+
+from click.testing import CliRunner
+
+from schwab_mcp import cli
+
+
+def test_auth_command_uses_max_token_age(monkeypatch, tmp_path):
+    captured: dict[str, Any] = {}
+
+    class DummyManager:
+        def __init__(self, path: str) -> None:
+            self.path = path
+            captured["token_path"] = path
+
+    def fake_easy_client(**kwargs):
+        captured["easy_client_kwargs"] = kwargs
+        return object()
+
+    monkeypatch.setattr(cli.tokens, "Manager", DummyManager)
+    monkeypatch.setattr(cli.schwab_auth, "easy_client", fake_easy_client)
+
+    runner = CliRunner()
+    token_file = tmp_path / "token.yaml"
+    result = runner.invoke(
+        cli.cli,
+        [
+            "auth",
+            "--token-path",
+            str(token_file),
+            "--client-id",
+            "cid",
+            "--client-secret",
+            "secret",
+        ],
+        catch_exceptions=False,
+    )
+
+    assert result.exit_code == 0
+    assert captured["token_path"] == str(token_file)
+    assert captured["easy_client_kwargs"]["max_token_age"] == cli.TOKEN_MAX_AGE_SECONDS

tests/test_cli_server_write_modes.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from click.testing import CliRunner
+from typing import Any
 
 from schwab_mcp import cli
 from schwab_mcp.approvals import ApprovalDecision, ApprovalManager, ApprovalRequest, NoOpApprovalManager
@@ -34,11 +35,12 @@ def authorized_user_ids(users):
         return frozenset(int(value) for value in users)
 
 
-def _patch_common(monkeypatch, captured):
+def _patch_common(monkeypatch, captured: dict[str, Any]) -> None:
     monkeypatch.setattr(cli, "AsyncClient", FakeAsyncClient)
 
     def fake_easy_client(**_kwargs):
         captured["easy_client_called"] = True
+        captured["easy_client_kwargs"] = _kwargs
         return FakeAsyncClient()
 
     monkeypatch.setattr(cli.schwab_auth, "easy_client", fake_easy_client)
@@ -58,7 +60,7 @@ async def run(self):
 
 
 def test_server_defaults_to_read_only(monkeypatch):
-    captured: dict[str, object] = {}
+    captured: dict[str, Any] = {}
     _patch_common(monkeypatch, captured)
 
     runner = CliRunner()
@@ -77,10 +79,11 @@ def test_server_defaults_to_read_only(monkeypatch):
     assert result.exit_code == 0
     assert captured["allow_write"] is False
     assert isinstance(captured["approval_manager"], NoOpApprovalManager)
+    assert captured["easy_client_kwargs"]["max_token_age"] == cli.TOKEN_MAX_AGE_SECONDS
 
 
 def test_server_enables_write_mode_when_flag_set(monkeypatch):
-    captured: dict[str, object] = {}
+    captured: dict[str, Any] = {}
     _patch_common(monkeypatch, captured)
 
     runner = CliRunner()
@@ -100,10 +103,11 @@ def test_server_enables_write_mode_when_flag_set(monkeypatch):
     assert result.exit_code == 0
     assert captured["allow_write"] is True
     assert isinstance(captured["approval_manager"], NoOpApprovalManager)
+    assert captured["easy_client_kwargs"]["max_token_age"] == cli.TOKEN_MAX_AGE_SECONDS
 
 
 def test_server_enables_write_mode_with_discord(monkeypatch):
-    captured: dict[str, object] = {}
+    captured: dict[str, Any] = {}
     _patch_common(monkeypatch, captured)
     monkeypatch.setattr(cli, "DiscordApprovalManager", DummyDiscordApprovalManager)
 
@@ -129,3 +133,4 @@ def test_server_enables_write_mode_with_discord(monkeypatch):
     assert result.exit_code == 0
     assert captured["allow_write"] is True
     assert isinstance(captured["approval_manager"], DummyDiscordApprovalManager)
+    assert captured["easy_client_kwargs"]["max_token_age"] == cli.TOKEN_MAX_AGE_SECONDS