新增token

mcp-for-erp/src/auth.py

@@ -0,0 +1,63 @@
+"""静态 Bearer Token 鉴权（Authorization）。
+
+token → 客户端 映射来自环境变量 MCP_AUTH_TOKENS，格式（逗号分隔，每对 token:client_id）：
+    MCP_AUTH_TOKENS=token-trae:trae,token-partner:partner-a
+
+每个客户端配置自己的 token 调用，服务端校验失败返回 401；
+工具内可通过 get_caller() 获取当前调用方标识。
+"""
+
+import os
+
+from mcp.server.auth.middleware.auth_context import get_access_token
+from mcp.server.auth.provider import AccessToken
+from mcp.server.auth.settings import AuthSettings
+
+
+class StaticTokenVerifier:
+    """静态 token 查表校验：命中返回 AccessToken（client_id 即客户端标识），未命中返回 None（401）。"""
+
+    def __init__(self, tokens: dict[str, str]):
+        self._tokens = tokens
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        client_id = self._tokens.get(token)
+        if client_id is None:
+            return None
+        return AccessToken(token=token, client_id=client_id, scopes=[], expires_at=None)
+
+
+def _parse_tokens(raw: str) -> dict[str, str]:
+    """解析 'token1:client1,token2:client2' → {token: client_id}"""
+    tokens: dict[str, str] = {}
+    for item in raw.split(","):
+        item = item.strip()
+        if not item:
+            continue
+        token, _, client_id = item.partition(":")
+        token, client_id = token.strip(), client_id.strip()
+        if token and client_id:
+            tokens[token] = client_id
+    return tokens
+
+
+def get_auth(port: int) -> tuple[AuthSettings, StaticTokenVerifier]:
+    """构建 MCPServer 的 (auth, token_verifier) 参数。
+
+    服务对外地址默认 http://localhost:{port}，部署时用 MCP_PUBLIC_URL 覆盖
+    （如 http://192.168.1.119:8002），用于 OAuth 资源元数据发现。
+    """
+    url = os.getenv("MCP_PUBLIC_URL") or f"http://localhost:{port}"
+    tokens = _parse_tokens(os.getenv("MCP_AUTH_TOKENS", ""))
+    if not tokens:
+        raise RuntimeError("环境变量 MCP_AUTH_TOKENS 未配置，格式：token1:client1,token2:client2")
+    return (
+        AuthSettings(issuer_url=url, resource_server_url=url),
+        StaticTokenVerifier(tokens),
+    )
+
+
+def get_caller() -> str:
+    """工具内获取当前调用方标识（未认证时返回 anonymous）。"""
+    access_token = get_access_token()
+    return access_token.client_id if access_token else "anonymous"

mcp-for-erp/src/server.py

@@ -7,6 +7,7 @@ HTTP 模式，支持远程调用
 import asyncio
 import json
 from mcp.server import MCPServer
+from auth import get_auth
 from db import get_pool
 
 
@@ -22,10 +23,15 @@ def _s(value):
     return value
 
 
+# 鉴权：静态 Bearer Token，token → 客户端 映射来自环境变量 MCP_AUTH_TOKENS
+_auth, _token_verifier = get_auth(port=8001)
+
 app = MCPServer(
     name="mcp-for-erp-auto",
     description="汽车零部件智能报价 ERP 数据服务",
-    version="1.0.0"
+    version="1.0.0",
+    auth=_auth,
+    token_verifier=_token_verifier,
 )