成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

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

發布于 2025-6-5 06:15
瀏覽
0收藏

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

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

  1. stdio
  2. http + sse

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

MCP組成和執行流程

MCP 架構分為以下 3 部分:

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

運行流程

  1. 用戶提問 LLM。
  2. LLM 查詢 MCP 服務列表。
  3. 找到需要調用 MCP 服務,調用 MCP 服務器端。
  4. MCP 服務器接收到指令。
  5. 調用對應工具(如數據庫)執行。
  6. 返回結果給 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

總結

至此,我們已成功地為MCP服務加入了認證機制。現在,只有經過授權的用戶才能獲取我們MCP服務的接入權,進一步確保了我們服務的安全性和管理效率。盡管存在多種實現MCP服務權限認證的方法,但值得注意的是,當前MCP在這個領域的研究還處于草案階段,我們會繼續關注并優化這一塊的工作。

參考資料

[1] Anthropic: ??https://zhida.zhihu.com/search?content_id=256440304&content_type=Article&match_order=1&q=Anthropic&zhida_source=entity??

[2] Claude 3: ??https://zhida.zhihu.com/search?content_id=256440304&content_type=Article&match_order=1&q=Claude+3&zhida_source=entity??

[3] MCP: ??https://zhida.zhihu.com/search?content_id=256440304&content_type=Article&match_order=1&q=MCP&zhida_source=entity??

[4] OAuth2 框架: ???https://modelcontextprotocol.info/specification/draft/basic/authorization/??


本文轉載自??AI 博物院?? 作者:longyunfeigu


?著作權歸作者所有,如需轉載,請注明出處,否則將追究法律責任
已于2025-6-5 06:15:34修改
收藏
回復
舉報
回復
相關推薦
主站蜘蛛池模板: 日本精品视频 | 国产一区不卡 | 国产精品久久久久久52avav | 日韩精品一区二区三区中文字幕 | 99久久中文字幕三级久久日本 | www.色综合| 国产日韩欧美中文字幕 | 欧美激情网站 | 久草成人网| 三级视频国产 | 欧美美女被c| 美国一级毛片a | 91一区二区三区 | 欧美一级三级 | 亚洲成人免费观看 | 亚洲国产精品成人无久久精品 | 毛片视频网站 | 国产在线精品一区二区 | 手机看片在线播放 | 国产91精品久久久久久久网曝门 | 在线视频一区二区三区 | 91亚洲国产成人精品一区二三 | 国产日韩精品久久 | xx视频在线 | 欧美日韩国产在线 | 国产精品毛片一区二区三区 | 三级特黄特色视频 | 欧美偷偷操 | 亚洲免费在线视频 | 国产精品欧美精品日韩精品 | 欧美日韩综合 | 亚洲综合久久久 | 国产在线高清 | 美女视频一区二区三区 | 99精品久久久久久久 | 91视频在线看 | 欧美999 | 亚洲一区二区日韩 | 欧美视频在线一区 | 成人不卡视频 | 国产网站在线播放 |