原创 秋山墨客 2025-08-26 09:00 江苏
MCP Server的安全问题：看完这篇就够了。
如果你正在构建一个基于MCP架构的企业级LLM应用：各个部门都能通过客户端UI，经由共享的MCP Server访问内部工具与执行自动化任务。一个现实的问题是，你如何确保只有经过授权的员工才可以访问特定的MCP资源？注意这并非一个简单API-Key可以解决，特别是有的企业还会要求集成已有的安全体系（如企业SSO）。
![]()
![]()
![]()
![]()
![]()
![]()
本文将从基础概念入手，并通过一个Python + MCP SDK + 第三方授权的完整示例项目，一步步说清如何在MCP架构集成OAuth安全认证。
【完整代码地址见文末】
OAuth 2.0是一个授权协议，允许第三方应用在未经用户透露密码的情况下，获得访问用户受保护资源的权限。其关键角色包括：
-
资源拥有者：通常指用户，即受保护资源的所有者。
-
客户端：要访问资源的第三方应用，它需要向资源拥有者请求授权（比如一个需要访问你QQ身份信息的应用）。
-
授权服务器：验证用户身份并在用户同意授权后发放授权码和令牌的服务器。
-
资源服务器：实际存放受保护的资源，它会验证令牌并根据权限返回资源。
可以把 OAuth 想象成银行的保险柜使用流程：
-
你（用户）是保险柜的主人（资源拥有者）。
-
一个朋友（客户端应用）需要临时取出你的资料（访问数据）。
-
你不会把柜子钥匙（账号密码）交给他，到银行前台（授权服务器）登记，并签发一张临时访问凭证（令牌）给他。
-
保险柜工作人员（资源服务器）认凭证给他资料，凭证过期后自动失效。
OAuth 2.0最常见的模式是授权码模式，常见于各种Web应用。其典型流程如下：
以一个需要获取你Google帐号信息实现登录的应用为例：
-
用户授权：用户点击“使用Google登录”后，应用将浏览器重定向到Google的OAuth的认证界面 。在重定向URL中，应用会附带自身ID、请求的授权范围（scope）、回调地址等信息 。
-
用户登录并同意：用户在Google授权服务器登录（如果尚未登录），并看到应用请求的权限范围，选择是否同意授权 。
-
授权码返回：若用户同意，授权服务器将浏览器重定向回先前应用提供的重定向URI，并附上一个授权码（code）。
-
交换令牌：应用拿到授权码后，向授权服务器发送后台请求，用授权码交换访问令牌（Access Token）。这一步在后台完成，对用户不可见。
-
访问资源：应用获得这个Access Token后，就可以访问你授权的部分Google账户信息（通过Google API）。
OAuth 2.1规范在此基础上要求所有应用在授权流程中使用PKCE校验码，以防止授权码被拦截利用 ：即在步骤1中客户端生成一对挑战码与验证串，授权服务器在步骤4中校验，确保令牌只发给最初发起请求的客户端。
现在我们来看MCP协议的新版本是如何将OAuth机制融入AI应用架构的，引入标准化的身份验证与授权框架，以便AI应用可以无缝对接现有的身份提供商和安全机制。
以借助第三方OAuth授权服务器，实现“用户登录 -> 获取OAuth令牌 -> 访问MCP服务”的闭环为例。首先有两个简单的映射关系：
MCP规范定义的标准授权流程为：
这里的流程与上述标准流程的主要区别有两点：
此外，MCP规范还要求MCP Server实现元数据发现接口（告诉客户端授权相关的端点），以及动态客户端注册接口，不过这些主要工作默认会由SDK“代劳”。
实战：在MCP中集成Google OAuth2.0认证
现在我们构建一个完整的示例：使用Python语言和官方MCP Python SDK，在本地运行一个MCP Server，并将其配置为使用Google的OAuth2服务进行用户认证授权。在实际生产中，你可以把OAuth Server替换成企业自己的统一授权服务。
【环境准备与OAuth配置】
-
确保已安装MCP官方的Python SDK最新版本。
-
免费注册一个Google OAuth应用，用来扮演OAuth授权服务器。
-
在Google API控制台创建OAuth客户端ID凭证，类型选择“Web应用”。添加重定向URI为MCP服务器的回调地址，比如http://localhost:3000/callback，并获取生成的客户端ID和客户端密钥，将其配置到环境变量（见下）。
服务端的主要配置信息如下：
from pydantic import AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
class ServerSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="MCP_")
host: str = "localhost"
port: int = 3000
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:3000")
callback_path: str = "http://localhost:3000/callback"
client_id: str
client_secret: str
auth_url: str = "https://accounts.google.com/o/oauth2/auth"
token_url: str = "https://oauth2.googleapis.com/token"
scope: str = "https://www.googleapis.com/auth/userinfo.email https://www.googleapis.com/auth/userinfo.profile openid"
【MCP Server端实现】
实现MCP Server的OAuth认证最主要的工作是实现OAuthAuthorizationServerProvider接口，该接口负责服务端的OAuth所有相关操作，这些接口会在OAuth授权流程中被框架所调用。这里我们实现GoogleOAuthProvider，其内部数据如下：
class GoogleOAuthProvider(OAuthAuthorizationServerProvider):
def __init__(self, settings: ServerSettings):
self.settings = settings
self.clients: Dict[str, OAuthClientInformationFull] = {}
self.auth_codes: Dict[str, AuthorizationCode] = {}
self.tokens: Dict[str, Dict] = {}
self.state_mapping: Dict[str, Dict[str, str]] = {}
剩下的就是实现必要的接口，我们将接口分个类，大致如下（详细请参考完整代码）：
客户端管理接口
MCP 客户端动态注册时的接口，简单的将SDK传递来的信息保存或读取即可。
async def get_client(self, client_id: str) -> Optional[OAuthClientInformationFull]:
return self.clients.get(client_id)
async def register_client(self, client_info: OAuthClientInformationFull):
self.clients[client_info.client_id] = client_info
授权流程接口
async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
state = secrets.token_hex(16)
self.state_mapping[state] = {
"redirect_uri": str(params.redirect_uri),
"client_id": client.client_id,
"code_challenge": getattr(params, 'code_challenge', None),
}
auth_params = {
"client_id": self.settings.client_id,
"redirect_uri": self.settings.callback_path,
"response_type": "code",
"scope": self.settings.scope,
"state": state,
"access_type": "online",
}
auth_url = f"{self.settings.auth_url}?{urllib.parse.urlencode(auth_params)}"
return auth_url
handle_callback(code, state)
– 处理第三方（Google）回调，并使用授权码向Google交换访问令牌；
– 生成MCP内部的授权码并存储对应客户信息，并重定向回客户端；
– 注意这里保存了google数据，用来后续客户端交换获得令牌；这是简化处理（直接用Google的令牌），正常流程应该生成属于MCP自己的token
async def handle_callback(self, code: str, state: str) -> str:
state_data = self.state_mapping.get(state)
if not state_data:
raise HTTPException(400, "Invalid state parameter")
token_response = await create_mcp_http_client().post(
self.settings.token_url,
data=urllib.parse.urlencode({
"client_id": self.settings.client_id,
"client_secret": self.settings.client_secret,
"code": code,
"redirect_uri": self.settings.callback_path,
"grant_type": "authorization_code",
}),
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
token_response.raise_for_status()
google_data = token_response.json()
mcp_code = f"mcp_{secrets.token_hex(16)}"
self.auth_codes[mcp_code] = AuthorizationCode(
code=mcp_code,
client_id=state_data["client_id"],
redirect_uri=state_data["redirect_uri"],
redirect_uri_provided_explicitly=True,
expires_at=time.time() + 300,
scopes=self.settings.scope.split(),
code_challenge=state_data.get("code_challenge") or "",
)
self.tokens[mcp_code] = google_data
del self.state_mapping[state]
return construct_redirect_uri(state_data["redirect_uri"], code=mcp_code, state=state)
令牌交换接口
这是客户端应用向MCP Server交换令牌时需要调用的接口。
load_authorization_code(client, authorization_code)
– 加载并验证授权码的有效性，检查是否存在、是否过期。这一步很简单，直接从保存的auth_codes中返回对应的信息即可。
exchange_authorization_code(client, authorization_code)
– 将MCP客户端传来的授权码交换为访问令牌。这里只需要从上面保存的tokens中取出code对应的google_data,然后把其中的access_token返回。
令牌验证接口
这是客户端应用调用MCP Server资源/工具时验证令牌的接口。
async def load_access_token(self, token: str) -> Optional[AccessToken]:
token_data = self.tokens.get(token)
if not token_data:
return None
if token_data.get("expires_at", 0) < time.time():
del self.tokens[token]
return None
return AccessToken(
token=token,
client_id=token_data["client_id"],
scopes=self.settings.scope.split(),
expires_at=int(token_data.get("expires_at", 0)),
)
创建与启动FastMCP Server
实现了Provider后，就可以创建Server实例并启动，只需借助于FastMCP：传入auth_server_provider参数（provider实例）与auth参数（设置信息）即可：
......
oauth_provider = GoogleOAuthProvider(settings)
auth_settings = AuthSettings(
issuer_url=settings.server_url,
client_registration_options=ClientRegistrationOptions(
enabled=True,
),required_scopes=["openid"],
)
app = FastMCP(
name="Google OAuth MCP Server",
auth_server_provider=oauth_provider,
host=settings.host,
port=settings.port,
auth=auth_settings,
)
此外，由于MCP Server需处理Google带有授权码的回调，需要增加一个路由：
@app.custom_route("/callback", methods=["GET"])
async def callback_handler(request: Request) -> Response:
try:
redirect_uri = await oauth_provider.handle_callback(code, state)
return RedirectResponse(url=redirect_uri)
...
在完成这些工作后，调用FastMCP的run方法启动server：
【MCP Client实现】
MCP OAuth客户端的核心是使用OAuthClientProvider处理完整的OAuth 2.0 Authorization Code流程，通过本地回调服务器和浏览器交互完成用户授权，最终建立安全的MCP会话连接。在MCP SDK中附有一个通用的带有OAuth认证授权的客户端，我们在其基础上简单修改（参考本文源代码）后直接使用。
1. 运行客户端
该命令行客户端启动后，你将会看到如下提示：
2. 通过Google帐号授权
在首次运行时，这时候可以看到客户端请求MCP Server进行授权，随后，MCP Server会将客户端重定向到Google，你会在浏览器看到：
输入你的Google帐号和密码，让MCP Server借助你的Google授权来对MCP客户端进行认证，并允许后续的资源访问。如果你的帐号密码正确，很快你会看到：
3. 授权成功，可以调用MCP Server
上面是MCP客户端在获得MCP Server返回的访问token后生成的提醒页面。代表授权成功，此时客户端就可以安全调用MCP Server的接口，比如查看工具列表：
4. 借助MCP Inspector直接测试服务端
除了这里的命令行客户端，也可以借助MCP Inspector做授权测试，启动inspector：
npx @modelcontextprotocol/inspector
在打开的web界面中配置好transport类型和服务端URL后，可以借助其内置的OAuth设置工具，进行一步步的认证授权过程，可以很方便的观察到每个步骤与返回信息：
这里选择“QuickOAuth Flow”，然后在下方逐步完成授权过程：
（注意其中有一步需要自行打开回调链接，并将获得的授权码拷贝到下一步）
以上演示了完整的MCP Server基于OAuth规范的认证授权过程。需要再次说明的是，这里我们直接使用Google颁发的token作为访问MCP资源的凭证，更严格的实现应该是在此处由MCP服务器签发自己的应用访问令牌，这里做了简化处理。
上述示例展示了如何在MCP架构中集成OAuth认证，实现了用户身份验证和权限控制。这种方案特别适合有UI界面的应用场景，如典型的使用MCP Server扩展LLM能力的ChatBot类应用等，用户可以通过浏览器完成授权流程，特别是你需要与现有企业身份证系统集成时。
在实际应用中，当MCP客户端应用是后台Agent应用或其他后端服务时，没有用户交互界面来完成典型的OAuth授权码流程，此时就需要考虑变通的方案，比如：
总的来说，选择合适的安全方案需要综合考虑应用架构、安全需求和用户体验，而不是单纯依赖某一种技术方案。
本文源代码：
https://github.com/pingcy/mcp-oauth-demo
推荐学习本公众号最新出版的作品：
📘《MCP原理揭秘与开发指南——构建可扩展的AI智能体》
本书基于 2025-03-26 最新 MCP 协议规范 与 1.9.0 + 版本的 SDK 编写，全面覆盖从核心设计理念、协议机制解析到MCP 开发实践与完整源码案例，帮助你从根本上理解 MCP，并掌握 SDK 的高阶开发能力。
请注意——这不是一本简单的 MCP Server使用的“工具说明书”，而是一本为MCP开发者编写的、以MCP SDK解析与实战为核心的技术指南。不仅教你用“工具”，更教你如何“打造工具”。
详情点击下方链接
识别以下名片
加入公众号交流群（请说明来意）
阅读原文
跳转微信打开
