1. 什么是 MCP 授权？ #

• 什么是 MCP 授权？

授权（Authorization）用来控制「谁可以访问什么」：只有通过身份验证、且被允许的用户，才能访问 MCP 服务器提供的敏感资源或执行敏感操作。

MCP 使用 OAuth 2.1 标准实现授权，不绑定特定身份系统，可与各类授权服务（如 Keycloak、Auth0）配合。

2. 本章你将学到 #

• 什么时候需要为 MCP 服务器加授权
• 本地服务器 vs 远程服务器在授权上的区别
• OAuth 授权流程的 6 个步骤
• 用 Keycloak + Python 搭建一个带授权的 MCP 服务器
• 常见安全陷阱及规避方法

3. 何时需要授权？ #

授权是可选的，但以下场景强烈建议使用：

3.1 本地 vs 远程：授权方式不同 #

小结：本地 STDIO 服务器授权方式灵活；远程 HTTP 服务器通常用 OAuth 建立「用户已授权」的信任。

4. 授权流程概览 #

当客户端连接受保护的 MCP 服务器时，大致经历以下步骤：

5. 授权流程：分步说明 #

5.1 初始握手：401 + 元数据地址 #

客户端首次请求时，服务器返回 401 Unauthorized，并在响应头中告诉客户端「去哪获取授权信息」：

HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer realm="mcp",
  resource_metadata="http://localhost:3000/.well-known/oauth-protected-resource"

客户端据此知道：需要授权，且元数据文档在 resource_metadata 指向的地址。

5.2 受保护资源元数据（PRM） #

客户端请求该元数据文档，获取授权服务器地址、支持的 scope 等：

{
  "resource": "http://localhost:3000/",
  "authorization_servers": [
    "http://localhost:8080/realms/master/"
  ],
  "scopes_supported": [],
  "bearer_methods_supported": [
    "header"
  ]
}

更多字段见 RFC 9728 第 3.2 节。

5.3 授权服务器发现 #

客户端根据 PRM 中的 authorization_servers，请求授权服务器的元数据（OIDC Discovery 或 OAuth 2.0 元数据），获取登录、换令牌等端点：

{
  "realm": "master",
  "public_key": "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAnD9j3I5VGfACYqjqwa2CoDjBH6EsiR/yUmcqkXxJ4Vxe5R/WLn5Pc9K4/3UxXJEReAnQjtQ93+yYwAKI+cHF1IBpQZLP3LnFkd0AtmwXyMsy2r4rYzsoBduummN1EKccm0EZgA7BD8UP3RZ58RIlWm4shnA9PXvtPrOHZIVXTtiIbkFHjzgArFPB+G7z6LMWPVsXMyaGeGqlpEaB1ef17DRSzAPksjraQdNXh3uPod608V1CuV4d2WahGydULWCtARzfvtRmQZObtIFR3pklO9UqVE0jVLibNF/qqo4IwQq8pMhkS+sSGiEMXMigFIFHmbZww5PjMIW1Hf1hT8Us0wIDAQAB",
  "token-service": "http://localhost:8080/realms/master/protocol/openid-connect",
  "account-service": "http://localhost:8080/realms/master/account",
  "tokens-not-before": 0
}

5.4 客户端注册 #

客户端需要在授权服务器上「登记」自己，有两种方式：

动态注册示例请求：

{
  "client_name": "My MCP Client",
  "redirect_uris": ["http://localhost:3000/callback"],
  "grant_types": ["authorization_code", "refresh_token"],
  "response_types": ["code"]
}

若授权服务器不支持 DCR，且客户端未预注册，则需要用户手动输入客户端信息。

5.5 用户授权 #

客户端打开浏览器，跳转到 /authorize，用户登录并同意授权。授权服务器重定向回客户端，并带上授权码；客户端用授权码换取访问令牌：

{
  "access_token": "eyJhbGciOiJSUzI1NiIs...",
  "refresh_token": "def502...",
  "token_type": "Bearer",
  "expires_in": 3600
}

5.5.1 「用户」指的是什么？ #

用户 = 真人，也就是使用 VSCode 等工具，要访问 MCP 服务器的人。

在 OAuth 2.0 里，用户是「资源所有者」（Resource Owner），即有权授权访问自己资源的人。

5.5.2 用户从哪来？ #

用户是 Keycloak 里预先创建好的账号，例如：

• 管理员在 Keycloak 管理界面创建：Users → Add user → 设置用户名、密码
• 或用户自己在 Keycloak 的自注册页面注册（若开启了自注册）

这些账号保存在 Keycloak 的用户库里，和 MCP 服务器、客户端应用是分开的。

5.5.3 用户和两个客户端的关系 #

可以这样理解三者：

5.5.4 流程中的具体关系 #

Client->>User: 引导用户授权
User->>AuthServer: 登录并授权
AuthServer-->>Client: 返回授权码

具体过程是：

1. VSCode（连接 MCP 的客户端）发现需要授权，打开浏览器，跳转到 Keycloak 的 /authorize 页面。
2. 用户（张三）在浏览器里看到 Keycloak 登录页，输入自己的用户名和密码。
3. 登录成功后，Keycloak 显示授权同意页（例如「是否允许 VSCode 访问你的 MCP 服务器？」）。
4. 用户点击「同意」后，Keycloak 重定向回 VSCode，并在 URL 里带上授权码。
5. VSCode 用授权码向 Keycloak 换取 access_token。
6. 这个 access_token 代表的是当前登录的用户，里面会包含用户身份（如 sub、username）。
7. 之后 VSCode 每次请求 MCP 服务器时，都会在请求头里带上这个 token。
8. MCP 服务器用 mcp-server 的 client_id + client_secret 调用 Keycloak 内省接口，验证 token 是否有效；验证通过后，就知道「是张三在访问」。

5.5.5 简要总结 #

一句话：用户是真人，在 Keycloak 有账号；连接 MCP 的客户端代表用户去拿 token；MCP 服务器用 mcp-server 客户端去验证这个 token。

此流程遵循 OAuth 2.1 授权码 + PKCE。

5.6 发起已认证请求 #

客户端在后续请求中，将 access_token 放在 Authorization 头中：

GET /mcp HTTP/1.1
Host: your-server.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIs...

服务器验证令牌有效性及 scope，通过后处理请求。

6. 实现示例：Keycloak + Python MCP 服务器 #

下面用 Keycloak（开源授权服务器）和 Python 搭建一个带授权的 MCP 服务器，供本地测试。

前置要求：已安装 Docker Desktop。

6.1 启动 Keycloak #

在终端执行：

docker run -p 127.0.0.1:8080:8080 -e KC_BOOTSTRAP_ADMIN_USERNAME=admin  -e KC_BOOTSTRAP_ADMIN_PASSWORD=admin   quay.io/keycloak/keycloak start-dev

• 浏览器访问 http://localhost:8080 可进入 Keycloak 管理界面
• 默认管理员：admin / admin

仅用于测试：上述配置不适合生产环境，详见 Keycloak 生产配置。

6.2 切换成中文 #

• Internationalization

6.3 创建 客户端范围 mcp:tools #

1. 进入 客户端范围 → 创建客户端作用域
2. 名称填 mcp:tools
3. 类型设为 默认，开启 包含在令牌作用域中

6.4 配置 audience（受众） #

1. 打开 mcp:tools scope → 映射 → 配置新映射
2. 选择 Audience
3. 名称：audience-config
4. 包含客户端受众：http://localhost:3000（本示例 MCP 服务器地址）

6.5 允许动态客户端注册 #

1. 进入 客户端 → 客户端注册 → Trusted Hosts
2. 关闭 客户端uri必须匹配 如果开启，所有客户端uri(重定向uri和其他)都是允许的，只要它们匹配了某个受信任的主机或域。
3. 受信任主机（如 ifconfig / ipconfig 查看，或从 Keycloak 日志中类似 Failed to verify remote host : 192.168.1.5 获取）

6.5.1 「客户端」指什么？ #

这里的客户端是 OAuth 2.0 里的 Client，不是指「用户」，而是指要访问 MCP 服务器的应用，例如：

• VSCode
• 其他支持 MCP 协议的 IDE 或工具

在授权流程中

Client->>AuthServer: 客户端注册（动态注册）
AuthServer-->>Client: 返回 client_id 等凭证

这里的 Client 就是这些应用。它们需要先在 Keycloak 里「登记」自己，拿到 client_id、client_secret 等，才能继续引导用户登录并换取 Token。

6.5.2 为什么要允许动态客户端注册？ #

两种注册方式

文档说明了两种方式：

6.5.3 为什么需要动态注册？ #

在 MCP 授权流程里，客户端一开始并不知道 Keycloak 的地址，流程是：

1. 用户配置要连接的 MCP 服务器（如 http://localhost:3000）
2. 客户端请求该 MCP 服务器
3. MCP 服务器返回 401，并在响应里给出授权服务器（Keycloak）的元数据地址
4. 客户端根据元数据才知道要去哪个 Keycloak、有哪些端点

也就是说，客户端是临时发现授权服务器的，事先没有预配置的 client_id。因此需要动态注册：在首次连接时，向 Keycloak 提交自己的信息（如 client_name、redirect_uris），由 Keycloak 自动分配 client_id 和 client_secret，然后才能继续走授权码流程。

6.5.4 如果不支持动态注册会怎样？ #

• 需要为每个客户端（VSCode、其他 IDE 等）在 Keycloak 里手动创建 Client
• 用户还要在客户端里手动填写 client_id 等
• 体验差，也不利于 MCP 这种「先连上再发现授权服务器」的模式

6.5.5 Trusted Hosts 的作用 #

动态注册如果完全放开，任何应用都能注册，存在滥用风险。所以 Keycloak 用 Trusted Hosts 做限制：

• 只有来自受信任主机（例如本机 IP 127.0.0.1 或 192.168.x.x）的注册请求才会被接受
• 文档中让你添加本机 IP，就是为了让 VSCode 等在本机运行时，能成功完成动态注册

6.5.6 简要总结 #

6.6 注册 MCP 服务器用客户端 #

MCP 服务器需要自己的客户端，用于向 Keycloak 做令牌内省（验证令牌是否有效）：

1. 客户端 → 创建客户端
2. 设置 客户端ID（如 mcp-server）
3. 启用 客户端认证
4. 保存后，在 凭证 中复制 客户端密码 配置到 .env的 OAUTH_CLIENT_SECRET =OXg1z6OunGbAVoeZXbqaf5JeRLBVcptg
5. 将客户端范围 mcp:tools 添加到 mcp-server

在 Keycloak 中，将 scope（尤其是自定义 scope）显式分配给客户端，主要是出于安全控制和权限管理的考虑。

1. 最小权限原则
   Keycloak 默认不会允许客户端请求任意 scope。客户端只能请求那些已明确分配给它的 scope，这样可以防止恶意或配置错误的客户端通过 scope 获取不应访问的资源或用户信息。

2. 协议规范要求
   OpenID Connect 和 OAuth 2.0 协议中，scope 用于限定令牌的权限范围。授权服务器（Keycloak）需要验证请求的 scope 是否在客户端注册时允许的范围内。未分配的 scope 会被视为非法请求，从而返回 invalid_scope 错误。

3. 权限与可管理性
   通过将 scope 分配给客户端，管理员可以集中管理哪些客户端有权请求哪些 scope。例如，mcp:tools 可能代表访问某个特定 API 的权限，只有某些客户端才应该拥有这个权限。这种分配机制使得权限模型清晰可控。

4. 自定义 scope 的行为定义
   在 Keycloak 中，scope 可以附带协议映射器（mapper），用于决定该 scope 被请求时应该向令牌中添加哪些 claims（如角色、属性等）。只有将 scope 分配给客户端，这些映射器才能生效，确保令牌中包含正确的信息。

5. 防止滥用
   如果没有分配机制，任何客户端都可以随意请求任何 scope，这可能导致用户信息泄露或 API 被非法访问。通过强制分配，Keycloak 确保了 scope 的使用是经过授权的。

简单来说，分配 scope 就是明确告诉 Keycloak：“这个客户端可以请求这些 scope，请允许它”。这样既遵循了协议规范，也强化了安全性。

安全：不要将 客户端密码 写进代码，应使用环境变量或密钥管理服务。

6.6.1 两种「客户端」的区别 #

这里的 「MCP 服务器用客户端」 和前面的 「连接 MCP 的客户端」 是两种不同的角色，可以这样区分：

6.6.2 MCP 服务器用客户端具体做什么？ #

当用户通过 VSCode 访问 MCP 服务器时，请求会带上：

Authorization: Bearer eyJhbGciOiJSUzI1NiIs...

MCP 服务器需要判断这个 token 是否有效、是否过期、是否被吊销。做法是调用 Keycloak 的令牌内省接口（RFC 7662）：

POST /realms/xxx/protocol/openid-connect/token/introspect
Content-Type: application/x-www-form-urlencoded

token=eyJhbGciOiJSUzI1NiIs...
client_id=mcp-server
client_secret=xxx

Keycloak 要求：只有已注册的客户端才能调用内省接口，所以调用时必须提供 client_id 和 client_secret。

因此，MCP 服务器必须在 Keycloak 里预先注册一个 Client（例如 mcp-server），拿到 client_id 和 client_secret，在每次验证 token 时用它们向 Keycloak 证明「我是有权调用内省接口的 MCP 服务器」。

6.6.3 流程中的角色关系 #

用户 → VSCode（连接 MCP 的客户端）→ 动态注册 → 引导登录 → 拿到 access_token
                                                      ↓
用户 → VSCode → 携带 access_token 请求 MCP 服务器
                                                      ↓
              MCP 服务器 → 用 mcp-server 的 client_id + client_secret
                        → 调用 Keycloak 内省接口验证 token
                        → 验证通过后处理请求

6.6.4 简要总结 #

「MCP 服务器用客户端」 是 MCP 服务器在 Keycloak 里注册的一个 Client，用来：

1. 调用 Keycloak 的令牌内省接口
2. 验证用户传来的 access_token 是否有效

它和 VSCode 这类「连接 MCP 的客户端」是不同角色：前者负责验证 token，后者负责获取 token。

6.3 MCP 服务器实现 #

6.3.1 初始化环境 #

uv init mcp-server
cd mcp-server
uv add mcp[cli] starlette uvicorn httpx python-dotenv pydantic pydantic-settings

6.3.2. main.py #

main.py

# MCP 服务器集成 Keycloak OAuth 认证
"""
MCP Server with Keycloak OAuth authentication.
入口点：启动应用。
"""

# 导入日志模块
import logging

# 导入Uvicorn服务器，用于运行ASGI应用
import uvicorn

# 导入应用工厂函数和配置相关方法
from app import create_app
from config import (
    get_oauth_protected_resource_metadata_url, # 获取受OAuth保护资源的元数据URL
    get_server_url,                           # 获取服务器URL
    load_config,                              # 加载配置
)

# 配置日志级别为INFO
logging.basicConfig(level=logging.INFO)

# 定义主函数
def main() -> None:
    # 加载配置
    config = load_config()
    # 获取服务器URL
    server_url = get_server_url(config)
    # 获取OAuth元数据URL
    metadata_url = get_oauth_protected_resource_metadata_url(server_url)

    # 打印MCP服务器运行信息
    print(f"MCP Server running on {server_url}")
    # 打印MCP接口可用信息
    print(f"MCP endpoint available at {server_url}")
    # 打印OAuth元数据信息
    print(f"OAuth metadata available at {metadata_url}")

    # 启动Uvicorn服务器，运行应用
    uvicorn.run(
        create_app(),
        host=config["host"],  # 绑定主机
        port=config["port"],  # 绑定端口
    )

# 如果当前模块为主程序，则调用main函数
if __name__ == "__main__":
    main()

6.3.3. .env #

.env

# Server host/port
HOST=localhost
PORT=3000

# Auth server location
AUTH_HOST=localhost
AUTH_PORT=8080
AUTH_REALM=master

# Keycloak OAuth client credentials
OAUTH_CLIENT_ID=mcp-server
OAUTH_CLIENT_SECRET=JnecqObTi9KJVbIvpDqAxR6Gtec3vzg0

6.3.4. app.py #

app.py

"""
应用模块：Starlette 应用、路由、中间件。
"""

# 导入 contextlib 用于生命周期管理
import contextlib
# 导入 logging 模块，用于日志记录
import logging
# 导入 time 模块，用于请求耗时统计
import time

# 导入 Starlette 应用程序类
from starlette.applications import Starlette
# 导入中间件相关类
from starlette.middleware import Middleware
# 导入 CORS 中间件
from starlette.middleware.cors import CORSMiddleware
# 导入请求对象
from starlette.requests import Request
# 导入响应对象、重定向响应
from starlette.responses import RedirectResponse, Response
# 导入路由和挂载配置
from starlette.routing import Mount, Route

# 导入配置构建方法
from config import create_oauth_urls, get_server_url, load_config
# 导入创建 MCP 服务器方法
from server import create_mcp_server

# 获取当前模块的日志对象
logger = logging.getLogger(__name__)

# 创建 Starlette 应用的工厂函数
def create_app() -> Starlette:
    """创建 Starlette 应用。"""
    # 加载配置
    config = load_config()
    # 获取服务器 URL
    server_url = get_server_url(config)
    # 根据配置生成 OAuth 相关 URL
    oauth_urls = create_oauth_urls(config)
    # 创建 MCP 服务器对象
    mcp = create_mcp_server(config, oauth_urls, server_url)

    # 定义生命周期管理函数
    @contextlib.asynccontextmanager
    async def lifespan(app: Starlette):
        # 应用启动时运行 session_manager
        async with mcp.session_manager.run():
            # 让出控制权，生命周期内执行
            yield

    # 定义 /authorize 路由处理函数，实现 Keycloak 授权跳转
    async def authorize_redirect(request: Request) -> Response:
        """重定向 /authorize 到 Keycloak 授权端点。"""
        # 获取查询参数字符串
        query = str(request.url.query) if request.url.query else ""
        # 拼接目标授权端点 URL
        target = f"{oauth_urls['authorization_endpoint']}?{query}"
        # 发送 302 跳转
        return RedirectResponse(url=target, status_code=302)

    # 创建 Starlette 应用并配置路由、中间件、生命周期等
    app = Starlette(
        routes=[
            # 配置 /authorize 路由
            Route("/authorize", authorize_redirect, methods=["GET"]),
            # 挂载 mcp 的主路由（可流式 HTTP 应用）
            Mount("/", app=mcp.streamable_http_app()),
        ],
        # 配置生命周期管理
        lifespan=lifespan,
        # 配置中间件
        middleware=[
            Middleware(
                CORSMiddleware,
                allow_origins=["*"], # 允许所有来源
                allow_methods=["GET", "POST", "DELETE", "OPTIONS"], # 允许的方法
                allow_headers=["*"], # 允许的请求头
                expose_headers=["Mcp-Session-Id"], # 允许暴露的自定义响应头
            ),
        ],
    )

    # 使用日志包装应用并返回
    return _wrap_with_logging(app)

# 定义包装应用以添加请求日志的函数
def _wrap_with_logging(app: Starlette) -> Starlette:
    """包装 app 添加请求日志。"""
    # 定义一个异步应用对象，实现 ASGI 协议
    async def logged_app(scope, receive, send):
        # 如果不是 HTTP 协议类型则直接转发
        if scope["type"] != "http":
            await app(scope, receive, send)
            return
        # 记录请求开始时间
        start = time.perf_counter()
        # 默认响应状态码 500
        status = [500]

        # 定义发送消息的包装器，提取实际的返回状态码
        async def send_wrapper(message):
            # 检查响应启动消息
            if message["type"] == "http.response.start":
                # 记录响应状态码
                status[0] = message.get("status", 500)
            # 继续发送消息
            await send(message)

        # 执行应用处理逻辑，传递包装 send
        await app(scope, receive, send_wrapper)
        # 计算请求耗时（毫秒）
        ms = (time.perf_counter() - start) * 1000
        # 记录请求方法、路径、状态码和耗时
        logger.info("%s %s -> %s %.0fms", scope.get("method", ""), scope.get("path", ""), status[0], ms)

    # 返回包装后的应用
    return logged_app

6.3.5. auth.py #

auth.py

"""
认证模块：Keycloak Token Introspection 验证器。
"""

# 导入日志模块
import logging

# 导入 httpx 用于异步 HTTP 请求
import httpx
# 导入 AccessToken 和 TokenVerifier 基类
from mcp.server.auth.provider import AccessToken, TokenVerifier
# 导入资源校验辅助方法
from mcp.shared.auth_utils import check_resource_allowed, resource_url_from_server_url

# 获取当前模块的日志对象
logger = logging.getLogger(__name__)

# 定义 Keycloak Token Introspection 验证器类，继承 TokenVerifier
class KeycloakIntrospectionTokenVerifier(TokenVerifier):
    """使用 Keycloak Token Introspection (RFC 7662) 验证 Bearer Token。"""

    # 构造函数，接收 introspection 端点、client_id、client_secret、资源服务器 URL
    def __init__(
        self,
        introspection_endpoint: str,
        client_id: str,
        client_secret: str,
        resource_server_url: str,
    ):
        # 储存 introspection 端点
        self.introspection_endpoint = introspection_endpoint
        # 储存 client_id
        self.client_id = client_id
        # 储存 client_secret
        self.client_secret = client_secret
        # 储存资源服务器 URL
        self.resource_server_url = resource_server_url
        # 解析出当前资源的 URL 格式
        self.resource_url = resource_url_from_server_url(resource_server_url)

    # 异步 token 验证方法，输入 token 字符串，返回 AccessToken 或 None
    async def verify_token(self, token: str) -> AccessToken | None:
        # 记录日志，输出部分 Token
        logger.info("verifyAccessToken %s", token[:20] + "..." if len(token) > 20 else token)

        # 如果 introspection 端点未配置，抛出异常
        if not self.introspection_endpoint:
            logger.error("[auth] no introspection endpoint in metadata")
            raise ValueError("No token verification endpoint available in metadata")

        # 构造 introspection 请求数据字典，至少包含 token 和 client_id
        data = {
            "token": token,
            "client_id": self.client_id,
        }
        # 如果配置了 client_secret，则一并加入数据
        if self.client_secret:
            data["client_secret"] = self.client_secret

        try:
            # 创建 httpx 异步客户端，超时 30 秒
            async with httpx.AsyncClient(timeout=30.0) as client:
                # 以表单格式 POST 到 introspection 端点
                response = await client.post(
                    self.introspection_endpoint,
                    data=data,
                    headers={"Content-Type": "application/x-www-form-urlencoded"},
                )
        # 请求发送失败，例如网络错误，记录日志并抛出异常
        except Exception as e:
            logger.error("[auth] introspection fetch threw %s", e)
            raise

        # 如果响应码不是 200，视为 token 无效或 introspection 失败
        if response.status_code != 200:
            # 获取响应文本
            txt = response.text
            logger.error("[auth] introspection non-OK status=%s", response.status_code)
            try:
                # 尝试解析响应，记录 info 日志
                logger.info("%s", response.json())
            except Exception:
                # 如果解析失败，直接输出文本
                logger.error("%s", txt)
            # 抛出异常，提示 token 无效或过期
            raise ValueError(f"Invalid or expired token: {txt}")

        try:
            # 尝试解析 JSON 响应体
            data = response.json()
        except Exception as e:
            # 解析失败，记录错误和内容
            txt = response.text
            logger.error("[auth] failed to parse introspection JSON error=%s body=%s", e, txt)
            raise

        # 若 introspection 返回 active: false，说明 token 失效或无效
        if data.get("active") is False:
            raise ValueError("Inactive token")

        # 获取 audience 字段（资源指示），如果不存在则抛出异常
        aud = data.get("aud")
        if not aud:
            raise ValueError("Resource indicator (aud) missing")

        # 兼容 aud 既可能是 str 也可能是 list
        audiences = [aud] if isinstance(aud, str) else aud
        # 检查至少有一个 audience 可访问本资源
        allowed = any(self._check_resource_allowed(a) for a in audiences)
        # 若都不允许，抛出异常，包含期望和实际 audience 信息
        if not allowed:
            raise ValueError(
                f"None of the provided audiences are allowed. "
                f"Expected {self.resource_server_url}, got: {', '.join(audiences)}"
            )

        # 封装并返回 AccessToken 对象
        return AccessToken(
            token=token,
            client_id=data.get("client_id", "unknown"),
            scopes=data.get("scope", "").split() if data.get("scope") else [],
            expires_at=data.get("exp"),
        )

    # audience 允许检查，判断该 resource 是否属于本服务
    def _check_resource_allowed(self, requested_resource: str) -> bool:
        """检查 token 的 audience 是否允许访问本资源服务器。"""
        return check_resource_allowed(
            requested_resource=requested_resource,
            configured_resource=self.resource_server_url,
        )

6.3.6. config.py #

config.py

"""
配置模块：从环境变量加载配置，生成 OAuth 端点 URL。
"""

# 导入 os 模块以用于环境变量获取
import os

# 尝试导入 dotenv，并加载 .env 文件中的环境变量
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    # 如果未安装 dotenv，则跳过
    pass

# 从环境变量加载服务器及认证配置
def load_config() -> dict:
    """从环境变量加载配置。"""
    # 返回一个包含主机、端口和认证详细信息的配置字典
    return {
        # MCP 服务器主机，默认 'localhost'
        "host": os.getenv("HOST", "localhost"),
        # MCP 服务器端口，默认 '3000'，并转换为整数类型
        "port": int(os.getenv("PORT", "3000")),
        # OAuth 认证相关配置
        "auth": {
            # 认证服务器主机，优先使用 AUTH_HOST，若无则使用 HOST，默认 'localhost'
            "host": os.getenv("AUTH_HOST") or os.getenv("HOST", "localhost"),
            # 认证服务器端口，默认 '8080'，并转换为整数类型
            "port": int(os.getenv("AUTH_PORT", "8080")),
            # Keycloak 域名，默认 'master'
            "realm": os.getenv("AUTH_REALM", "master"),
            # OAuth 客户端 ID，默认 'mcp-server'
            "client_id": os.getenv("OAUTH_CLIENT_ID", "mcp-server"),
            # OAuth 客户端密钥，默认空字符串
            "client_secret": os.getenv("OAUTH_CLIENT_SECRET", ""),
        },
    }

# 根据配置动态生成 Keycloak OAuth 端点的相关 URL
def create_oauth_urls(config: dict) -> dict:
    """根据配置创建 Keycloak OAuth 端点 URL。"""
    # 读取认证相关的配置
    auth = config["auth"]
    # 构造基础认证路径，例如：http://auth_host:auth_port/realms/realm/
    auth_base = f"http://{auth['host']}:{auth['port']}/realms/{auth['realm']}/"
    # 返回 OAuth 端点的 URL 字典
    return {
        # 发行者（Issuer）URL
        "issuer": auth_base,
        # Token 内省端点 URL
        "introspection_endpoint": auth_base.rstrip("/") + "/protocol/openid-connect/token/introspect",
        # 授权端点 URL
        "authorization_endpoint": auth_base.rstrip("/") + "/protocol/openid-connect/auth",
        # Token 获取端点 URL
        "token_endpoint": auth_base.rstrip("/") + "/protocol/openid-connect/token",
    }

# 获取 MCP 服务器的地址
def get_server_url(config: dict) -> str:
    """获取 MCP 服务器 URL。"""
    # 构造服务器 URL，例如：http://host:port
    return f"http://{config['host']}:{config['port']}"

# 获取 OAuth 保护资源的元数据 URL
def get_oauth_protected_resource_metadata_url(server_url: str) -> str:
    """获取 RFC 9728 Protected Resource Metadata URL。"""
    # 去掉结尾的斜杠以标准化 URL
    base = server_url.rstrip("/")
    # 返回块资源元数据的标准 URL 路径
    return f"{base}/.well-known/oauth-protected-resource"

6.3.7. server.py #

server.py

"""
MCP 服务器模块：FastMCP 实例与工具注册。
"""

from pydantic import AnyHttpUrl

from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp import FastMCP

from auth import KeycloakIntrospectionTokenVerifier


def create_mcp_server(config: dict, oauth_urls: dict, server_url: str) -> FastMCP:
    """创建并配置 MCP 服务器。"""
    token_verifier = KeycloakIntrospectionTokenVerifier(
        introspection_endpoint=oauth_urls["introspection_endpoint"],
        client_id=config["auth"]["client_id"],
        client_secret=config["auth"]["client_secret"],
        resource_server_url=server_url,
    )

    mcp = FastMCP(
        name="example-server",
        json_response=True,
        token_verifier=token_verifier,
        auth=AuthSettings(
            issuer_url=AnyHttpUrl(oauth_urls["issuer"]),
            resource_server_url=AnyHttpUrl(server_url),
            required_scopes=[],
        ),
    )

    mcp.settings.streamable_http_path = "/"
    _register_tools(mcp)
    return mcp


def _register_tools(mcp: FastMCP) -> None:
    """注册 MCP 工具。"""

    @mcp.tool()
    def add(a: float, b: float) -> str:
        """Add two numbers together."""
        return f"{a} + {b} = {a + b}"

    @mcp.tool()
    def multiply(x: float, y: float) -> str:
        """Multiply two numbers together."""
        return f"{x} × {y} = {x * y}"

6.3.8 启动服务器 #

uv run main.py

7. 测试 MCP 服务器 #

以 VS Code 为例（任意支持 MCP 授权的客户端均可）：

1. Cmd + Shift + P（Mac）或 Ctrl + Shift + P（Windows）→ 选择 MCP: Add server…
2. 选择HTTP(HTTP or Server-Sent Events)
3. 输入 URL of the MCP server 如： http://localhost:3000
4. 选择Global并确认
5. VSCode会尝试认证服务

C:\Users\Administrator\AppData\Roaming\Code\User\mcp.json

{
    "servers": {
        "my-mcp-server-f68a3e0b": {
            "url": "http://localhost:3000",
            "type": "http"
        }
    },
    "inputs": []
}

2026-03-06 14:39:21.542 [info] Starting server my-mcp-server-f68a3e0b
2026-03-06 14:39:21.542 [info] Connection state: Starting
2026-03-06 14:39:21.543 [info] Starting server from LocalProcess extension host
2026-03-06 14:39:21.544 [info] Connection state: Running
2026-03-06 14:39:21.555 [info] Discovered resource metadata at http://localhost:3000/.well-known/oauth-protected-resource
2026-03-06 14:39:21.555 [info] Using auth server metadata url: http://localhost:8080/realms/master/
2026-03-06 14:39:21.558 [info] Discovered authorization server metadata at http://localhost:8080/.well-known/oauth-authorization-server/realms/master/
2026-03-06 14:39:26.549 [info] Waiting for server to respond to `initialize` request...

http://localhost:8080/realms/master/protocol/openid-connect/auth?client_id=mcp-server&response_type=code&state=vscode%3A%2F%2Fdynamicauthprovider%2Flocalhost%253A8080%2Fauthorize%3Fnonce%253D09b06b670ed47668b0fe307e78c97ca3%2526windowId%253D1&code_challenge=dJX3PvstG_aWsn06KnlXuLSAXjm_0diILx6aabJyc5k&code_challenge_method=S256&resource=http%3A%2F%2Flocalhost%3A3000%2F&redirect_uri=https%3A%2F%2Fvscode.dev%2Fredirect

连接时，会打开浏览器，提示你同意 VS Code 访问 mcp:tools scope。同意后，在 mcp.json 中可看到该服务器及其工具。

在聊天中可通过 # 调用 add_numbers、multiply_numbers 等工具。

8. 常见安全陷阱及规避 #

完整安全指南见 安全最佳实践。

9. 相关标准与延伸阅读 #

MCP 授权基于以下标准：

延伸阅读：

• Authorization Specification
• Security Best Practices
• SDKs