如果你正在构建一个基于MCP架构的企业级LLM应用：各个部门都能通过客户端UI，经由共享的MCP Server访问内部工具与执行自动化任务。一个现实的问题是，你如何确保只有经过授权的员工才可以访问特定的MCP资源？注意这并非一个简单API-Key可以解决，特别是有的企业还会要求集成已有的安全体系（如企业SSO）。

本文将从基础概念入手，并通过一个Python + MCP SDK + 第三方授权的完整示例项目，一步步说清如何在MCP架构集成OAuth安全认证。

• OAuth 2.0核心概念速览

• MCP对OAuth2.0的支持架构

• 实战：在MCP中集成Google OAuth2.0认证

• 因地制宜：与其他安全方案的配合

【完整代码地址见文末】

01

OAuth 2.0核心概念速览

OAuth 2.0是一个授权协议，允许第三方应用在未经用户透露密码的情况下，获得访问用户受保护资源的权限。其关键角色包括：

• 资源拥有者：通常指用户，即受保护资源的所有者。

• 客户端：要访问资源的第三方应用，它需要向资源拥有者请求授权（比如一个需要访问你QQ身份信息的应用）。

• 授权服务器：验证用户身份并在用户同意授权后发放授权码和令牌的服务器。

• 资源服务器：实际存放受保护的资源，它会验证令牌并根据权限返回资源。

可以把 OAuth 想象成银行的保险柜使用流程：

• 你（用户）是保险柜的主人（资源拥有者）。

• 一个朋友（客户端应用）需要临时取出你的资料（访问数据）。

• 你不会把柜子钥匙（账号密码）交给他，到银行前台（授权服务器）登记，并签发一张临时访问凭证（令牌）给他。

• 保险柜工作人员（资源服务器）认凭证给他资料，凭证过期后自动失效。

OAuth 2.0最常见的模式是授权码模式，常见于各种Web应用。其典型流程如下：

以一个需要获取你Google帐号信息实现登录的应用为例：

1. 用户授权：用户点击“使用Google登录”后，应用将浏览器重定向到Google的OAuth的认证界面 。在重定向URL中，应用会附带自身ID、请求的授权范围（scope）、回调地址等信息 。

2. 用户登录并同意：用户在Google授权服务器登录（如果尚未登录），并看到应用请求的权限范围，选择是否同意授权 。

3. 授权码返回：若用户同意，授权服务器将浏览器重定向回先前应用提供的重定向URI，并附上一个授权码（code）。

4. 交换令牌：应用拿到授权码后，向授权服务器发送后台请求，用授权码交换访问令牌（Access Token）。这一步在后台完成，对用户不可见。

5. 访问资源：应用获得这个Access Token后，就可以访问你授权的部分Google账户信息（通过Google API）。

OAuth 2.1规范在此基础上要求所有应用在授权流程中使用PKCE校验码，以防止授权码被拦截利用 ：即在步骤1中客户端生成一对挑战码与验证串，授权服务器在步骤4中校验，确保令牌只发给最初发起请求的客户端。

02

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“代劳”。

03

实战：在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 AnyHttpUrlfrom 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客户端进行认证，并允许后续的资源访问。如果你的帐号密码正确，很快你会看到：

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服务器签发自己的应用访问令牌，这里做了简化处理。

04

因地制宜：与其他安全方案的配合

上述示例展示了如何在MCP架构中集成OAuth认证，实现了用户身份验证和权限控制。这种方案特别适合有UI界面的应用场景，如典型的使用MCP Server扩展LLM能力的ChatBot类应用等，用户可以通过浏览器完成授权流程，特别是你需要与现有企业身份证系统集成时。

在实际应用中，当MCP客户端应用是后台Agent应用或其他后端服务时，没有用户交互界面来完成典型的OAuth授权码流程，此时就需要考虑变通的方案，比如：

• 简单的API Key模式：常见的携带API Key的模式

• OAuth Client Credentials Flow：客户端凭证模式，无需前端用户参与，直接使用Client ID+Secret向授权服务器请求令牌，常用于两个后端服务间的安全通信

• 混合认证策略

总的来说，选择合适的安全方案需要综合考虑应用架构、安全需求和用户体验，而不是单纯依赖某一种技术方案。

本文源代码：

https://github.com/pingcy/mcp-oauth-demo