【万字长文】Python+MCP架构实战:从零集成OAuth2.0安全认证!
本文详细讲解如何在MCP架构中集成OAuth 2.0认证,实现企业级LLM应用的安全访问控制。通过Python+MCP SDK+第三方授权的完整示例,展示了从OAuth核心概念到实战实现的完整流程。文章介绍了MCP对OAuth 2.0的支持架构,实现了Google OAuth认证集成,并讨论了与其他安全方案的配合方法,为开发者提供了可落地的安全认证解决方案。
如果你正在构建一个基于MCP架构的企业级LLM应用:各个部门都能通过客户端UI,经由共享的MCP Server访问内部工具与执行自动化任务。一个现实的问题是,你如何确保只有经过授权的员工才可以访问特定的MCP资源?注意这并非一个简单API-Key可以解决,特别是有的企业还会要求集成已有的安全体系(如企业SSO)。

本文将从基础概念入手,并通过一个Python + MCPSDK+ 第三方授权的完整示例项目,一步步说清如何在MCP架构集成OAuth安全认证。
- OAuth 2.0核心概念速览
- MCP对OAuth2.0的支持架构
- 实战:在MCP中集成Google OAuth2.0认证
- 因地制宜:与其他安全方案的配合
【完整代码地址见文末】
一、OAuth 2.0核心概念速览
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中校验,确保令牌只发给最初发起请求的客户端。
一直在更新,更多的大模型学习和面试资料已经上传带到CSDN的官方了,有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇

二、MCP对OAuth2.0的支持架构
现在我们来看MCP协议的新版本是如何将OAuth机制融入AI应用架构的,引入标准化的身份验证与授权框架,以便AI应用可以无缝对接现有的身份提供商和安全机制。
以借助第三方OAuth授权服务器,实现“用户登录 -> 获取OAuth令牌 -> 访问MCP服务”的闭环为例。首先有两个简单的映射关系:
- 资源服务器对应到现在的MCP Server
- 客户端应用对应到现在的MCP Client应用。如一个带有UI的ChatBot。
MCP规范定义的标准授权流程为:

这里的流程与上述标准流程的主要区别有两点:
- 客户端应用在第三方认证授权之前,需要首先向MCP服务器发起授权,MCP服务器负责把客户端重定向到第三方授权服务器。
- 客户端应用不直接获得第三方授权的令牌,而是由MCP Server获得(通过授权码)后;再生成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_")
#MCP Server配置
host: str = "localhost"
port: int = 3000
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:3000")
callback_path: str = "http://localhost:3000/callback"
#Google Auth配置信息
client_id: str #配置MCP_CLIENT_ID环境变量
client_secret: str #配置MCP_CLIENT_SECRET环境变量
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] = {}
# 存储用来交换的tokens和相关信息
self.tokens: Dict[str, Dict] = {}
# 存储 state 和相关信息(用来验证回调请求)
self.state_mapping: Dict[str, Dict[str, str]] = {}
剩下的就是实现必要的接口,我们将接口分个类,大致如下(详细请参考完整代码):
客户端管理接口
MCP 客户端动态注册时的接口,简单的将SDK传递来的信息保存或读取即可。
-
register_client(client_info)
- 注册新的OAuth客户端,SDK会自动创建ID和Secret,这里保存起来即可。
-
get_client(client_id)
- 根据client_id从保存的信息中获取已注册的客户端信息即可。
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
授权流程接口
-
authorize(client, params)- 处理MCP客户端授权请求,这里只要生成并返回第三方授权服务器(Google)的授权URL即可,同时维护state映射关系(state是用于后续回调时验证的码):
async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
state = secrets.token_hex(16)
#state是用来防攻击验证的一个码 ,先存起来
self.state_mapping[state] = {
"redirect_uri": str(params.redirect_uri),
"client_id": client.client_id,
"code_challenge": getattr(params, 'code_challenge', None),
}
# google oauth的url构建,返回给MCP客户端,用来重定向
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", # 不需要 refresh token
}
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")
# 用google返回的code和google交换token
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 "",
)
#存储google的返回token等,用来后续交换
self.tokens[mcp_code] = google_data
del self.state_mapping[state]
# 回调客户端的uri ,将内部生成的授权码返回
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资源/工具时验证令牌的接口。
-
load_access_token(token)
- 验证访问令牌的有效性,检查令牌是否存在、是否过期,这个方法在每次API请求时被调用,也正是该接口保证了后续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参数(设置信息)即可:
......
#传入server_setting
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:
#...取出request中的code,state信息...
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客户端进行认证,并允许后续的资源访问。如果你的帐号密码正确,很快你会看到:

- 授权成功,可以调用MCP Server
上面是MCP客户端在获得MCP Server返回的访问token后生成的提醒页面。代表授权成功,此时客户端就可以安全调用MCP Server的接口,比如查看工具列表:

- 借助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授权码流程,此时就需要考虑变通的方案,比如:
- 简单的API Key模式:常见的携带API Key的模式
- OAuth Client Credentials Flow:客户端凭证模式,无需前端用户参与,直接使用Client ID+Secret向授权服务器请求令牌,常用于两个后端服务间的安全通信
- 混合认证策略
总的来说,选择合适的安全方案需要综合考虑应用架构、安全需求和用户体验,而不是单纯依赖某一种技术方案。
五、AI大模型从0到精通全套学习大礼包
我在一线互联网企业工作十余年里,指导过不少同行后辈。帮助很多人得到了学习和成长。
只要你是真心想学AI大模型,我这份资料就可以无偿共享给你学习。大模型行业确实也需要更多的有志之士加入进来,我也真心希望帮助大家学好这门技术,如果日后有什么学习上的问题,欢迎找我交流,有技术上面的问题,我是很愿意去帮助大家的!
如果你也想通过学大模型技术去帮助就业和转行,可以扫描下方链接👇👇
大模型重磅福利:入门进阶全套104G学习资源包免费分享!

01.从入门到精通的全套视频教程
包含提示词工程、RAG、Agent等技术点
02.AI大模型学习路线图(还有视频解说)
全过程AI大模型学习路线


03.学习电子书籍和技术文档
市面上的大模型书籍确实太多了,这些是我精选出来的


04.大模型面试题目详解


05.这些资料真的有用吗?
这份资料由我和鲁为民博士共同整理,鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位,在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利,同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。
所有的视频由智泊AI老师录制,且资料与智泊AI共享,相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目,无论你是小白还是有些技术基础的,这份资料都绝对能帮助你提升薪资待遇,转行大模型岗位。


智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念,通过动态追踪大模型开发、数据标注伦理等前沿技术趋势,构建起"前沿课程+智能实训+精准就业"的高效培养体系。
课堂上不光教理论,还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作,把课本知识变成真本事!

如果说你是以下人群中的其中一类,都可以来智泊AI学习人工智能,找到高薪工作,一次小小的“投资”换来的是终身受益!
应届毕业生:无工作经验但想要系统学习AI大模型技术,期待通过实战项目掌握核心技术。
零基础转型:非技术背景但关注AI应用场景,计划通过低代码工具实现“AI+行业”跨界。
业务赋能 突破瓶颈:传统开发者(Java/前端等)学习Transformer架构与LangChain框架,向AI全栈工程师转型。
👉获取方式:
😝有需要的小伙伴,可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓
更多推荐
所有评论(0)