MCP Server怎么做權限控制?5分鐘教會你快速給MCP 服務成功添加授權 原創
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的職責在于審查所有請求中的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??
[4] OAuth2 框架: ???https://modelcontextprotocol.info/specification/draft/basic/authorization/??
本文轉載自??AI 博物院?? 作者:longyunfeigu
