使用 Logto 为你的 MCP 驱动应用启用认证 (Authentication)
本指南将带你集成 Logto 与 MCP 服务器,使用 mcp-auth 实现用户认证 (Authentication) 并通过标准 OpenID Connect 流程安全获取其身份信息。
你将学会:
- 将 Logto 配置为 MCP 服务器的授权服务器 (Authorization)。
- 在 MCP 服务器中设置 “whoami” 工具,用于返回当前用户的身份声明 (Claims)。
- 使用 MCP Inspector(MCP 客户端)测试整个流程。
完成本教程后,你的 MCP 服务器将能够:
- 在你的 Logto 租户中认证 (Authentication) 用户。
- 在调用 "whoami" 工具时返回身份声明 (Claims)(如
sub
、username
、name
、email
等)。
集成完成后,你可以用你自己的 MCP 客户端(如 Web 应用)替换 MCP Inspector,以访问 MCP 服务器暴露的工具和资源。
前置条件
- 一个 Logto Cloud(或自托管)租户
- Node.js 或 Python 环境
理解架构
- MCP 服务器:向 MCP 客户端开放工具和资源的服务器。
- MCP 客户端:用于发起认证 (Authentication) 流程并测试集成的客户端。本指南中我们将使用 MCP Inspector 作为客户端。
- Logto:作为 OpenID Connect 提供方(授权 (Authorization) 服务器),管理用户身份。
下面的非规范时序图展示了整个流程:
由于 MCP 仍在快速迭代,上述流程图可能不是最新。请参考 mcp-auth 文档获取最新信息。
在 Logto 中设置应用
- 登录到你的 Logto 控制台。
- 前往 应用程序 → 创建应用程序 → 不使用框架创建应用。
- 选择类型:单页应用。
- 填写应用名称及其他必填项,然后点击 创建应用程序。
- 保存并复制 App ID 和 Issuer endpoint(发行者端点)。
设置 MCP 服务器
创建项目并安装依赖
- Python
- Node.js
mkdir mcp-server
cd mcp-server
uv init # 或使用你自己的项目结构
uv add "mcp[cli]" starlette uvicorn mcpauth # 或使用你喜欢的包管理器
mkdir mcp-server
cd mcp-server
npm init -y
npm install @modelcontextprotocol/sdk express mcp-auth # 或使用你喜欢的包管理器
使用 Logto 配置 MCP 认证 (Authentication)
记得将 <your-logto-issuer-endpoint>
替换为你之前复制的发行者 (Issuer) 端点。
- Python
- Node.js
在 whoami.py
中:
from mcpauth import MCPAuth
from mcpauth.config import AuthServerType
from mcpauth.utils import fetch_server_config
auth_issuer = '<your-logto-issuer-endpoint>'
auth_server_config = fetch_server_config(auth_issuer, type=AuthServerType.OIDC)
mcp_auth = MCPAuth(server=auth_server_config)
在 whoami.js
中:
import { MCPAuth, fetchServerConfig } from 'mcp-auth';
const authIssuer = '<your-logto-issuer-endpoint>';
const mcpAuth = new MCPAuth({
server: await fetchServerConfig(authIssuer, { type: 'oidc' }),
});
实现令牌验证
由于我们需要验证访问令牌 (Access token) 并获取用户信息,因此需要如下实现访问令牌 (Access token) 验证:
- Python
- Node.js
import requests
from mcpauth.types import AuthInfo
def verify_access_token(token: str) -> AuthInfo:
endpoint = auth_server_config.metadata.userinfo_endpoint
response = requests.get(
endpoint,
headers={"Authorization": f"Bearer {token}"},
)
response.raise_for_status()
data = response.json()
return AuthInfo(
token=token,
subject=data.get("sub"),
issuer=auth_server_config.metadata.issuer,
claims=data,
)
const verifyToken = async (token) => {
const { userinfoEndpoint, issuer } = mcpAuth.config.server.metadata;
const response = await fetch(userinfoEndpoint, {
headers: { Authorization: `Bearer ${token}` },
});
if (!response.ok) throw new Error('Token verification failed');
const userInfo = await response.json();
return {
token,
issuer,
subject: userInfo.sub,
claims: userInfo,
};
};
实现 "whoami" 工具
现在,让我们实现 "whoami" 工具,它会使用客户端发送的访问令牌 (Access token) 请求 userinfo 端点,并返回当前用户的身份声明 (Claims)。
由于当前 SDK 版本尚未正式支持 Streamable HTTP 传输,本示例使用 SSE 传输。理论上,你可以使用任何兼容 HTTP 的传输方式。
- Python
- Node.js
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware import Middleware
mcp = FastMCP("WhoAmI")
@mcp.tool()
def whoami() -> dict:
"""
返回当前用户的身份信息。
"""
return (
mcp_auth.auth_info.claims
if mcp_auth.auth_info
else {"error": "Not authenticated"}
)
bearer_auth = Middleware(mcp_auth.bearer_auth_middleware(verify_access_token))
app = Starlette(
routes=[
mcp_auth.metadata_route(), # 提供 OIDC 元数据用于发现
Mount('/', app=mcp.sse_app(), middleware=[bearer_auth]),
],
)
运行服务器:
uvicorn whoami:app --host 0.0.0.0 --port 3001
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import express from 'express';
// 创建 MCP 服务器并注册 whoami 工具
const server = new McpServer({ name: 'WhoAmI', version: '0.0.0' });
server.tool('whoami', ({ authInfo }) => ({
content: [
{ type: 'text', text: JSON.stringify(authInfo?.claims ?? { error: 'Not authenticated' }) },
],
}));
// Express 应用 & MCP Auth 中间件
const app = express();
app.use(mcpAuth.delegatedRouter());
app.use(mcpAuth.bearerAuth(verifyToken));
// SSE 传输(如 SDK 文档所示)
const transports = {};
app.get('/sse', async (_req, res) => {
const transport = new SSEServerTransport('/messages', res);
transports[transport.sessionId] = transport;
res.on('close', () => delete transports[transport.sessionId]);
await server.connect(transport);
});
app.post('/messages', async (req, res) => {
const sessionId = String(req.query.sessionId);
const transport = transports[sessionId];
if (transport) await transport.handlePostMessage(req, res, req.body);
else res.status(400).send('No transport found for sessionId');
});
app.listen(3001);
运行服务器:
node whoami.js
测试集成
-
启动 MCP 服务器
-
启动 MCP Inspector。
由于当前 MCP Inspector 实现的限制,我们需要使用 mcp-auth 的 fork 版本:
git clone https://github.com/mcp-auth/inspector.git
cd inspector
npm install
npm run dev然后,打开终端中显示的 URL。
-
在 MCP Inspector 中:
- Transport Type:
SSE
- URL:
http://localhost:3001/sse
- OAuth Client ID:粘贴你的 Logto App ID
- Auth Params:
{"scope": "openid profile email"}
- Redirect URI:此 URL 应会自动填充。复制它。
- Transport Type:
-
在 Logto 控制台中找到你之前创建的应用,打开详情页,并将 redirect URI 粘贴到 设置 / Redirect URIs 部分。保存更改。
-
回到 MCP Inspector,点击 Connect。这将跳转到 Logto 登录体验。
完成登录后,你应该会被重定向回 MCP Inspector。进入 Tools -> List Tools -> whoami -> Run Tool。
你应该会看到类似如下的用户声明 (Claims):
{
"sub": "user_XXXX",
"username": "alice",
"name": "Alice Smith",
"email": "[email protected]"
}
- Python
- Node.js
完整的 MCP 服务器代码可以在 mcp-auth/python 仓库中找到。
完整的 MCP 服务器代码可以在 mcp-auth/js 仓库中找到。