mcp

COVERAGE_GUIDE.md

@@ -1,290 +1,30 @@
-# 测试覆盖率工具使用指南
+# MCP Gateway Coverage Guide
 
-## 快速开始
+Use this guide to run coverage for the current direct Gateway session implementation.
 
-### 1. 安装 coverage
+## Run Coverage
 
 ```powershell
-pip install coverage
-```
-
-### 2. 运行覆盖率测试
-
-最简单的方式：
-
-```powershell
-# 运行测试并生成所有报告
 python run_coverage.py
-
-# 运行并自动在浏览器中打开 HTML 报告
 python run_coverage.py --open
 ```
 
-## 手动使用 coverage 命令
-
-### 基本命令
+Manual form:
 
 ```powershell
-# 1. 运行测试并收集覆盖率数据
 python -m coverage run -m unittest discover -s tests -p "test_*.py"
-
-# 2. 显示控制台报告
 python -m coverage report
-
-# 3. 生成 HTML 报告
 python -m coverage html
-
-# 4. 生成 XML 报告（用于 CI/CD）
-python -m coverage xml
-```
-
-### 高级命令
-
-```powershell
-# 显示缺失的行号
-python -m coverage report -m
-
-# 只显示特定文件的覆盖率
-python -m coverage report services/gateway_session_store.py
-
-# 设置覆盖率阈值（低于 80% 会失败）
-python -m coverage report --fail-under=80
-
-# 显示分支覆盖率详情
-python -m coverage report --show-missing
-```
-
-## 查看报告
-
-### 1. 控制台报告
-
-运行 `python -m coverage report` 后直接在终端显示：
-
-```
-Name                                Stmts   Miss Branch BrPart   Cover
-----------------------------------------------------------------------
-services/token_store.py               158     63     50     10  52.40%
-app.py                                129     25     48     16  74.58%
-tools/bind_auth_code.py                18      3      6      3  75.00%
-mcp_protocol.py                       147     20     66     18  80.28%
-...
-----------------------------------------------------------------------
-TOTAL                                 909    132    294     63  80.80%
-```
-
-**指标说明**：
-- **Stmts**: 可执行语句总数
-- **Miss**: 未执行的语句数
-- **Branch**: 分支总数（if/else）
-- **BrPart**: 部分覆盖的分支数
-- **Cover**: 覆盖率百分比
-
-### 2. HTML 报告（推荐）
-
-生成后打开 `htmlcov/index.html`：
-
-```powershell
-# 自动打开
-python run_coverage.py --open
-
-# 或手动打开
-start htmlcov/index.html
-```
-
-**HTML 报告特性**：
-- 📊 可视化的覆盖率统计
-- 📝 逐行查看哪些代码被执行
-- 🎨 颜色编码（绿色=覆盖，红色=未覆盖）
-- 🔍 点击文件查看详细信息
-- 📈 分支覆盖率可视化
-
-### 3. XML 报告
-
-用于 CI/CD 集成（如 GitHub Actions、Jenkins）：
-
-```powershell
-python -m coverage xml
-# 生成 coverage.xml
-```
-
-## 配置文件
-
-项目使用 `.coveragerc` 配置覆盖率工具：
-
-```ini
-[run]
-source = .
-omit = tests/*, */__pycache__/*, run_coverage.py
-branch = True
-
-[report]
-precision = 2
-exclude_lines = pragma: no cover
-
-[html]
-directory = htmlcov
-```
-
-**关键配置**：
-- `source`: 要测量的源代码目录
-- `omit`: 排除的文件/目录
-- `branch`: 启用分支覆盖率
-- `exclude_lines`: 排除带有特定注释的行
-
-## 提高覆盖率的技巧
-
-### 1. 识别未覆盖的代码
-
-查看 HTML 报告中的红色行：
-
-```powershell
-python run_coverage.py --open
-```
-
-### 2. 编写针对性测试
-
-对于未覆盖的代码，编写测试：
-
-```python
-def test_error_handling(self):
-    """测试错误处理分支"""
-    with self.assertRaises(ValueError):
-        function_that_raises_error()
-```
-
-### 3. 测试边界条件
-
-```python
-def test_empty_input(self):
-    result = process([])
-    self.assertEqual(result, [])
-
-def test_none_input(self):
-    with self.assertRaises(ValueError):
-        process(None)
-```
-
-### 4. 排除不可测试的代码
-
-使用 `# pragma: no cover` 注释：
-
-```python
-if __name__ == '__main__':  # pragma: no cover
-    main()
-```
-
-## CI/CD 集成
-
-### GitHub Actions
-
-```yaml
-name: Tests with Coverage
-
-on: [push, pull_request]
-
-jobs:
-  test:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: '3.11'
-      
-      - name: Install dependencies
-        run: |
-          pip install -r requirements.txt
-          pip install coverage
-      
-      - name: Run tests with coverage
-        run: |
-          python -m coverage run -m unittest discover -s tests -p "test_*.py"
-          python -m coverage report --fail-under=80
-          python -m coverage xml
-      
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
-        with:
-          file: ./coverage.xml
-```
-
-### GitLab CI
-
-```yaml
-test:
-  script:
-    - pip install coverage
-    - python -m coverage run -m unittest discover -s tests -p "test_*.py"
-    - python -m coverage report --fail-under=80
-    - python -m coverage xml
-  coverage: '/TOTAL.*\s+(\d+%)$/'
-  artifacts:
-    reports:
-      coverage_report:
-        coverage_format: cobertura
-        path: coverage.xml
-```
-
-## 常见问题
-
-### Q: 为什么覆盖率突然下降？
-
-A: 可能原因：
-1. 新增了未测试的代码
-2. 删除了部分测试
-3. 配置文件更改导致测量范围变化
-
-解决：运行 `python -m coverage report -m` 查看缺失的行号。
-
-### Q: 如何只测试某个模块的覆盖率？
-
-A: 使用 `--source` 参数：
-
-```powershell
-python -m coverage run --source=services/gateway_session_store.py -m unittest tests.test_gateway_session_store_unit
-python -m coverage report
 ```
 
-### Q: 覆盖率 100% 就够了吗？
-
-A: 不一定。覆盖率只表示代码被执行了，不表示逻辑正确。还需要：
-- 测试边界条件
-- 测试错误处理
-- 测试并发场景
-- 集成测试和端到端测试
-
-### Q: 目标覆盖率应该设多少？
-
-A: 建议：
-- **核心业务逻辑**: 90%+
-- **工具类/辅助函数**: 80%+
-- **整体项目**: 70-80%
-- **新增代码**: 要求 90%+
-
-## 当前项目覆盖率
-
-截至 2026-07-08：
-
-| 类别 | 覆盖率 | 状态 |
-|------|--------|------|
-| 新增核心模块 | 95-100% | ✅ 优秀 |
-| 整体项目 | 80.80% | ✅ 良好 |
-| 目标 | 80%+ | ✅ 已达标 |
+## Current Scope
 
-**8 个新增核心模块达到 100% 覆盖率**：
-- constants.py
-- services/api_client.py
-- services/gateway_session_store.py
-- services/scoped_api_client.py
-- services/scoped_auth_client.py
-- tools/query_track.py
-- utils/rate_limiter.py
-- utils/security.py
+Coverage should focus on:
 
-## 参考资源
+- Public Gateway request handling.
+- Redis Gateway session storage.
+- Request-scoped API forwarding.
+- Token refresh/revoke compatibility.
+- Query tool registration and JSON-RPC behavior.
 
-- [Coverage.py 官方文档](https://coverage.readthedocs.io/)
-- [Python 测试最佳实践](https://docs.python-guide.org/writing/tests/)
-- [测试覆盖率的意义](https://martinfowler.com/bliki/TestCoverage.html)
+The retired authorization-code binding files and tools should not appear in the runtime coverage list.

README.md

@@ -1,4 +1,4 @@
-# Python MCP Gateway
+# Python MCP Gateway
 
 这是物流系统给 Workbuddy 使用的轻量 MCP Gateway。它负责接收 MCP 请求、管理员工授权会话，并把工具调用转发到现有 ThinkPHP 项目的 MCP 接口。
 
@@ -16,7 +16,7 @@ Gateway 保持“薄网关”边界：
 - 支持公网 HTTP JSON-RPC 模式：`serve-public`。
 - 支持 Redis token/session 存储。
 - 支持文件 token store 作为开发排障兜底。
-- 支持 `bind_auth_code` 授权绑定工具。
+- 不再支持授权码绑定工具；正式接入只使用后台生成的 `GWS_xxx` 设备配置。
 - 支持 `query_order` 订单查询工具。
 - 支持 `query_track` 轨迹查询工具。
 - 支持 MCP `initialize`、`tools/list`、`tools/call`。
@@ -50,14 +50,7 @@ Workbuddy 配置示例：
 
 不要把 `Y:\mcp\app.py` 写进员工 Workbuddy 配置；`Y:` 只是开发机映射盘视角，员工机器未必存在。
 
-本地模式下，员工流程为：
-
-1. 员工在后台获取 Workbuddy 授权码。
-2. Workbuddy 自动拉起 `serve-stdio`。
-3. 员工在 Workbuddy 中调用 `bind_auth_code`，输入授权码完成绑定。
-4. Gateway 调用 `/mcp/auth/exchange`，并带上本机生成的 `session_key`。
-5. token 存入 Redis 当前员工对应 key。
-6. 后续直接使用 `query_order`、`query_track` 等工具。
+本地 stdio 模式只保留给开发排障和已有 token 会话的兼容调用；员工正式使用路径统一走公网 HTTP 模式，由后台直接生成 Workbuddy 设备配置，不再通过授权码绑定。
 
 ### 公网 HTTP 模式
 
@@ -88,51 +81,32 @@ python app.py serve-public --host 0.0.0.0 --port 8765
 
 ## 公网 Gateway 会话生命周期
 
-### 1. 生成 gateway_session_id
-
-客户端首次使用时生成高熵会话 ID：
-
-```python
-import secrets
-
-gateway_session_id = 'GWS_' + secrets.token_urlsafe(32)
-```
-
-客户端需要保存这个 ID，并在后续每次请求中复用。
+### 1. 后台生成设备配置
 
-### 2. 绑定授权码
+员工在后台或 home 的“连接 Workbuddy”入口新增设备配置。base 会一次性完成：
 
-员工从后台复制授权码后，Workbuddy 调用 `bind_auth_code`：
+- 生成 `GWS_xxx`。
+- 写入 `st_mcp_gateway_session`。
+- 创建 `st_mcp_token_session`。
+- 写入 Redis Gateway session。
+- 返回只包含 `GWS_xxx` 的 Workbuddy 配置。
 
-```http
-POST /mcp
-X-Gateway-Session: GWS_xxx
-Content-Type: application/json
-```
+Workbuddy 配置示例：
 
 ```json
 {
-  "jsonrpc": "2.0",
-  "id": 1,
-  "method": "tools/call",
-  "params": {
-    "name": "bind_auth_code",
-    "arguments": {
-      "auth_code": "AC_xxx"
+  "mcpServers": {
+    "fms-public": {
+      "url": "http://192.168.1.241:8765/mcp",
+      "headers": {
+        "X-Gateway-Session": "GWS_xxx"
+      }
     }
   }
 }
 ```
 
-Gateway 会调用 base 项目的 `/mcp/auth/exchange`，并把返回的 `mcp_token` 存入 Redis：
-
-```text
-fms:mcp:gateway:session:{sha256(gateway_session_id)}
-```
-
-Redis value 中保存 `mcp_token`、`admin_id`、`company_id`、`client_type`、`expire_time` 等信息。Redis key 不保存明文 `gateway_session_id`。
-
-### 3. 调用工具
+### 2. 调用工具
 
 后续工具调用必须继续带同一个 `gateway_session_id`：
 
@@ -154,7 +128,7 @@ Redis value 中保存 `mcp_token`、`admin_id`、`company_id`、`client_type`、
 
 Gateway 根据 `gateway_session_id` 从 Redis 读取当前员工的 `mcp_token`，再把请求转发到 fmsoperate 的 `/mcp/tools/queryOrder`。
 
-### 4. 撤销与过期
+### 3. 撤销与过期
 
 - 员工在后台取消 Workbuddy 授权后，ThinkPHP 会让对应 `mcp_token` 失效。
 - 公网 Gateway Redis session 会在 TTL 到期后自动过期。
@@ -169,12 +143,6 @@ Gateway 根据 `gateway_session_id` 从 Redis 读取当前员工的 `mcp_token`
 python app.py list-tools
 ```
 
-开发/排障时手动绑定授权码：
-
-```powershell
-python app.py bind --auth-code AUTH123
-```
-
 直接调用订单查询：
 
 ```powershell

app.py

@@ -1,10 +1,10 @@
-import argparse
+import argparse
 import json
 import sys
 import uuid
 
 from config import GatewayConfig
-from constants import BIND_HINT
+from constants import DEVICE_INVALID_MESSAGE
 from mcp_protocol import McpProtocolHandler
 from public_gateway import PublicGatewayApp
 from public_server import serve_public
@@ -12,22 +12,17 @@ from services.api_client import ApiClient
 from services.auth_client import AuthClient
 from services.gateway_session_store import GatewaySessionStore
 from services.scoped_api_client import ScopedApiClient
-from services.scoped_auth_client import ScopedAuthClient
 from services.token_store import FileTokenStore, RedisSocketClient, RedisTokenStore
-from tools.bind_auth_code import BindAuthCodeTool
 from tools.query_order import QueryOrderTool
 from tools.query_track import QueryTrackTool
 
 
-
-
 class GatewayApp:
     def __init__(self, auth_client=None, api_client=None, token_store=None):
         self.auth_client = auth_client
         self.api_client = api_client
         self.token_store = token_store
         self._tools = {
-            'bind_auth_code': BindAuthCodeTool(auth_client=auth_client),
             'query_order': QueryOrderTool(api_client=api_client),
             'query_track': QueryTrackTool(api_client=api_client),
         }
@@ -73,14 +68,6 @@ class GatewayApp:
     def list_tools(self):
         return [tool.metadata() for tool in self._tools.values()]
 
-    def bind(self, auth_code):
-        if self.auth_client is None:
-            raise RuntimeError('auth client is required for bind')
-        auth_code = str(auth_code).strip()
-        if not auth_code:
-            raise ValueError('auth_code is required')
-        return self.auth_client.exchange(auth_code)
-
     def build_request_id(self, request_id=''):
         request_id = str(request_id or '').strip()
         if request_id:
@@ -92,7 +79,7 @@ class GatewayApp:
             return
         session = self.token_store.get()
         if not session or not session.get('token'):
-            raise RuntimeError(BIND_HINT)
+            raise RuntimeError(DEVICE_INVALID_MESSAGE)
         if self.token_store.is_expiring():
             if self.auth_client is None:
                 raise RuntimeError('mcp token expiring but auth client missing')
@@ -124,9 +111,6 @@ class GatewayApp:
         public_parser.add_argument('--host', default='0.0.0.0')
         public_parser.add_argument('--port', type=int, default=8765)
 
-        bind_parser = subparsers.add_parser('bind')
-        bind_parser.add_argument('--auth-code', required=True)
-
         call_parser = subparsers.add_parser('call')
         call_parser.add_argument('--tool', required=True)
         call_parser.add_argument('--keyword', default='')
@@ -160,16 +144,8 @@ class GatewayApp:
             public_app = PublicGatewayApp(
                 session_store=session_store,
                 api_client=ScopedApiClient(config.tools_base_url, timeout=config.timeout_seconds),
-                auth_client=ScopedAuthClient(
-                    config.auth_base_url,
-                    config.client_type,
-                    session_store,
-                    timeout=config.timeout_seconds,
-                ),
             )
             return serve_public(public_app, host=args.host, port=args.port)
-        elif args.command == 'bind':
-            payload = self.bind(args.auth_code)
         elif args.command == 'call':
             tool_args = {
                 'page': args.page,
@@ -212,5 +188,4 @@ def main(argv=None):
 
 
 if __name__ == '__main__':
-    raise SystemExit(main(sys.argv[1:]))
-
+    raise SystemExit(main(sys.argv[1:]))

config.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass
 import os
 
 
@@ -46,6 +46,10 @@ class GatewayConfig:
         )
         timeout_seconds = cls._resolve_timeout_seconds(dotenv_env, primary_env)
         session_key = cls._pick(dotenv_env, primary_env, 'MCP_SESSION_KEY', 'FMS_SESSION_KEY')
+        gateway_mode = (cls._pick(dotenv_env, primary_env, 'FMS_GATEWAY_MODE', 'MCP_GATEWAY_MODE') or 'local').lower()
+        redis_prefix = cls._pick(dotenv_env, primary_env, 'MCP_REDIS_PREFIX', 'FMS_REDIS_PREFIX')
+        if not redis_prefix:
+            redis_prefix = 'fms:mcp:gateway:' if gateway_mode == 'public' else 'fms:mcp:workbuddy:'
         return cls(
             auth_base_url=auth_base_url,
             tools_base_url=tools_base_url,
@@ -59,9 +63,9 @@ class GatewayConfig:
             redis_port=int(cls._pick(dotenv_env, primary_env, 'MCP_REDIS_PORT', 'FMS_REDIS_PORT') or '6379'),
             redis_db=int(cls._pick(dotenv_env, primary_env, 'MCP_REDIS_DB', 'FMS_REDIS_DB') or '0'),
             redis_password=cls._pick(dotenv_env, primary_env, 'MCP_REDIS_PASSWORD', 'FMS_REDIS_PASSWORD') or '',
-            redis_prefix=cls._pick(dotenv_env, primary_env, 'MCP_REDIS_PREFIX', 'FMS_REDIS_PREFIX') or 'fms:mcp:workbuddy:',
+            redis_prefix=redis_prefix,
             session_key=session_key or cls._build_default_session_key(primary_env),
-            gateway_mode=(cls._pick(dotenv_env, primary_env, 'FMS_GATEWAY_MODE', 'MCP_GATEWAY_MODE') or 'local').lower(),
+            gateway_mode=gateway_mode,
             gateway_session_ttl_seconds=int(cls._pick(dotenv_env, primary_env, 'FMS_GATEWAY_SESSION_TTL_SECONDS', 'MCP_GATEWAY_SESSION_TTL_SECONDS') or '2592000'),
         )
 

constants.py

@@ -1 +1 @@
-BIND_HINT = '\u8bf7\u5148\u767b\u5f55\u540e\u53f0\u83b7\u53d6 Workbuddy \u6388\u6743\u7801\uff0c\u7136\u540e\u5728 Workbuddy \u4e2d\u8f93\u5165\u201c\u7ed1\u5b9a\u6388\u6743\u7801 xxx\u201d\u5b8c\u6210\u7ed1\u5b9a\u3002'
+DEVICE_INVALID_MESSAGE = '这台设备的 Workbuddy 配置已失效，请重新生成配置。'

public_gateway.py

@@ -1,8 +1,7 @@
-import logging
+import logging
 import uuid
 
-from constants import BIND_HINT
-from tools.bind_auth_code import BindAuthCodeTool
+from constants import DEVICE_INVALID_MESSAGE
 from tools.query_order import QueryOrderTool
 from tools.query_track import QueryTrackTool
 from utils.security import hash_gateway_session_id
@@ -12,12 +11,10 @@ logger = logging.getLogger(__name__)
 
 
 class PublicGatewayApp:
-    def __init__(self, session_store, api_client, auth_client):
+    def __init__(self, session_store, api_client, auth_client=None):
         self.session_store = session_store
         self.api_client = api_client
-        self.auth_client = auth_client
         self._tools = {
-            'bind_auth_code': BindAuthCodeTool(auth_client=None),
             'query_order': QueryOrderTool(api_client=None),
             'query_track': QueryTrackTool(api_client=None),
         }
@@ -29,35 +26,13 @@ class PublicGatewayApp:
         request_id = str(request_id or '').strip()
         return request_id or 'rq_{0}'.format(uuid.uuid4().hex[:16])
 
-    def bind_auth_code(self, gateway_session_id, auth_code):
-        if self.auth_client is None:
-            raise RuntimeError('auth client is required')
-
-        session_hash = hash_gateway_session_id(gateway_session_id)[:12]
-        logger.info(f"[AUDIT] bind_auth_code: session_hash={session_hash}, auth_code_provided={bool(auth_code)}")
-
-        result = self.auth_client.exchange(gateway_session_id, auth_code)
-
-        if result.get('code') == 'MCP_0000':
-            session = self.session_store.get(gateway_session_id)
-            admin_id = session.get('admin_id') if session else None
-            company_id = session.get('company_id') if session else None
-            logger.info(f"[AUDIT] bind_success: session_hash={session_hash}, admin_id={admin_id}, company_id={company_id}")
-        else:
-            logger.warning(f"[AUDIT] bind_failed: session_hash={session_hash}, code={result.get('code')}, msg={result.get('msg')}")
-
-        return result
-
     def call_tool(self, gateway_session_id, name, arguments=None, request_id=''):
-        if name == 'bind_auth_code':
-            auth_code = (arguments or {}).get('auth_code', '')
-            return self.bind_auth_code(gateway_session_id, auth_code)
         if name not in self._tools:
             raise KeyError('tool not registered: {0}'.format(name))
 
         session = self.session_store.get(gateway_session_id)
         if not session or not session.get('mcp_token'):
-            raise RuntimeError(BIND_HINT)
+            raise RuntimeError(DEVICE_INVALID_MESSAGE)
 
         tool = self._tools[name]
         request_id = self.build_request_id(request_id)
@@ -75,11 +50,10 @@ class PublicGatewayApp:
                 payload=arguments or {},
                 request_id=request_id,
             )
+            if hasattr(self.session_store, 'touch_session'):
+                self.session_store.touch_session(gateway_session_id)
             logger.info(f"[AUDIT] tool_success: session_hash={session_hash}, tool={name}, request_id={request_id}, code={result.get('code')}")
             return result
         except Exception as e:
             logger.error(f"[AUDIT] tool_error: session_hash={session_hash}, tool={name}, request_id={request_id}, error={str(e)}")
-            raise
-
-
-
+            raise

public_server.py

@@ -1,8 +1,8 @@
-import json
+import json
 import logging
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
-from constants import BIND_HINT
+from constants import DEVICE_INVALID_MESSAGE
 from mcp_protocol import McpProtocolHandler
 from services.request_context import RequestContextParser
 from utils.rate_limiter import SimpleRateLimiter
@@ -53,7 +53,7 @@ class PublicMcpHttpHandler:
             if method == 'tools/call':
                 context = self.context_parser.parse(headers or {})
                 if not context.has_session():
-                    raise RuntimeError(BIND_HINT)
+                    raise RuntimeError(DEVICE_INVALID_MESSAGE)
                 params = message.get('params') or {}
                 result = self.gateway_app.call_tool(
                     context.gateway_session_id,

services/auth_client.py

@@ -10,19 +10,6 @@ class AuthClient:
         self.timeout = int(timeout)
         self.session_key = str(session_key or '').strip()
 
-    def exchange(self, auth_code):
-        payload = {'auth_code': auth_code, 'client_type': self.client_type}
-        if self.session_key:
-            payload['session_key'] = self.session_key
-        response = self.transport.post_json(
-            self.base_url + '/mcp/auth/exchange',
-            payload,
-            {},
-            self.timeout,
-        )
-        self._persist_token(response)
-        return response
-
     def refresh(self, mcp_token):
         response = self.transport.post_json(
             self.base_url + '/mcp/auth/refresh',

services/gateway_session_store.py

@@ -1,4 +1,5 @@
-import json
+import json
+from datetime import datetime, timezone
 
 from utils.security import hash_gateway_session_id
 
@@ -30,5 +31,12 @@ class GatewaySessionStore:
             raw = raw.decode('utf-8')
         return json.loads(raw)
 
+    def touch_session(self, gateway_session_id):
+        session = self.get(gateway_session_id)
+        if not session:
+            return None
+        session['last_access_time'] = datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')
+        return self.save(gateway_session_id, session)
+
     def delete(self, gateway_session_id):
         return self.client.delete(self.key_for(gateway_session_id))

services/scoped_auth_client.py

@@ -1,39 +0,0 @@
-from services.api_client import JsonTransport
-
-
-class ScopedAuthClient:
-    def __init__(self, base_url, client_type, session_store, transport=None, timeout=10):
-        self.base_url = (base_url or '').rstrip('/')
-        self.client_type = client_type
-        self.session_store = session_store
-        self.transport = transport or JsonTransport()
-        self.timeout = int(timeout)
-
-    def exchange(self, gateway_session_id, auth_code):
-        session_key = self.build_session_key(gateway_session_id)
-        response = self.transport.post_json(
-            self.base_url + '/mcp/auth/exchange',
-            {
-                'auth_code': auth_code,
-                'client_type': self.client_type,
-                'session_key': session_key,
-            },
-            {},
-            self.timeout,
-        )
-        data = response.get('data') or {}
-        token = data.get('mcp_token')
-        if (response.get('code') or '') == 'MCP_0000' and token:
-            self.session_store.save(gateway_session_id, {
-                'session_key': session_key,
-                'mcp_token': token,
-                'admin_id': data.get('admin_id'),
-                'company_id': data.get('company_id'),
-                'client_type': self.client_type,
-                'expire_time': data.get('expire_time'),
-            })
-        return response
-
-    @staticmethod
-    def build_session_key(gateway_session_id):
-        return 'gws_' + str(gateway_session_id or '').strip()

tests/TEST_README.md

@@ -1,225 +1,29 @@
-# MCP Gateway 测试说明
+# MCP Gateway Tests
 
-本目录存放 MCP Gateway 的单元测试和集成测试，覆盖本地 stdio 模式、公网 HTTP 模式、Redis session、认证客户端、工具转发和安全工具函数。
+This directory contains unit and integration tests for the Python MCP Gateway.
 
-## 测试结构
+## Current Flow
 
-```text
-tests/
-├── test_rate_limiter.py               # 简单滑动窗口限流
-├── test_gateway_session_store_unit.py # 公网 Gateway Redis session 存储
-├── test_request_context.py            # 从 header、Bearer、cookie 解析 gateway_session_id
-├── test_scoped_auth_client.py         # 公网模式授权码换票客户端
-├── test_scoped_api_client.py          # 公网模式工具转发客户端
-├── test_public_gateway_unit.py        # PublicGatewayApp 业务路由和 session 校验
-├── test_security.py                   # session ID、hash、常量时间比较
-├── test_public_server_integration.py  # HTTP 服务集成测试
-├── test_run_coverage.py               # 覆盖率脚本回归测试
-└── test_*.py                          # 既有本地 Gateway 测试
-```
+The production Workbuddy flow is direct Gateway session access:
 
-## 运行全部测试
+1. Base creates a Workbuddy config containing a `GWS_xxx` credential.
+2. Public Gateway reads the session from Redis.
+3. Tool calls use the request-scoped MCP token stored in that Gateway session.
+4. Missing or revoked sessions return the device-invalid message.
 
-```powershell
-python -m unittest discover -s tests -p "test_*.py"
-```
+The old authorization-code binding flow is no longer part of the supported runtime.
 
-## 运行指定测试文件
-
-```powershell
-python -m unittest tests.test_rate_limiter
-```
-
-## 运行指定测试类
-
-```powershell
-python -m unittest tests.test_rate_limiter.TestSimpleRateLimiter
-```
-
-## 运行指定测试方法
-
-```powershell
-python -m unittest tests.test_rate_limiter.TestSimpleRateLimiter.test_allows_requests_within_limit
-```
-
-## 输出详细日志
+## Run All Tests
 
 ```powershell
 python -m unittest discover -s tests -p "test_*.py" -v
 ```
 
-## 覆盖范围
-
-单元测试覆盖：
-
-- 限流器：滑动窗口、清理、多 key 隔离。
-- Gateway session store：Redis CRUD、key hash、TTL。
-- Request context：从 header、Authorization、cookie 解析 session，并验证优先级。
-- Scoped auth client：授权码换票、session 持久化、错误处理。
-- Scoped API client：工具调用、token 校验、header 组装。
-- PublicGatewayApp：工具路由、session 校验、request_id 生成。
-- Security utils：session ID 生成、SHA256 hash、常量时间比较。
-
-集成测试覆盖：
-
-- HTTP `/health`。
-- MCP `initialize`、`tools/list`、`tools/call`。
-- 缺少 session 的错误响应。
-- header、Bearer、cookie 三种 session 传递方式。
-- 404 和未知方法处理。
-
-## 测试依赖
-
-当前测试使用 Python 标准库：
-
-- `unittest`
-- `unittest.mock`
-- `http.client`
-- `threading`
-
-不需要额外安装测试依赖。
-
-## 新增测试约定
-
-- 文件名使用 `test_<module_name>.py`。
-- 测试类使用 `Test<ClassName>`。
-- 测试方法使用 `test_<scenario>`。
-- 每个测试尽量只验证一个行为。
-- Redis、HTTP 客户端等外部依赖使用 mock 或 fake 实现。
-- 安全修复必须补回归测试，例如不能记录授权码、不能信任可伪造 header。
-
-## 常见问题
-
-### 集成测试提示 Connection refused
-
-HTTP 测试服务可能还未启动完成。可适当增加 `setUpClass` 中的等待时间：
-
-```python
-sleep(1.0)
-```
-
-### mock 断言异常
-
-确认每个测试前重置 mock：
-
-```python
-def setUp(self):
-    self.mock_object.reset_mock()
-```
-
-## 覆盖率报告
-
-### 快速开始
-
-使用便捷脚本：
-
-```powershell
-# 运行测试并生成所有覆盖率报告
-python run_coverage.py
-
-# 运行并自动在浏览器中打开 HTML 报告
-python run_coverage.py --open
-```
-
-### 手动命令
-
-如需手动生成覆盖率报告，需要额外安装 `coverage`：
-
-```powershell
-pip install coverage
-
-# 运行测试并收集覆盖率数据
-coverage run -m unittest discover -s tests -p "test_*.py"
-
-# 生成控制台报告
-coverage report
-
-# 生成 HTML 报告
-coverage html
-
-# 生成 XML 报告（用于 CI/CD）
-coverage xml
-```
-
-### 报告类型
-
-运行覆盖率测试后，会生成三种报告：
-
-1. **控制台报告**：在终端显示，包含每个模块的覆盖率百分比
-2. **HTML 报告**：交互式网页界面，位于 `htmlcov/index.html`
-   - 显示逐行覆盖情况，颜色编码
-   - 点击文件可查看哪些行被执行
-   - 红色 = 未覆盖，绿色 = 已覆盖
-3. **XML 报告**：机器可读格式，用于 CI/CD 工具（`coverage.xml`）
-
-### 覆盖率指标说明
-
-- **Stmts**: 可执行语句总数
-- **Miss**: 未执行的语句数
-- **Branch**: 分支总数（if/else 条件）
-- **BrPart**: 部分覆盖的分支数
-- **Cover**: 总体覆盖率百分比
-
-### 当前覆盖率 (2026-07-08)
-
-| 模块 | 覆盖率 |
-|------|----------|
-| **新增模块（Public 模式）** | |
-| `constants.py` | 100% ✅ |
-| `services/api_client.py` | 100% ✅ |
-| `services/gateway_session_store.py` | 100% ✅ |
-| `services/scoped_api_client.py` | 100% ✅ |
-| `services/scoped_auth_client.py` | 100% ✅ |
-| `tools/query_track.py` | 100% ✅ |
-| `utils/rate_limiter.py` | 100% ✅ |
-| `utils/security.py` | 100% ✅ |
-| `services/request_context.py` | 97.37% ✅ |
-| `public_gateway.py` | 95.31% ✅ |
-| `services/auth_client.py` | 94.59% ✅ |
-| `config.py` | 93.18% ✅ |
-| **现有模块** | |
-| `public_server.py` | 83.47% |
-| `app.py` | 74.58% |
-| `tools/query_order.py` | 80.95% |
-| `mcp_protocol.py` | 80.28% |
-| `tools/bind_auth_code.py` | 75.00% |
-| `services/token_store.py` | 52.40% |
-| **总体** | **80.80%** ✅ |
-
-### 覆盖率配置
-
-覆盖率通过 `.coveragerc` 配置：
-
-- **源代码**: 项目根目录的所有 Python 文件
-- **排除**: 测试文件、缓存、虚拟环境、覆盖率运行器 `run_coverage.py`
-- **分支覆盖**: 已启用（追踪 if/else 路径）
-- **报告精度**: 2 位小数
-- **排除规则**: `# pragma: no cover` 注释、调试代码、`if __name__ == '__main__'`
-
-### 提高覆盖率
-
-对于低覆盖率的文件：
-
-1. 在 HTML 报告中识别未覆盖的行
-2. 编写测试来执行这些代码路径
-3. 关注边界情况和错误处理
-4. 继续补充主入口点（`app.py`）中 `serve-public` 和异常分支的集成测试
-
-### CI/CD 集成
-
-GitHub Actions 工作流示例：
-
-```yaml
-- name: Run tests with coverage
-  run: |
-    pip install coverage
-    coverage run -m unittest discover -s tests -p "test_*.py"
-    coverage report --fail-under=80
-
-- name: Upload coverage to Codecov
-  uses: codecov/codecov-action@v3
-  with:
-    file: ./coverage.xml
-```
+## Key Test Areas
 
-`coverage html` 会在 `htmlcov/` 目录生成 HTML 报告。
+- `test_public_gateway_unit.py`: public tool routing and Gateway session checks.
+- `test_public_server_integration.py`: HTTP JSON-RPC behavior.
+- `test_gateway_session_store_unit.py`: Redis Gateway session storage.
+- `test_request_context.py`: session extraction from headers, bearer token, and cookies.
+- `test_auth_client.py`: token refresh/revoke only.
+- `test_gateway_runtime.py`: local runtime tool registration and token refresh.

tests/TEST_REPORT.md

@@ -1,226 +1,26 @@
-# 测试完成报告
+# MCP Gateway Test Report
 
-## 测试执行时间
-2026-07-08 11:19
+Updated: 2026-07-08
 
-## 测试结果
-✅ **所有测试通过！**
+## Result
 
-- **总计**: 145 个测试
-- **通过**: 145 个
-- **失败**: 0 个
-- **执行时间**: 1.865 秒
+The current MCP Gateway test suite targets the direct Gateway session flow.
 
-## 新增测试模块
+- Public Workbuddy access uses `GWS_xxx` Gateway sessions.
+- The legacy authorization-code binding path has been retired.
+- Public and stdio tool lists expose only query tools such as `query_order` and `query_track`.
+- `AuthClient` keeps only token refresh/revoke behavior.
 
-### 1. `test_rate_limiter.py` (5 测试)
-- ✅ 测试速率限制内允许请求
-- ✅ 测试超出限制后阻止请求
-- ✅ 测试不同 key 独立计数
-- ✅ 测试滑动窗口机制
-- ✅ 测试旧条目清理
-
-### 2. `test_security.py` (15 测试)
-- ✅ session ID 生成格式验证
-- ✅ session ID 唯一性验证
-- ✅ SHA256 哈希一致性
-- ✅ 不同输入产生不同哈希
-- ✅ 哈希长度验证 (64 字符)
-- ✅ 空值和 None 错误处理
-- ✅ 常量时间比较功能
-- ✅ URL 安全性验证
-
-### 3. `test_request_context.py` (14 测试)
-- ✅ 从 header 提取 session ID
-- ✅ 从 Authorization Bearer 提取
-- ✅ 从 Cookie 提取
-- ✅ 优先级处理 (header > auth > cookie)
-- ✅ 大小写不敏感处理
-- ✅ 无效格式拒绝
-- ✅ 空值处理
-
-### 4. `test_scoped_auth_client.py` (6 测试)
-- ✅ session_key 构建
-- ✅ exchange 成功场景
-- ✅ exchange 失败处理
-- ✅ 部分数据处理
-- ✅ 缺少数据字段处理
-- ✅ URL 斜杠处理
-
-### 5. `test_scoped_api_client.py` (9 测试)
-- ✅ 工具调用成功
-- ✅ 空 token 错误
-- ✅ None token 错误
-- ✅ URL 路径处理
-- ✅ 超时参数传递
-- ✅ 空 payload 处理
-- ✅ Header 格式验证
-
-### 6. `test_gateway_session_store_unit.py` (8 测试)
-- ✅ Redis key 生成
-- ✅ session 保存和哈希
-- ✅ session 读取
-- ✅ 未找到返回 None
-- ✅ bytes 数据处理
-- ✅ session 删除
-- ✅ 不同 session ID 隔离
-- ✅ 自定义前缀
-
-### 7. `test_public_gateway_unit.py` (12 测试)
-- ✅ 工具列表返回
-- ✅ request ID 生成
-- ✅ auth_code 绑定成功
-- ✅ 缺少 auth client 错误
-- ✅ 工具调用路由
-- ✅ 未注册工具错误
-- ✅ 缺少 session 错误
-- ✅ 缺少 token 错误
-- ✅ 工具调用成功
-- ✅ request ID 自动生成
-- ✅ 自定义 request ID
-
-### 8. `test_public_server_integration.py` (10 测试)
-- ✅ 健康检查端点
-- ✅ initialize 方法
-- ✅ tools/list 方法
-- ✅ 无 session 调用工具
-- ✅ bind_auth_code 工具
-- ✅ 有效 session 调用工具
-- ✅ 方法未找到错误
-- ✅ 404 路径处理
-- ✅ Authorization header 传递
-- ✅ Cookie 传递
-
-### 9. `test_run_coverage.py` (2 测试)
-- ✅ 覆盖率命令使用当前 Python 解释器执行
-- ✅ 子进程调用不使用 shell，兼容 UNC 网络路径
-
-## 测试覆盖的核心功能
-
-### 安全性
-- ✅ 256 位熵的 session ID 生成
-- ✅ SHA256 哈希存储
-- ✅ 常量时间字符串比较（防止时序攻击）
-- ✅ 速率限制（60 请求/分钟/IP）
-
-### 认证与授权
-- ✅ SSO exchange 流程
-- ✅ session 隔离
-- ✅ token 存储和读取
-- ✅ 多种 session 传递方式
-
-### 工具调用
-- ✅ 工具注册和列表
-- ✅ 工具路由
-- ✅ request ID 生成
-- ✅ header 传递
-
-### HTTP 服务
-- ✅ JSON-RPC 协议
-- ✅ 多种 HTTP 方法
-- ✅ 错误处理
-- ✅ 日志记录
-
-### 数据持久化
-- ✅ Redis 存储
-- ✅ TTL 管理
-- ✅ key 哈希
-- ✅ session 隔离
-
-## 审计日志验证
-
-集成测试中成功验证了审计日志：
-```
-[AUDIT] bind_auth_code: session_hash=85ee69407122, auth_code_provided=True
-[AUDIT] bind_success: session_hash=85ee69407122, admin_id=123, company_id=100
-[AUDIT] tool_call: session_hash=8069b9a33c6d, admin_id=456, company_id=200, tool=query_order, request_id=rq_http_1
-[AUDIT] tool_success: session_hash=8069b9a33c6d, tool=query_order, request_id=rq_http_1, code=0
-[HTTP] request: ip=127.0.0.1, method=tools/call, id=1
-```
-
-## 运行命令
+## Recommended Verification
 
 ```powershell
-# 运行所有测试
 python -m unittest discover -s tests -p "test_*.py" -v
-
-# 运行特定模块
-python -m unittest tests.test_rate_limiter -v
-python -m unittest tests.test_security -v
-python -m unittest tests.test_public_server_integration -v
 ```
 
-## 代码质量保证
-
-所有新增功能均有对应的单元测试和集成测试，确保：
-1. 功能正确性
-2. 边界条件处理
-3. 错误处理
-4. 安全性保障
-5. 性能可靠性
-
-## 下一步建议
-
-1. **增加测试覆盖率工具**：✅ 已完成
-   - 安装：`pip install coverage`
-   - 运行：`python run_coverage.py --open`
-   - 当前覆盖率：**80.80%**
-2. **CI/CD 集成**：在 GitHub Actions 或其他 CI 中自动运行测试
-3. **性能测试**：添加负载测试验证速率限制和并发性能
-4. **端到端测试**：使用真实 Redis 和后端服务进行完整流程测试
-
-## 测试覆盖率详情
-
-### 总体覆盖率：81.30%
-
-| 模块 | 语句数 | 未覆盖 | 分支数 | 部分覆盖 | 覆盖率 |
-|------|--------|--------|--------|----------|--------|
-| **新增模块（100% 覆盖）** |||||
-| `constants.py` | 1 | 0 | 0 | 0 | 100% ✅ |
-| `public_gateway.py` | 54 | 0 | 10 | 0 | 100% ✅ |
-| `services/api_client.py` | 18 | 0 | 0 | 0 | 100% ✅ |
-| `services/auth_client.py` | 31 | 0 | 6 | 0 | 100% ✅ |
-| `services/gateway_session_store.py` | 23 | 0 | 4 | 0 | 100% ✅ |
-| `services/request_context.py` | 28 | 0 | 10 | 0 | 100% ✅ |
-| `services/scoped_api_client.py` | 13 | 0 | 2 | 0 | 100% ✅ |
-| `services/scoped_auth_client.py` | 19 | 0 | 2 | 0 | 100% ✅ |
-| `tools/query_track.py` | 23 | 0 | 12 | 0 | 100% ✅ |
-| `utils/rate_limiter.py` | 29 | 0 | 8 | 0 | 100% ✅ |
-| `utils/security.py` | 12 | 0 | 2 | 0 | 100% ✅ |
-| **新增模块（90%+ 覆盖）** |||||
-| `config.py` | 94 | 4 | 38 | 5 | 93.18% ✅ |
-| **现有模块** |||||
-| `public_server.py` | 95 | 12 | 26 | 6 | 83.47% |
-| `tools/query_order.py` | 17 | 2 | 4 | 2 | 80.95% |
-| `mcp_protocol.py` | 147 | 20 | 66 | 18 | 80.28% |
-| `tools/bind_auth_code.py` | 18 | 3 | 6 | 3 | 75.00% |
-| `app.py` | 129 | 25 | 48 | 16 | 74.58% |
-| `services/token_store.py` | 158 | 63 | 50 | 10 | 52.40% |
-| **总计** | **909** | **129** | **294** | **60** | **81.30%** |
-
-### 覆盖率亮点
-
-✅ **11 个新增核心模块达到 100% 覆盖率**
-- 所有 Public 模式的关键组件
-- 完整的单元测试覆盖
-- 边界条件和错误处理全覆盖
-- 包括异常路径和日志记录
-
-✅ **1 个新增模块达到 90%+ 覆盖率**
-- config.py 达到 93.18%
-
-✅ **总体覆盖率 81.30%**
-- 超过行业标准（70-80%）
-- 新增代码质量极高
-
-### 查看详细报告
-
-```powershell
-# 生成所有报告
-python run_coverage.py --open
-
-# 或手动生成
-coverage html
-# 然后在浏览器中打开 htmlcov/index.html
-```
+## Important Coverage Areas
 
+- Gateway session parsing from header, bearer token, and cookie.
+- Redis Gateway session lookup and invalid-session user message.
+- Scoped API forwarding with request-level MCP token.
+- Local token refresh/revoke compatibility.
+- Query tool registration and JSON-RPC routing.

tests/test_auth_client.py

@@ -1,5 +1,4 @@
 import os
-import os
 import tempfile
 import unittest
 
@@ -11,7 +10,7 @@ from services.token_store import InMemoryTokenStore
 class DummyTransport:
     def __init__(self):
         self.calls = []
-        self.responses = []  # Queue of responses to return
+        self.responses = []
 
     def post_json(self, url, payload, headers, timeout):
         self.calls.append(
@@ -23,20 +22,9 @@ class DummyTransport:
             }
         )
 
-        # If custom responses are queued, return the next one
         if self.responses:
             return self.responses.pop(0)
 
-        # Default responses
-        if url.endswith('/mcp/auth/exchange'):
-            return {
-                'code': 'MCP_0000',
-                'msg': 'success',
-                'data': {
-                    'mcp_token': 'MT_exchange',
-                    'expire_time': '2099-01-01T00:00:00',
-                },
-            }
         if url.endswith('/mcp/auth/refresh'):
             return {
                 'code': 'MCP_0000',
@@ -71,7 +59,7 @@ class AuthClientTest(unittest.TestCase):
         self.assertEqual(9, config.timeout_seconds)
         self.assertEqual(120, config.refresh_skew_seconds)
 
-    def test_auth_client_uses_auth_routes_and_updates_token_store(self):
+    def test_auth_client_uses_token_routes_and_updates_token_store(self):
         transport = DummyTransport()
         store = InMemoryTokenStore(refresh_skew_seconds=60)
         client = AuthClient(
@@ -82,21 +70,17 @@ class AuthClientTest(unittest.TestCase):
             timeout=9,
         )
 
-        exchange = client.exchange('AUTH123')
         refresh = client.refresh('MT_exchange')
         self.assertEqual('MT_refresh', store.get()['token'])
         revoke = client.revoke('MT_refresh')
 
-        self.assertEqual('MT_exchange', exchange['data']['mcp_token'])
         self.assertEqual('MT_refresh', refresh['data']['mcp_token'])
         self.assertIsNone(store.get())
         self.assertEqual({}, revoke['data'])
-        self.assertEqual('http://base.example.test/mcp/auth/exchange', transport.calls[0]['url'])
-        self.assertEqual('http://base.example.test/mcp/auth/refresh', transport.calls[1]['url'])
-        self.assertEqual('http://base.example.test/mcp/auth/revoke', transport.calls[2]['url'])
-        self.assertEqual({'auth_code': 'AUTH123', 'client_type': 'workbuddy'}, transport.calls[0]['payload'])
-        self.assertEqual({'mcp_token': 'MT_exchange'}, transport.calls[1]['payload'])
-        self.assertEqual({'mcp_token': 'MT_refresh'}, transport.calls[2]['payload'])
+        self.assertEqual('http://base.example.test/mcp/auth/refresh', transport.calls[0]['url'])
+        self.assertEqual('http://base.example.test/mcp/auth/revoke', transport.calls[1]['url'])
+        self.assertEqual({'mcp_token': 'MT_exchange'}, transport.calls[0]['payload'])
+        self.assertEqual({'mcp_token': 'MT_refresh'}, transport.calls[1]['payload'])
 
     def test_refresh_and_revoke_send_bearer_header_and_revoke_clears_token(self):
         transport = DummyTransport()
@@ -118,7 +102,6 @@ class AuthClientTest(unittest.TestCase):
         self.assertIsNone(store.get())
 
     def test_revoke_does_not_clear_token_on_failure(self):
-        """Test that revoke does not clear token when response code is not MCP_0000"""
         transport = DummyTransport()
         transport.responses = [
             {'code': 'MCP_9999', 'msg': 'revoke failed'}
@@ -136,19 +119,16 @@ class AuthClientTest(unittest.TestCase):
 
         result = client.revoke('MT_token')
 
-        # Token should NOT be cleared on failure
         self.assertIsNotNone(store.get())
         self.assertEqual(result['code'], 'MCP_9999')
 
-    def test_persist_token_handles_missing_expire_time(self):
-        """Test that token is not persisted when expire_time is missing"""
+    def test_refresh_does_not_persist_token_without_expire_time(self):
         transport = DummyTransport()
         transport.responses = [
             {
                 'code': 'MCP_0000',
                 'data': {
                     'mcp_token': 'MT_no_expire'
-                    # Missing expire_time
                 }
             }
         ]
@@ -163,11 +143,19 @@ class AuthClientTest(unittest.TestCase):
             session_key='test_session'
         )
 
-        client.exchange('AC_test')
+        client.refresh('MT_old')
 
-        # Token should not be saved without expire_time
         self.assertIsNone(store.get())
 
+    def test_auth_client_no_longer_exposes_exchange(self):
+        client = AuthClient(
+            base_url='http://base.example.test',
+            client_type='workbuddy',
+            token_store=InMemoryTokenStore(),
+        )
+
+        self.assertFalse(hasattr(client, 'exchange'))
+
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

tests/test_bind_auth_code_tool.py

@@ -5,24 +5,6 @@ from mcp_protocol import McpProtocolHandler
 from services.token_store import InMemoryTokenStore
 
 
-class DummyAuthClient:
-    def __init__(self, token_store):
-        self.token_store = token_store
-        self.exchange_calls = []
-
-    def exchange(self, auth_code):
-        self.exchange_calls.append(auth_code)
-        self.token_store.save('MT_bound_secret', '2099-01-01T00:00:00')
-        return {
-            'code': 'MCP_0000',
-            'msg': 'success',
-            'data': {
-                'mcp_token': 'MT_bound_secret',
-                'expire_time': '2099-01-01T00:00:00',
-            },
-        }
-
-
 class DummyApiClient:
     def call_tool(self, tool_code, route_path, payload, request_id):
         return {
@@ -39,21 +21,20 @@ class DummyApiClient:
         }
 
 
-class BindAuthCodeToolTest(unittest.TestCase):
-    def build_handler(self, with_token=False):
+class NoBindAuthCodeToolTest(unittest.TestCase):
+    def build_handler(self, with_token=True):
         store = InMemoryTokenStore(refresh_skew_seconds=60)
         if with_token:
             store.save('MT_ready', '2099-01-01T00:00:00')
-        auth_client = DummyAuthClient(store)
         app = GatewayApp(
-            auth_client=auth_client,
+            auth_client=None,
             api_client=DummyApiClient(),
             token_store=store,
         )
-        return McpProtocolHandler(app), auth_client, store
+        return McpProtocolHandler(app), store
 
-    def test_tools_list_includes_bind_auth_code(self):
-        handler, _, _ = self.build_handler()
+    def test_tools_list_does_not_include_bind_auth_code(self):
+        handler, _ = self.build_handler()
 
         response = handler.handle_request(
             {
@@ -65,12 +46,12 @@ class BindAuthCodeToolTest(unittest.TestCase):
         )
 
         names = [tool['name'] for tool in response['result']['tools']]
-        self.assertIn('bind_auth_code', names)
-        bind_tool = [tool for tool in response['result']['tools'] if tool['name'] == 'bind_auth_code'][0]
-        self.assertEqual(['auth_code'], bind_tool['inputSchema']['required'])
+        self.assertIn('query_order', names)
+        self.assertIn('query_track', names)
+        self.assertNotIn('bind_auth_code', names)
 
-    def test_bind_auth_code_exchanges_code_and_hides_token(self):
-        handler, auth_client, store = self.build_handler()
+    def test_bind_auth_code_call_is_not_registered(self):
+        handler, _ = self.build_handler()
 
         response = handler.handle_request(
             {
@@ -80,24 +61,17 @@ class BindAuthCodeToolTest(unittest.TestCase):
                 'params': {
                     'name': 'bind_auth_code',
                     'arguments': {
-                        'auth_code': ' AUTH123 ',
+                        'auth_code': 'AUTH123',
                     },
                 },
             }
         )
 
-        self.assertFalse(response['result']['isError'])
-        self.assertEqual(['AUTH123'], auth_client.exchange_calls)
-        self.assertEqual('MT_bound_secret', store.get()['token'])
-        text = response['result']['content'][0]['text']
-        structured = response['result']['structuredContent']
-        self.assertIn('绑定成功', text)
-        self.assertNotIn('MT_bound_secret', text)
-        self.assertNotIn('mcp_token', structured)
-        self.assertEqual('bound', structured['status'])
+        self.assertTrue(response['result']['isError'])
+        self.assertIn('tool not registered', response['result']['content'][0]['text'])
 
-    def test_query_order_without_token_returns_employee_binding_hint(self):
-        handler, _, _ = self.build_handler(with_token=False)
+    def test_query_order_without_token_returns_device_invalid_message(self):
+        handler, _ = self.build_handler(with_token=False)
 
         response = handler.handle_request(
             {
@@ -114,7 +88,8 @@ class BindAuthCodeToolTest(unittest.TestCase):
         )
 
         self.assertTrue(response['result']['isError'])
-        self.assertIn('请先登录后台获取 Workbuddy 授权码', response['result']['content'][0]['text'])
+        self.assertNotIn('授权码', response['result']['content'][0]['text'])
+        self.assertIn('Workbuddy', response['result']['content'][0]['text'])
 
 
 if __name__ == '__main__':

tests/test_cli_and_file_store.py

@@ -1,4 +1,4 @@
-import io
+import io
 import json
 import os
 import tempfile
@@ -11,17 +11,17 @@ from services.token_store import FileTokenStore
 class DummyAuthClient:
     def __init__(self, token_store):
         self.token_store = token_store
-        self.exchange_calls = []
+        self.refresh_calls = []
 
-    def exchange(self, auth_code):
-        self.exchange_calls.append(auth_code)
-        self.token_store.save('MT_bound', '2099-01-01T00:00:00')
+    def refresh(self, mcp_token):
+        self.refresh_calls.append(mcp_token)
+        self.token_store.save('MT_refreshed', '2099-01-02T00:00:00')
         return {
             'code': 'MCP_0000',
             'msg': 'success',
             'data': {
-                'mcp_token': 'MT_bound',
-                'expire_time': '2099-01-01T00:00:00',
+                'mcp_token': 'MT_refreshed',
+                'expire_time': '2099-01-02T00:00:00',
             },
         }
 
@@ -67,40 +67,41 @@ class CliAndFileStoreTest(unittest.TestCase):
             self.assertIsNone(second.get())
             self.assertFalse(os.path.exists(path))
 
-    def test_run_cli_bind_and_call_emit_json(self):
+    def test_run_cli_call_emit_json_without_bind_command(self):
         with tempfile.TemporaryDirectory() as tmp_dir:
             path = os.path.join(tmp_dir, 'token.json')
             token_store = FileTokenStore(path, refresh_skew_seconds=60)
-            auth_client = DummyAuthClient(token_store)
+            token_store.save('MT_bound', '2099-01-01T00:00:00')
             api_client = DummyApiClient()
             app = GatewayApp(
-                auth_client=auth_client,
+                auth_client=DummyAuthClient(token_store),
                 api_client=api_client,
                 token_store=token_store,
             )
 
-            bind_stdout = io.StringIO()
-            bind_code = app.run_cli(['bind', '--auth-code', 'AUTH123'], stdout=bind_stdout)
-            bind_payload = json.loads(bind_stdout.getvalue())
-
-            call_stdout = io.StringIO()
+            stdout = io.StringIO()
             call_code = app.run_cli([
                 'call',
                 '--tool', 'query_order',
                 '--keyword', 'SO20260706001',
                 '--page', '2',
                 '--limit', '15',
-            ], stdout=call_stdout)
-            call_payload = json.loads(call_stdout.getvalue())
+            ], stdout=stdout)
+            call_payload = json.loads(stdout.getvalue())
 
-            self.assertEqual(0, bind_code)
-            self.assertEqual('MCP_0000', bind_payload['code'])
             self.assertEqual(0, call_code)
             self.assertEqual('MCP_0000', call_payload['code'])
-            self.assertEqual(['AUTH123'], auth_client.exchange_calls)
             self.assertEqual('query_order', api_client.calls[0]['tool_code'])
             self.assertEqual({'keyword': 'SO20260706001', 'page': 2, 'limit': 15}, api_client.calls[0]['payload'])
 
+    def test_run_cli_bind_command_is_not_registered(self):
+        app = GatewayApp(auth_client=None, api_client=None, token_store=None)
+
+        with self.assertRaises(SystemExit) as error:
+            app.run_cli(['bind', '--auth-code', 'AUTH123'])
+
+        self.assertNotEqual(0, error.exception.code)
+
     def test_run_cli_query_track_accepts_tracking_number(self):
         with tempfile.TemporaryDirectory() as tmp_dir:
             path = os.path.join(tmp_dir, 'token.json')
@@ -128,6 +129,7 @@ class CliAndFileStoreTest(unittest.TestCase):
             self.assertEqual('1471904540000000301', api_client.calls[0]['payload']['tracking_number'])
             self.assertNotIn('order_id', api_client.calls[0]['payload'])
             self.assertNotIn('order_number', api_client.calls[0]['payload'])
+
     def test_run_cli_serve_stdio_handles_initialize_request(self):
         with tempfile.TemporaryDirectory() as tmp_dir:
             path = os.path.join(tmp_dir, 'token.json')
@@ -170,5 +172,6 @@ class CliAndFileStoreTest(unittest.TestCase):
 
         self.assertEqual(0, error.exception.code)
 
+
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

tests/test_config_compat.py

@@ -1,4 +1,4 @@
-import os
+import os
 import tempfile
 import unittest
 
@@ -83,6 +83,15 @@ class GatewayConfigCompatTest(unittest.TestCase):
         self.assertEqual(600, config.gateway_session_ttl_seconds)
         self.assertEqual('fms:mcp:gateway:', config.redis_prefix)
 
+    def test_public_gateway_mode_defaults_to_gateway_redis_prefix(self):
+        config = GatewayConfig.from_env(env={
+            'FMS_API_BASE': 'https://base.example.com',
+            'FMS_GATEWAY_MODE': 'public',
+        }, dotenv_path='missing.env')
+
+        self.assertEqual('public', config.gateway_mode)
+        self.assertEqual('fms:mcp:gateway:', config.redis_prefix)
+
     def test_timeout_ms_conversion_in_preferred_env(self):
         """Test FMS_TIMEOUT_MS conversion in preferred env"""
         config = GatewayConfig.from_env(env={

tests/test_gateway_runtime.py

@@ -8,21 +8,8 @@ from services.token_store import InMemoryTokenStore
 class DummyAuthClient:
     def __init__(self, token_store):
         self.token_store = token_store
-        self.exchange_calls = []
         self.refresh_calls = []
 
-    def exchange(self, auth_code):
-        self.exchange_calls.append(auth_code)
-        self.token_store.save('MT_exchange', '2099-01-01T00:00:00')
-        return {
-            'code': 'MCP_0000',
-            'msg': 'success',
-            'data': {
-                'mcp_token': 'MT_exchange',
-                'expire_time': '2099-01-01T00:00:00',
-            },
-        }
-
     def refresh(self, mcp_token):
         self.refresh_calls.append(mcp_token)
         self.token_store.save('MT_refresh', '2099-01-02T00:00:00')
@@ -64,7 +51,7 @@ class DummyApiClient:
 
 
 class GatewayRuntimeTest(unittest.TestCase):
-    def test_bind_exchanges_auth_code_and_persists_token(self):
+    def test_list_tools_does_not_include_bind_auth_code(self):
         store = InMemoryTokenStore(refresh_skew_seconds=60)
         app = GatewayApp(
             auth_client=DummyAuthClient(store),
@@ -72,10 +59,12 @@ class GatewayRuntimeTest(unittest.TestCase):
             token_store=store,
         )
 
-        response = app.bind('AUTH123')
+        names = [tool['name'] for tool in app.list_tools()]
 
-        self.assertEqual('MCP_0000', response['code'])
-        self.assertEqual('MT_exchange', store.get()['token'])
+        self.assertIn('query_order', names)
+        self.assertIn('query_track', names)
+        self.assertNotIn('bind_auth_code', names)
+        self.assertFalse(hasattr(app, 'bind'))
 
     def test_call_tool_refreshes_expiring_token_and_generates_request_id(self):
         store = InMemoryTokenStore(refresh_skew_seconds=60)

tests/test_gateway_session_store.py

@@ -1,4 +1,4 @@
-import unittest
+import unittest
 
 from services.gateway_session_store import GatewaySessionStore
 
@@ -60,6 +60,17 @@ class GatewaySessionStoreTest(unittest.TestCase):
         self.assertIsNone(store.get('GWS_employee_a'))
         self.assertEqual('MT_B', store.get('GWS_employee_b')['mcp_token'])
 
+    def test_touch_session_updates_last_access_time(self):
+        redis = FakeRedis()
+        store = GatewaySessionStore(redis, prefix='fms:mcp:gateway:', ttl_seconds=3600)
+
+        store.save('GWS_employee_a', {'mcp_token': 'MT_A'})
+        touched = store.touch_session('GWS_employee_a')
+
+        self.assertIsNotNone(touched)
+        self.assertIn('last_access_time', touched)
+        self.assertTrue(touched['last_access_time'].endswith('Z'))
+        self.assertEqual(touched['last_access_time'], store.get('GWS_employee_a')['last_access_time'])
 
 if __name__ == '__main__':
     unittest.main()

tests/test_mcp_protocol.py

@@ -152,7 +152,7 @@ class McpProtocolTest(unittest.TestCase):
         self.assertEqual(2, response['id'])
         by_name = {tool['name']: tool for tool in response['result']['tools']}
         self.assertIn('query_order', by_name)
-        self.assertIn('bind_auth_code', by_name)
+        self.assertNotIn('bind_auth_code', by_name)
         self.assertIn('inputSchema', by_name['query_order'])
 
     def test_tools_call_wraps_gateway_result_as_structured_content(self):

tests/test_public_gateway.py

@@ -1,4 +1,4 @@
-import unittest
+import unittest
 
 from public_gateway import PublicGatewayApp
 
@@ -6,10 +6,14 @@ from public_gateway import PublicGatewayApp
 class FakeSessionStore:
     def __init__(self):
         self.sessions = {}
+        self.touched = []
 
     def get(self, gateway_session_id):
         return self.sessions.get(gateway_session_id)
 
+    def touch_session(self, gateway_session_id):
+        self.touched.append(gateway_session_id)
+
 
 class FakeApiClient:
     def __init__(self):
@@ -36,13 +40,31 @@ class PublicGatewayAppTest(unittest.TestCase):
         self.assertEqual('MT_A', api_client.calls[0][0])
         self.assertEqual('MT_B', api_client.calls[1][0])
 
-    def test_missing_session_returns_bind_hint(self):
+    def test_public_tools_do_not_include_bind_auth_code(self):
+        app = PublicGatewayApp(session_store=FakeSessionStore(), api_client=FakeApiClient(), auth_client=None)
+
+        tool_names = [tool['name'] for tool in app.list_tools()]
+
+        self.assertNotIn('bind_auth_code', tool_names)
+        self.assertIn('query_order', tool_names)
+        self.assertIn('query_track', tool_names)
+
+    def test_missing_session_returns_human_device_message(self):
         app = PublicGatewayApp(session_store=FakeSessionStore(), api_client=FakeApiClient(), auth_client=None)
 
         with self.assertRaises(RuntimeError) as error:
             app.call_tool('GWS_missing', 'query_order', {'keyword': 'A'}, request_id='rq_missing')
 
-        self.assertIn('绑定授权码', str(error.exception))
+        self.assertIn('这台设备的 Workbuddy 配置已失效，请重新生成配置', str(error.exception))
+
+    def test_successful_tool_call_touches_gateway_session(self):
+        store = FakeSessionStore()
+        store.sessions['GWS_A'] = {'mcp_token': 'MT_A'}
+        app = PublicGatewayApp(session_store=store, api_client=FakeApiClient(), auth_client=None)
+
+        app.call_tool('GWS_A', 'query_order', {'keyword': 'A'}, request_id='rq_a')
+
+        self.assertEqual(['GWS_A'], store.touched)
 
 
 if __name__ == '__main__':

tests/test_public_gateway_unit.py

@@ -1,7 +1,7 @@
-import unittest
-from unittest.mock import MagicMock, patch
+import unittest
+from unittest.mock import MagicMock
 
-from constants import BIND_HINT
+from constants import DEVICE_INVALID_MESSAGE
 from public_gateway import PublicGatewayApp
 
 
@@ -9,12 +9,10 @@ class TestPublicGatewayApp(unittest.TestCase):
     def setUp(self):
         self.mock_session_store = MagicMock()
         self.mock_api_client = MagicMock()
-        self.mock_auth_client = MagicMock()
 
         self.app = PublicGatewayApp(
             session_store=self.mock_session_store,
             api_client=self.mock_api_client,
-            auth_client=self.mock_auth_client
         )
 
     def test_list_tools_returns_metadata(self):
@@ -22,8 +20,11 @@ class TestPublicGatewayApp(unittest.TestCase):
 
         self.assertIsInstance(tools, list)
         self.assertGreater(len(tools), 0)
+        tool_names = [tool['name'] for tool in tools]
+        self.assertIn('query_order', tool_names)
+        self.assertIn('query_track', tool_names)
+        self.assertNotIn('bind_auth_code', tool_names)
 
-        # Check that each tool has required fields
         for tool in tools:
             self.assertIn('name', tool)
             self.assertIn('description', tool)
@@ -33,7 +34,7 @@ class TestPublicGatewayApp(unittest.TestCase):
         request_id = self.app.build_request_id('')
 
         self.assertTrue(request_id.startswith('rq_'))
-        self.assertEqual(len(request_id), 3 + 16)  # 'rq_' + 16 hex chars
+        self.assertEqual(len(request_id), 3 + 16)
 
     def test_build_request_id_uses_provided_id(self):
         provided_id = 'custom_request_123'
@@ -41,75 +42,11 @@ class TestPublicGatewayApp(unittest.TestCase):
 
         self.assertEqual(request_id, provided_id)
 
-    def test_bind_auth_code_success(self):
-        gateway_session_id = 'GWS_test123'
-        auth_code = 'AC_xyz789'
-
-        self.mock_auth_client.exchange.return_value = {
-            'code': 'MCP_0000',
-            'msg': 'success',
-            'data': {'mcp_token': 'MT_token'}
-        }
-
-        self.mock_session_store.get.return_value = {
-            'admin_id': 123,
-            'company_id': 100
-        }
-
-        result = self.app.bind_auth_code(gateway_session_id, auth_code)
-
-        self.mock_auth_client.exchange.assert_called_once_with(gateway_session_id, auth_code)
-        self.assertEqual(result['code'], 'MCP_0000')
-
-    def test_bind_auth_code_log_does_not_include_auth_code_value(self):
-        gateway_session_id = 'GWS_test123'
-        auth_code = 'AC_secret_value_123'
-        self.mock_auth_client.exchange.return_value = {'code': 'MCP_9999', 'msg': 'failed'}
-
-        with self.assertLogs('public_gateway', level='INFO') as logs:
-            self.app.bind_auth_code(gateway_session_id, auth_code)
-
-        joined_logs = '\n'.join(logs.output)
-        self.assertIn('bind_auth_code', joined_logs)
-        self.assertNotIn(auth_code, joined_logs)
-        self.assertNotIn('AC_secre', joined_logs)
-    def test_bind_auth_code_log_does_not_include_auth_code_value(self):
-        gateway_session_id = 'GWS_test123'
-        auth_code = 'AC_secret_value_123'
-        self.mock_auth_client.exchange.return_value = {'code': 'MCP_9999', 'msg': 'failed'}
-
-        with self.assertLogs('public_gateway', level='INFO') as logs:
-            self.app.bind_auth_code(gateway_session_id, auth_code)
-
-        joined_logs = '\n'.join(logs.output)
-        self.assertIn('bind_auth_code', joined_logs)
-        self.assertNotIn(auth_code, joined_logs)
-        self.assertNotIn('AC_secre', joined_logs)
-    def test_bind_auth_code_raises_when_no_auth_client(self):
-        app_no_auth = PublicGatewayApp(
-            session_store=self.mock_session_store,
-            api_client=self.mock_api_client,
-            auth_client=None
-        )
-
-        with self.assertRaises(RuntimeError) as context:
-            app_no_auth.bind_auth_code('GWS_test', 'AC_code')
-
-        self.assertIn('auth client is required', str(context.exception))
-
-    def test_call_tool_bind_auth_code(self):
-        gateway_session_id = 'GWS_bind'
-        arguments = {'auth_code': 'AC_test'}
-
-        self.mock_auth_client.exchange.return_value = {
-            'code': 'MCP_0000',
-            'data': {}
-        }
-        self.mock_session_store.get.return_value = {'admin_id': 1}
-
-        result = self.app.call_tool(gateway_session_id, 'bind_auth_code', arguments)
+    def test_call_tool_bind_auth_code_is_not_registered_in_public_mode(self):
+        with self.assertRaises(KeyError) as context:
+            self.app.call_tool('GWS_bind', 'bind_auth_code', {'auth_code': 'AC_test'})
 
-        self.mock_auth_client.exchange.assert_called_once_with(gateway_session_id, 'AC_test')
+        self.assertIn('tool not registered', str(context.exception))
 
     def test_call_tool_raises_on_unregistered_tool(self):
         with self.assertRaises(KeyError) as context:
@@ -123,19 +60,18 @@ class TestPublicGatewayApp(unittest.TestCase):
         with self.assertRaises(RuntimeError) as context:
             self.app.call_tool('GWS_nosession', 'query_order', {})
 
-        self.assertEqual(str(context.exception), BIND_HINT)
+        self.assertEqual(str(context.exception), DEVICE_INVALID_MESSAGE)
 
     def test_call_tool_raises_when_no_token_in_session(self):
         self.mock_session_store.get.return_value = {
             'admin_id': 123,
             'company_id': 100
-            # Missing mcp_token
         }
 
         with self.assertRaises(RuntimeError) as context:
             self.app.call_tool('GWS_notoken', 'query_order', {})
 
-        self.assertEqual(str(context.exception), BIND_HINT)
+        self.assertEqual(str(context.exception), DEVICE_INVALID_MESSAGE)
 
     def test_call_tool_success(self):
         gateway_session_id = 'GWS_valid'
@@ -156,15 +92,12 @@ class TestPublicGatewayApp(unittest.TestCase):
 
         result = self.app.call_tool(gateway_session_id, tool_name, arguments, 'rq_test')
 
-        # Verify api_client.call_tool was called correctly
         self.mock_api_client.call_tool.assert_called_once()
         call_args = self.mock_api_client.call_tool.call_args[1]
         self.assertEqual(call_args['token'], 'MT_token123')
         self.assertEqual(call_args['tool_code'], 'query_order')
         self.assertEqual(call_args['payload'], arguments)
-        self.assertTrue(call_args['request_id'].startswith('rq_'))
-
-        # Verify result
+        self.assertEqual(call_args['request_id'], 'rq_test')
         self.assertEqual(result['code'], '0')
 
     def test_call_tool_generates_request_id_when_not_provided(self):
@@ -197,7 +130,6 @@ class TestPublicGatewayApp(unittest.TestCase):
         self.assertEqual(call_args['request_id'], custom_request_id)
 
     def test_call_tool_logs_error_on_exception(self):
-        """Test that tool call exceptions are logged and re-raised"""
         gateway_session_id = 'GWS_error_test'
         tool_name = 'query_order'
 
@@ -207,17 +139,13 @@ class TestPublicGatewayApp(unittest.TestCase):
             'company_id': 888
         }
 
-        # Make API client raise an exception
-        self.mock_api_client.call_tool.side_effect = RuntimeError("API connection failed")
+        self.mock_api_client.call_tool.side_effect = RuntimeError('API connection failed')
 
         with self.assertRaises(RuntimeError) as context:
             self.app.call_tool(gateway_session_id, tool_name, {}, 'rq_err')
 
-        self.assertIn("API connection failed", str(context.exception))
+        self.assertIn('API connection failed', str(context.exception))
 
 
 if __name__ == '__main__':
-    unittest.main()
-
-
-
+    unittest.main()

tests/test_public_server.py

@@ -1,4 +1,4 @@
-import unittest
+import unittest
 
 from public_server import PublicMcpHttpHandler, extract_client_ip
 
@@ -63,7 +63,7 @@ class PublicMcpHttpHandlerTest(unittest.TestCase):
         )
 
         self.assertTrue(response['result']['isError'])
-        self.assertIn('绑定授权码', response['result']['content'][0]['text'])
+        self.assertIn('这台设备的 Workbuddy 配置已失效，请重新生成配置', response['result']['content'][0]['text'])
 
     def test_extract_client_ip_ignores_spoofable_forwarded_for_header(self):
         client_ip = extract_client_ip(

tests/test_public_server_integration.py

@@ -5,6 +5,7 @@ from threading import Thread
 from time import sleep
 from unittest.mock import MagicMock
 
+from constants import DEVICE_INVALID_MESSAGE
 from public_gateway import PublicGatewayApp
 from public_server import serve_public
 from utils.security import generate_gateway_session_id
@@ -91,6 +92,10 @@ class TestPublicServerIntegration(unittest.TestCase):
         self.assertIn('tools', result['result'])
         tools = result['result']['tools']
         self.assertGreater(len(tools), 0)
+        tool_names = [tool['name'] for tool in tools]
+        self.assertIn('query_order', tool_names)
+        self.assertIn('query_track', tool_names)
+        self.assertNotIn('bind_auth_code', tool_names)
 
         # Check tool structure
         for tool in tools:
@@ -103,23 +108,12 @@ class TestPublicServerIntegration(unittest.TestCase):
 
         self.assertIn('result', result)
         self.assertTrue(result['result']['isError'])
-        self.assertIn('请先登录后台', result['result']['content'][0]['text'])
+        self.assertIn(DEVICE_INVALID_MESSAGE, result['result']['content'][0]['text'])
 
-    def test_tools_call_bind_auth_code(self):
+    def test_tools_call_bind_auth_code_is_not_registered_in_public_mode(self):
         session_id = generate_gateway_session_id()
         auth_code = 'AC_test123'
 
-        self.mock_auth_client.exchange.return_value = {
-            'code': 'MCP_0000',
-            'msg': 'success',
-            'data': {'mcp_token': 'MT_token'}
-        }
-
-        self.mock_session_store.get.return_value = {
-            'admin_id': 123,
-            'company_id': 100
-        }
-
         result = self._make_request(
             'tools/call',
             {
@@ -130,8 +124,9 @@ class TestPublicServerIntegration(unittest.TestCase):
         )
 
         self.assertIn('result', result)
-        self.assertFalse(result['result']['isError'])
-        self.mock_auth_client.exchange.assert_called_once_with(session_id, auth_code)
+        self.assertTrue(result['result']['isError'])
+        self.assertIn('tool not registered', result['result']['content'][0]['text'])
+        self.mock_auth_client.exchange.assert_not_called()
 
     def test_tools_call_with_valid_session(self):
         session_id = generate_gateway_session_id()

tests/test_redis_token_store.py

@@ -1,4 +1,4 @@
-import json
+import json
 import os
 import tempfile
 import unittest
@@ -97,7 +97,7 @@ class RedisTokenStoreTest(unittest.TestCase):
         self.assertEqual('MT_second', second.get()['token'])
         self.assertEqual(['fms:mcp:workbuddy:pc-a:user-a'], redis.deleted)
 
-    def test_auth_client_exchange_sends_session_key(self):
+    def test_auth_client_refresh_persists_token_and_sends_bearer(self):
         transport = DummyTransport()
         redis = FakeRedisClient()
         store = RedisTokenStore(redis, prefix='fms:mcp:workbuddy:', session_key='pc-a:user-a')
@@ -110,16 +110,11 @@ class RedisTokenStoreTest(unittest.TestCase):
             session_key='pc-a:user-a',
         )
 
-        client.exchange('AUTH123')
+        client.refresh('MT_old')
 
-        self.assertEqual(
-            {
-                'auth_code': 'AUTH123',
-                'client_type': 'workbuddy',
-                'session_key': 'pc-a:user-a',
-            },
-            transport.calls[0]['payload'],
-        )
+        self.assertEqual('http://auth.example.test/mcp/auth/refresh', transport.calls[0]['url'])
+        self.assertEqual({'mcp_token': 'MT_old'}, transport.calls[0]['payload'])
+        self.assertEqual({'Authorization': 'Bearer MT_old'}, transport.calls[0]['headers'])
         self.assertEqual('MT_session', store.get()['token'])
 
     def test_gateway_from_config_uses_redis_token_store_when_configured(self):
@@ -143,4 +138,4 @@ class RedisTokenStoreTest(unittest.TestCase):
 
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

tests/test_scoped_auth_client.py

@@ -1,151 +0,0 @@
-import unittest
-from unittest.mock import MagicMock
-
-from services.scoped_auth_client import ScopedAuthClient
-
-
-class TestScopedAuthClient(unittest.TestCase):
-    def setUp(self):
-        self.mock_transport = MagicMock()
-        self.mock_session_store = MagicMock()
-        self.client = ScopedAuthClient(
-            base_url='http://test.example.com',
-            client_type='workbuddy',
-            session_store=self.mock_session_store,
-            transport=self.mock_transport,
-            timeout=10
-        )
-
-    def test_build_session_key(self):
-        gateway_session_id = 'GWS_abc123'
-        session_key = ScopedAuthClient.build_session_key(gateway_session_id)
-
-        self.assertEqual(session_key, 'gws_GWS_abc123')
-
-    def test_exchange_success(self):
-        gateway_session_id = 'GWS_test123'
-        auth_code = 'AC_xyz789'
-
-        # Mock successful response
-        self.mock_transport.post_json.return_value = {
-            'code': 'MCP_0000',
-            'msg': 'success',
-            'data': {
-                'mcp_token': 'MT_token123',
-                'admin_id': 456,
-                'company_id': 200,
-                'expire_time': '2099-12-31 23:59:59',
-            }
-        }
-
-        result = self.client.exchange(gateway_session_id, auth_code)
-
-        # Verify transport call
-        self.mock_transport.post_json.assert_called_once_with(
-            'http://test.example.com/mcp/auth/exchange',
-            {
-                'auth_code': auth_code,
-                'client_type': 'workbuddy',
-                'session_key': 'gws_GWS_test123',
-            },
-            {},
-            10
-        )
-
-        # Verify session store save
-        self.mock_session_store.save.assert_called_once()
-        save_call_args = self.mock_session_store.save.call_args
-        self.assertEqual(save_call_args[0][0], gateway_session_id)
-
-        saved_session = save_call_args[0][1]
-        self.assertEqual(saved_session['session_key'], 'gws_GWS_test123')
-        self.assertEqual(saved_session['mcp_token'], 'MT_token123')
-        self.assertEqual(saved_session['admin_id'], 456)
-        self.assertEqual(saved_session['company_id'], 200)
-        self.assertEqual(saved_session['client_type'], 'workbuddy')
-        self.assertEqual(saved_session['expire_time'], '2099-12-31 23:59:59')
-
-        # Verify result
-        self.assertEqual(result['code'], 'MCP_0000')
-
-    def test_exchange_failure_no_token(self):
-        gateway_session_id = 'GWS_fail'
-        auth_code = 'AC_invalid'
-
-        # Mock failure response
-        self.mock_transport.post_json.return_value = {
-            'code': 'MCP_1001',
-            'msg': 'auth_code invalid',
-            'data': {}
-        }
-
-        result = self.client.exchange(gateway_session_id, auth_code)
-
-        # Should not save to session store
-        self.mock_session_store.save.assert_not_called()
-
-        # Should still return the response
-        self.assertEqual(result['code'], 'MCP_1001')
-
-    def test_exchange_partial_data(self):
-        gateway_session_id = 'GWS_partial'
-        auth_code = 'AC_partial'
-
-        # Mock response with missing admin_id and company_id
-        self.mock_transport.post_json.return_value = {
-            'code': 'MCP_0000',
-            'msg': 'success',
-            'data': {
-                'mcp_token': 'MT_token456',
-                'expire_time': '2099-12-31 23:59:59',
-                # Missing admin_id and company_id
-            }
-        }
-
-        result = self.client.exchange(gateway_session_id, auth_code)
-
-        # Should still save, with None values
-        self.mock_session_store.save.assert_called_once()
-        saved_session = self.mock_session_store.save.call_args[0][1]
-        self.assertEqual(saved_session['mcp_token'], 'MT_token456')
-        self.assertIsNone(saved_session['admin_id'])
-        self.assertIsNone(saved_session['company_id'])
-
-    def test_exchange_no_data_field(self):
-        gateway_session_id = 'GWS_nodata'
-        auth_code = 'AC_nodata'
-
-        # Mock response without data field
-        self.mock_transport.post_json.return_value = {
-            'code': 'MCP_9999',
-            'msg': 'error'
-        }
-
-        result = self.client.exchange(gateway_session_id, auth_code)
-
-        # Should not save
-        self.mock_session_store.save.assert_not_called()
-
-    def test_exchange_strips_base_url_trailing_slash(self):
-        client_with_slash = ScopedAuthClient(
-            base_url='http://test.example.com/',
-            client_type='workbuddy',
-            session_store=self.mock_session_store,
-            transport=self.mock_transport,
-            timeout=10
-        )
-
-        self.mock_transport.post_json.return_value = {
-            'code': 'MCP_0000',
-            'data': {'mcp_token': 'MT_test'}
-        }
-
-        client_with_slash.exchange('GWS_test', 'AC_test')
-
-        # Should call without double slash
-        call_url = self.mock_transport.post_json.call_args[0][0]
-        self.assertEqual(call_url, 'http://test.example.com/mcp/auth/exchange')
-
-
-if __name__ == '__main__':
-    unittest.main()

tests/test_scoped_clients.py

@@ -1,7 +1,6 @@
-import unittest
+import unittest
 
 from services.scoped_api_client import ScopedApiClient
-from services.scoped_auth_client import ScopedAuthClient
 
 
 class FakeTransport:
@@ -35,47 +34,5 @@ class ScopedApiClientTest(unittest.TestCase):
         self.assertEqual(7, timeout)
 
 
-class ScopedAuthClientTest(unittest.TestCase):
-    def test_exchange_sends_gateway_session_key_and_stores_token(self):
-        transport = FakeTransport()
-        store = {}
-
-        class Store:
-            def save(self, gateway_session_id, session):
-                store[gateway_session_id] = session
-                return session
-
-        def post_json(url, payload, headers, timeout):
-            transport.calls.append((url, payload, headers, timeout))
-            return {
-                'code': 'MCP_0000',
-                'data': {
-                    'mcp_token': 'MT_employee_a',
-                    'expire_time': '2099-12-31 23:59:59',
-                    'admin_id': 1,
-                    'company_id': 10,
-                },
-            }
-
-        transport.post_json = post_json
-        client = ScopedAuthClient(
-            base_url='https://base.example.com',
-            client_type='workbuddy',
-            session_store=Store(),
-            transport=transport,
-            timeout=5,
-        )
-
-        response = client.exchange('GWS_employee_a', 'AUTH_CODE_A')
-
-        self.assertEqual('MCP_0000', response['code'])
-        url, payload, headers, timeout = transport.calls[0]
-        self.assertEqual('https://base.example.com/mcp/auth/exchange', url)
-        self.assertEqual('AUTH_CODE_A', payload['auth_code'])
-        self.assertEqual('workbuddy', payload['client_type'])
-        self.assertEqual('gws_' + 'GWS_employee_a', payload['session_key'])
-        self.assertEqual('MT_employee_a', store['GWS_employee_a']['mcp_token'])
-
-
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

tools/bind_auth_code.py

@@ -1,49 +0,0 @@
-class BindAuthCodeTool:
-    name = 'bind_auth_code'
-    requires_session = False
-
-    def __init__(self, auth_client=None):
-        self.auth_client = auth_client
-
-    def metadata(self):
-        return {
-            'name': self.name,
-            'description': 'Bind a one-time Workbuddy auth code and create the local MCP session.',
-            'input_schema': {
-                'type': 'object',
-                'properties': {
-                    'auth_code': {
-                        'type': 'string',
-                        'description': 'One-time auth code copied from the FMS backend.',
-                    },
-                },
-                'required': ['auth_code'],
-            },
-        }
-
-    def call(self, auth_code, request_id=''):
-        if self.auth_client is None:
-            raise RuntimeError('auth client is required for bind_auth_code')
-        auth_code = str(auth_code or '').strip()
-        if not auth_code:
-            raise ValueError('auth_code is required')
-        response = self.auth_client.exchange(auth_code)
-        if (response.get('code') or '') != 'MCP_0000':
-            raise RuntimeError(response.get('msg') or 'auth_code bind failed')
-        data = response.get('data') or {}
-        return {
-            'code': 'MCP_0000',
-            'msg': 'success',
-            'data': {
-                'status': 'bound',
-                'summary': '绑定成功，可以继续使用 query_order 查询订单。',
-                'expire_time': data.get('expire_time'),
-                'tips': [
-                    '后续工具调用会自动使用本机已保存的 MCP 会话。',
-                    '如果授权失效，请回后台重新获取 Workbuddy 授权码。',
-                ],
-            },
-            'meta': {
-                'request_id': request_id,
-            },
-        }