MCP Server怎么做權限控制？5分鐘教會你快速給MCP 服務成功添加授權原創

AI博物院

發布于 2025-6-5 06:15

瀏覽

0收藏

2024 年，Anthropic^{[1] 發布 Claude 3[2] 系列的同時推出了 MCP[3]（Model Context Protocol），定位更底層：不是“調函數”，而是“定義模型理解系統的結構協議”。}

在企業級應用場景下，對于一些私有環境，MCP服務器可能不需要嚴格的身份認證。但如果在企業級別進行部署，對這些接口的安全性和權限管理就顯得至關重要。MCP服務器可以通過兩種方式運行：

stdio
http + sse

這兩種運行方式的權限控制策略各不相同。在深入討論MCP服務器的權限控制之前，我們先簡單回顧一下MCP的基本原理。

MCP組成和執行流程

MCP 架構分為以下 3 部分：

客戶端：大模型應用（如 DeepSeek、ChatGPT）發起 MCP 協議請求。
服務器端：服務器端響應客戶端的請求，并查詢自己的業務實現請求處理和結果返回。

運行流程：

用戶提問 LLM。
LLM 查詢 MCP 服務列表。
找到需要調用 MCP 服務，調用 MCP 服務器端。
MCP 服務器接收到指令。
調用對應工具（如數據庫）執行。
返回結果給 LLM。

stdio模式

在 stdio 這種模式下，mcp server是作為mcp client 的一個子進程運行的。在這種模式下，可以在mcp client 中配置環境變量，mcp server讀取環境變量。

環境變量是操作系統為每個進程維護的一組鍵值對（如??PATH???、??PYTHONPATH??等），存儲在進程的內存空間中。子進程啟動時，會繼承其父進程的環境變量（如 Shell 啟動的程序會繼承 Shell 的環境變量）。

下面是cursor中mcp的配置：

{
    "mcpServers": [
      {
        "name": "my-mcp-server",
        "command": "/home/cy/Downloads/anaconda3/envs/deep-searcher/bin/python3",
        "args": ["/home/cy/Desktop/git/deep-searcher/mcp_server.py"],
        "env": {"token": "value"}
      }
    ]
}

在生產中，我們一般會把mcp server單獨部署成一個服務，所以對這個服務的權限驗證才是重點。

HTTP + SSE

2025 年 3 月發布的最新 MCP 規范引入了安全基礎，借助了廣泛使用的 OAuth2 框架^[4]。

MCP Server怎么做權限控制？5分鐘教會你快速給MCP 服務成功添加授權-AI.x社區

作為資源托管中心，MCP的職責在于審查所有請求中的Authorization頭部信息。這個頭部信息必須攜帶一個OAuth2 access_token（代表客戶端權限的令牌）。通常情況下，這個令牌是一個JWT（JSON Web Token），或者只是一串無法解讀的隨機字符。如果令牌丟失或無效（例如無法解析、已經過期、不屬于該服務器等），請求將會被駁回。
同時，作為授權中心，MCP還需要具備為客戶端安全簽發access_token的能力。在簽發令牌之前，服務器會驗證客戶端的證明，有時還需要確認訪問用戶的身份。授權中心會確定令牌的有效時間、權限范圍、目標聽眾等屬性。

在我們的內部應用體系中，token 的簽發并非由我們的 MCP 服務器完成，而是由其他服務提供。因此，MCP服務器并不完全實現了OAuth2服務器的所有功能，它主要承擔著授權校驗的職責。為了實現這一目標，我們使用了一個專門處理token相關操作的 JWT類。

# pip install python-jose[cryptography]
from jose import jwt, JWTError
class JWTManager:
    def __init__(
            self,
            secret_key: str,
            algorithm: str = "HS256",
            access_token_expire_minutes: int = 30,
            refresh_token_expire_days: int = 30
    ):
        """
        初始化JWT管理器

        Args:
            secret_key: 用于簽名JWT的密鑰
            algorithm: 加密算法，默認HS256
            access_token_expire_minutes: 訪問令牌過期時間（分鐘）
            refresh_token_expire_days: 刷新令牌過期時間（天）
        """
        self.secret_key = secret_key
        self.algorithm = algorithm
        self.access_token_expire_minutes = access_token_expire_minutes
        self.refresh_token_expire_days = refresh_token_expire_days

    def create_access_token(
            self,
            data: Dict[str, Any],
            expires_delta: Optional[timedelta] = None
    ) -> str:
        """
        創建訪問令牌

        Args:
            data: 要編碼到令牌中的數據
            expires_delta: 可選的過期時間增量

        Returns:
            str: 編碼后的JWT令牌
        """
        to_encode = data.copy()
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(minutes=self.access_token_expire_minutes)

        to_encode.update({"exp": expire})
        encoded_jwt = jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
        return encoded_jwt

    def create_refresh_token(
            self,
            data: Dict[str, Any],
            expires_delta: Optional[timedelta] = None
    ) -> str:
        """
        創建刷新令牌

        Args:
            data: 要編碼到令牌中的數據
            expires_delta: 可選的過期時間增量

        Returns:
            str: 編碼后的JWT令牌
        """
        to_encode = data.copy()
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(days=self.refresh_token_expire_days)

        to_encode.update({"exp": expire})
        encoded_jwt = jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
        return encoded_jwt

    def verify_token(self, token: str) -> Dict[str, Any]:
        """
        驗證并解碼JWT令牌

        Args:
            token: 要驗證的JWT令牌

        Returns:
            Dict[str, Any]: 解碼后的令牌數據

        Raises:
            HTTPException: 當令牌無效或過期時
        """
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except JWTError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials",
                headers={"WWW-Authenticate": "Bearer"},
            )

    def is_token_expired(self, token: str) -> bool:
        """
        檢查令牌是否過期

        Args:
            token: 要檢查的JWT令牌

        Returns:
            bool: 如果令牌過期返回True，否則返回False
        """
        try:
            payload = self.verify_token(token)
            exp = payload.get("exp")
            if exp isNone:
                returnTrue
            return datetime.utcnow() > datetime.fromtimestamp(exp)
        except HTTPException:
            returnTrue

在這個環境下，我們采用了中間件的方法。通過深入研究mcp這個package的源代碼，我們了解到其底層實際上還是利用starlette創建的app來提供服務。因此，我們需要對這個app添加自定義的中間件。

class JwtMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp, dispatch: DispatchFunction | None = None) -> None:
        super().__init__(app, dispatch)
        self.jwt_manager = JWTManager(
            secret_key="your-secret-key",
            access_token_expire_minutes=30,
            refresh_token_expire_days=30
        )

    def verify_token(self, token):
        try:
            payload = self.jwt_manager.verify_token(token)
            return payload
        except Exception:
            raise



    asyncdef dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        token = request.headers.get("Authorization")

        ifnot token:
            return JSONResponse({"error": "unauthorized"}, status_code=401)
        token = token.lstrip("Bearer").strip()
        try:
            payload = self.verify_token(token)
            logging.info(payload)
        except Exception as e:
            logging.error(e)
            return JSONResponse({"error": "unauthorized"}, status_code=401)
        else:
            response = await call_next(request)
            return response

然而，在MCP這個包的源代碼中，并沒有提供直接添加中間件的入口。因此，我們需要自行進行擴展。為了實現這一目標，我們決定繼承原有的FastMCP類來完成這個擴展。

class CustomFastMCP(FastMCP):
    def __init__(
            self, name: str | None = None, instructions: str | None = None, **settings: Any
    ):
        super().__init__(name, instructions, **settings)
        self.app = self.sse_app()

# 添加的方法
    def add_middleware(
        self,
        middleware_class
    ) -> None:
        self.app.add_middleware(middleware_class)

    asyncdef run_sse_async(self) -> None:
        """Run the server using SSE transport."""
        starlette_app = self.app

        config = uvicorn.Config(
            starlette_app,
            host=self.settings.host,
            port=self.settings.port,
            log_level=self.settings.log_level.lower(),
        )
        server = uvicorn.Server(config)
        await server.serve()

在此新類中，我們對run_sse_async和init方法進行了重寫，并添加了一個新的函數add_middleware來支持中間件的添加。現在，我們可以通過使用這個自定義的CustomFastMCP類來運行MCP server。如果請求頭中未帶有Authorization或其值錯誤，系統會返回一個錯誤信息：

{"error":"unauthorized"}

一旦正確地攜帶了Authorization，請求將正常處理。

curl -X GET "http://127.0.0.1:8000/sse" \
>      -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE3NDg5NDExMzd9.GadUzsaiJYK215KIzl1c3ACQYiwQAmWaheb2o4ZI8d8"
event: endpoint
data: /messages/?session_id=eb18beb2ff4e4e70a61066d74f59a064

: ping - 2025-06-03 08:29:47.996746+00:00

: ping - 2025-06-03 08:30:02.997993+00:00

: ping - 2025-06-03 08:30:17.998910+00:00