MCP에서 인증 이해하기
MCP에서 인증 이해하기 (Understanding Authorization in MCP)
MCP 서버의 민감한 리소스와 작업을 보호하기 위해 OAuth 2.1로 안전한 인증을 구현하는 방법을 배워 보세요. MCP 서버가 사용자 데이터나 관리 작업을 다룬다면 인증을 통해 허용된 사용자만 접근하도록 해야 합니다.
출처: 문서
본문
Model Context Protocol(MCP)에서 인증은 MCP 서버가 노출하는 민감한 리소스와 작업에 대한 접근을 보호합니다. MCP 서버가 사용자 데이터나 관리 동작을 다룬다면, 인증은 허용된 사용자만 그 엔드포인트에 접근할 수 있게 합니다.
MCP는 표준화된 인증 흐름을 사용해 MCP 클라이언트와 MCP 서버 사이의 신뢰를 구축해요. 그 설계는 특정 인증·정체성 시스템 하나에 집중하지 않고, OAuth 2.1에 명시된 관례를 따릅니다. 자세한 내용은 인증 사양을 참고하세요.
인증을 언제 사용해야 하나요?
MCP 서버의 인증은 선택 사항이지만, 다음 경우 강력히 권장됩니다.
- 서버가 사용자별 데이터(이메일, 문서, 데이터베이스)에 접근할 때
- 누가 어떤 작업을 수행했는지 감사해야 할 때
- 사용자 동의를 요구하는 API 접근을 서버가 부여할 때
- 엄격한 접근 통제가 있는 엔터프라이즈 환경을 위해 구축할 때
- 사용자별 속도 제한이나 사용량 추적을 구현하고 싶을 때
로컬 MCP 서버의 인증
STDIO 전송을 사용하는 MCP 서버는 대신 환경 기반 자격 증명이나, MCP 서버에 직접 내장된 제3자 라이브러리가 제공하는 자격 증명을 사용할 수 있어요. STDIO 기반 MCP 서버는 로컬에서 실행되므로, 사용자 자격 증명을 획득하는 데 브라우저 내 인증·허가 흐름에 의존할 수도 있고 그렇지 않을 수도 있는 다양한 유연한 옵션에 접근할 수 있기 때문이죠.
반면 OAuth 흐름은 MCP 서버가 원격 호스팅되고 클라이언트가 OAuth로 사용자가 그 원격 서버에 접근 권한이 있음을 확립하는 HTTP 기반 전송용으로 설계되었어요.
인증 흐름: 단계별
클라이언트가 보호된 MCP 서버에 연결하려고 할 때 무슨 일이 일어나는지 살펴볼게요.
1단계: 초기 핸드셰이크
MCP 클라이언트가 처음 연결을 시도하면, 서버는 401 Unauthorized로 응답하고 클라이언트에게 인증 정보를 어디서 찾을지 알려줘요. 이는 Protected Resource Metadata(PRM) 문서로 캡처됩니다. 문서는 MCP 서버가 호스팅하고 예측 가능한 경로 패턴을 따르며, WWW-Authenticate 헤더의 resource_metadata 매개변수로 클라이언트에 제공됩니다.
HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer realm="mcp",
resource_metadata="https://your-server.com/.well-known/oauth-protected-resource"
이것은 해당 MCP 서버에 인증이 필요하고, 인증 흐름을 시작하는 데 필요한 정보를 어디서 얻을 수 있는지 클라이언트에 알려 줍니다.
2단계: 보호 리소스 메타데이터 발견
PRM 문서로의 URI 포인터를 받고 나면, 클라이언트는 메타데이터를 가져와 인증 서버, 지원 스코프, 기타 리소스 정보를 배웁니다. 데이터는 보통 아래와 비슷한 JSON 블롭으로 캡슐화됩니다.
{
"resource": "https://your-server.com/mcp",
"authorization_servers": ["https://auth.your-server.com"],
"scopes_supported": ["mcp:tools", "mcp:resources"]
}
더 포괄적인 예시는 RFC 9728 Section 3.2에서 확인할 수 있어요.
3단계: 인증 서버 발견
다음으로 클라이언트는 인증 서버의 메타데이터를 가져와 인증 서버가 무엇을 할 수 있는지 발견합니다. PRM 문서가 두 개 이상의 인증 서버를 나열하면, 클라이언트는 어느 것을 사용할지 결정할 수 있어요.
인증 서버를 선택한 뒤 클라이언트는 표준 메타데이터 URI를 구성하고 OpenID Connect(OIDC) Discovery 또는 OAuth 2.0 Auth Server Metadata 엔드포인트(인증 서버 지원에 따라)에 요청해, 인증 흐름을 완료하는 데 필요한 엔드포인트를 알려 주는 또 다른 메타데이터 속성 집합을 검색합니다.
{
"issuer": "https://auth.your-server.com",
"authorization_endpoint": "https://auth.your-server.com/authorize",
"token_endpoint": "https://auth.your-server.com/token",
"registration_endpoint": "https://auth.your-server.com/register"
}
4단계: 클라이언트 등록
모든 메타데이터를 치운 뒤 클라이언트는 이제 인증 서버에 등록돼 있는지 확인해야 합니다. 이는 두 가지 방법으로 할 수 있어요.
첫째, 클라이언트는 주어진 인증 서버에 사전 등록될 수 있으며, 이 경우 인증 흐름을 완료하는 데 사용하는 내장 클라이언트 등록 정보를 지닐 수 있어요.
대안으로 클라이언트는 Dynamic Client Registration(DCR) 을 사용해 인증 서버에 동적으로 등록할 수 있습니다. 후자의 시나리오는 인증 서버가 DCR을 지원해야 합니다. 인증 서버가 DCR을 지원한다면, 클라이언트는 registration_endpoint에 자신의 정보와 함께 요청을 보내요.
{
"client_name": "My MCP Client",
"redirect_uris": ["http://localhost:3000/callback"],
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"]
}
등록이 성공하면 인증 서버는 클라이언트 등록 정보가 담긴 JSON 블롭을 반환합니다.
DCR 또는 사전 등록 없음
MCP 클라이언트가 DCR을 지원하지 않는 인증 서버를 사용하는 MCP 서버에 연결하고 그 인증 서버에 사전 등록되지 않은 경우, 클라이언트 개발자가 최종 사용자가 클라이언트 정보를 수동으로 입력할 수 있는 기능을 제공할 책임이 있습니다.
5단계: 사용자 인증
클라이언트는 이제 브라우저를 /authorize 엔드포인트로 열어야 하며, 여기서 사용자가 로그인하고 필요한 권한을 부여할 수 있습니다. 인증 서버는 그런 다음 클라이언트가 토큰으로 교환하는 인증 코드와 함께 리다이렉트해요.
{
"access_token": "eyJhbG...NiIs...",
"refresh_token": "def502...",
"token_type": "Bearer",
"expires_in": 3600
}
액세스 토큰은 클라이언트가 MCP 서버에 요청을 인증하는 데 사용하는 것입니다. 이 단계는 표준 OAuth 2.1 authorization code with PKCE 관례를 따릅니다.
6단계: 인증된 요청 만들기
마지막으로 클라이언트는 Authorization 헤더에 액세스 토큰을 담아 MCP 서버에 요청을 보낼 수 있습니다.
GET /mcp HTTP/1.1
Host: your-server.com
Authorization: Bearer eyJhbG...s...
MCP 서버는 토큰을 검증해야 하며, 토큰이 유효하고 필요한 권한이 있으면 요청을 처리합니다.
구현 예시
실용적인 구현으로 시작하기 위해, Docker 컨테이너에 호스팅된 Keycloak 인증 서버를 사용할게요. Keycloak은 로컬에서 테스트와 실험을 위해 쉽게 배포할 수 있는 오픈소스 인증 서버입니다.
Docker Desktop을 다운로드해 설치해야 합니다. 개발 머신에 Keycloak을 배포하는 데 필요해요.
Keycloak 설정
터미널 애플리케이션에서 다음 명령을 실행해 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
이 명령은 Keycloak 컨테이너 이미지를 로컬로 가져오고 기본 구성을 부트스트랩합니다. 포트 8080에서 실행되며 패스워드 admin의 admin 사용자가 있어요.
프로덕션용 아님
위 구성은 테스트와 실험에는 적합하지만, 프로덕션에서는 절대 사용해서는 안 됩니다. 신뢰성, 보안, 고가용성이 필요한 시나리오에서 인증 서버를 배포하는 방법은 Configuring Keycloak for production 가이드를 참고하세요.
브라우저에서 http://localhost:8080으로 Keycloak 인증 서버에 접근할 수 있습니다.
기본 구성으로 실행할 때 Keycloak은 Dynamic Client Registration을 포함해 MCP 서버에 필요한 많은 기능을 이미 지원합니다. 이는 다음 위치의 OIDC 구성에서 확인할 수 있어요.
http://localhost:8080/realms/master/.well-known/openid-configuration
또한 Keycloak이 우리 스코프를 지원하고 우리 호스트(로컬 머신)가 클라이언트를 동적으로 등록할 수 있도록 설정해야 해요. 기본 정책이 익명 동적 클라이언트 등록을 제한하기 때문입니다.
Keycloak 대시보드에서 Client scopes로 가서 새 mcp:tools 스코프를 만드세요. MCP 서버의 모든 도구에 접근하는 데 사용할 것입니다.
스코프를 만든 후에는 그 유형을 Default로 지정하고 Include in token scope 스위치를 켜야 해요. 토큰 검증에 필요합니다.
이제 Keycloak이 발급하는 토큰의 audience도 설정해 봅시다. audience는 발급된 액세스 토큰에 의도된 목적지를 직접 내장하기 때문에 구성하는 것이 중요해요. 이렇게 하면 MCP 서버가 받은 토큰이 실제로 자신을 위한 것임을, 다른 API를 위한 것이 아니라 검증할 수 있습니다. 이는 토큰 패스스루 시나리오를 피하는 데 핵심입니다.
이를 위해 mcp:tools 클라이언트 스코프를 열고 Mappers를 클릭한 다음 Configure a new mapper를 클릭하세요. Audience를 선택합니다.
Name에는 audience-config를 사용하세요. Included Custom Audience에 http://localhost:3000 값을 추가합니다. 이것이 테스트 서버의 URI가 될 것입니다.
프로덕션용 아님
위의 audience 구성은 테스트용입니다. 프로덕션 시나리오에서는 발급된 토큰에 대해 audience가 적절히 제한되도록 추가 설정과 구성이 필요합니다. 구체적으로, audience는 클라이언트가 전달한 resource 매개변수에 기반해야 하며 고정 값이 아니어야 합니다.
이제 Clients, Client registration, Trusted Hosts로 이동하세요. Client URIs Must Match 설정을 끄고 테스트하는 호스트를 추가합니다. 현재 호스트 IP는 Linux나 macOS에서 ifconfig, Windows에서 ipconfig 명령으로 얻을 수 있어요. 추가해야 할 IP 주소는 keycloak 로그에서 Failed to verify remote host : 192.168.215.1 같은 줄을 보면 확인할 수 있습니다. IP 주소가 호스트와 연결되어 있는지 확인하세요. Docker 설정에 따라 브리지 네트워크일 수 있습니다.
호스트 얻기
Keycloak을 컨테이너에서 실행하고 있다면, 컨테이너 로그의 Terminal에서도 호스트 IP를 볼 수 있습니다.
마지막으로, token introspection 같은 일을 위해 Keycloak과 통신할 MCP 서버 자체에 사용할 새 클라이언트를 등록해야 합니다. 그러려면:
- Clients로 가세요.
- Create client를 클릭하세요.
- 클라이언트에 고유한 Client ID를 부여하고 Next를 클릭하세요.
- Client authentication을 켜고 Next를 클릭하세요.
- Save를 클릭하세요.
토큰 인트로스펙션은 토큰을 검증할 수 있는 여러 접근 방식 중 하나일 뿐이라는 점이 주목할 만해요. 이는 언어·플랫폼별 독립형 라이브러리로도 할 수 있습니다.
클라이언트 상세 정보를 열면 Credentials로 가서 Client Secret을 기록해 두세요.
비밀 처리
클라이언트 자격 증명을 코드에 직접 내장하지 마세요. 환경 변수나 전용 비밀 저장 솔루션을 권장합니다.
Keycloak이 구성되면, 인증 흐름이 트리거될 때마다 MCP 서버는 다음과 같은 토큰을 받습니다.
eyJhbG...kGYg
디코딩하면 다음과 같이 보입니다.
{
"alg": "RS256",
"typ": "JWT",
"kid": "5N710l5YnLZMwXfuVRJXkBKvY36sorgDnlrirgkeLys"
}.{
"exp": 1755540817,
"iat": 1755540757,
"auth_time": 1755538888,
"jti": "onrtac:b34080ff-8404-6867-81be-1231b50593a8",
"iss": "http://localhost:8080/realms/master",
"aud": "http://localhost:3000",
"sub": "33ed6c6b-c6e0-4928-a161-f2f69c7a03b9",
"typ": "Bearer",
"azp": "7975a5b6-8b59-4a85-9cba-8faebdab8974",
"sid": "8f7ec276-358f-4ccc-b313-db08290f376f",
"scope": "mcp:tools"
}.[Signature]
내장 Audience
토큰에 내장된
audclaim을 주목하세요. 현재 테스트 MCP 서버의 URI로 설정되어 있고, 이전에 구성한 스코프에서 유추됩니다. 이는 구현에서 검증할 때 중요합니다.
MCP 서버 설정
이제 로컬에서 실행되는 Keycloak 인증 서버를 사용하도록 MCP 서버를 설정할게요. 프로그래밍 언어 선호에 따라 지원되는 MCP SDK 중 하나를 사용할 수 있습니다.
테스트 목적으로 더하기 하나, 곱하기 하나, 두 도구를 노출하는 아주 간단한 MCP 서버를 만들게요. 서버는 이에 접근하려면 인증을 요구합니다.
TypeScript
완전한 TypeScript 프로젝트는 샘플 저장소에서 볼 수 있어요.
아래 코드를 실행하기 전에 다음 내용의 .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=<YOUR_SERVER_CLIENT_ID>
OAUTH_CLIENT_SECRET=<YOUR_SERVER_CLIENT_SECRET>
OAUTH_CLIENT_ID와 OAUTH_CLIENT_SECRET은 앞서 만든 MCP 서버 클라이언트와 연결됩니다.
MCP 인증 사양을 구현하는 것 외에도, 아래 서버는 Keycloak을 통한 토큰 인트로스펙션을 수행해 클라이언트가 보낸 토큰이 유효한지 확인합니다. 문제를 쉽게 진단할 수 있도록 기본 로깅도 구현합니다.
import "dotenv/config";
import express from "express";
import { randomUUID } from "node:crypto";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";
import cors from "cors";
import {
mcpAuthMetadataRouter,
getOAuthProtectedResourceMetadataUrl,
} from "@modelcontextprotocol/sdk/server/auth/router.js";
import { requireBearerAuth } from "@modelcontextprotocol/sdk/server/auth/middleware/bearerAuth.js";
import { OAuthMetadata } from "@modelcontextprotocol/sdk/shared/auth.js";
import { checkResourceAllowed } from "@modelcontextprotocol/sdk/shared/auth-utils.js";
const CONFIG = {
host: process.env.HOST || "localhost",
port: Number(process.env.PORT) || 3000,
auth: {
host: process.env.AUTH_HOST || process.env.HOST || "localhost",
port: Number(process.env.AUTH_PORT) || 8080,
realm: process.env.AUTH_REALM || "master",
clientId: process.env.OAUTH_CLIENT_ID || "mcp-server",
clientSecret: process.env.OAUTH_CLIENT_SECRET || "",
},
};
function createOAuthUrls() {
const authBaseUrl = new URL(
`http://${CONFIG.auth.host}:${CONFIG.auth.port}/realms/${CONFIG.auth.realm}/`,
);
return {
issuer: authBaseUrl.toString(),
introspection_endpoint: new URL(
"protocol/openid-connect/token/introspect",
authBaseUrl,
).toString(),
authorization_endpoint: new URL(
"protocol/openid-connect/auth",
authBaseUrl,
).toString(),
token_endpoint: new URL(
"protocol/openid-connect/token",
authBaseUrl,
).toString(),
};
}
function createRequestLogger() {
return (req: any, res: any, next: any) => {
const start = Date.now();
res.on("finish", () => {
const ms = Date.now() - start;
console.log(
`${req.method} ${req.originalUrl} -> ${res.statusCode} ${ms}ms`,
);
});
next();
};
}
const app = express();
app.use(
express.json({
verify: (req: any, _res, buf) => {
req.rawBody = buf?.toString() ?? "";
},
}),
);
app.use(
cors({
origin: "*",
exposedHeaders: ["Mcp-Session-Id"],
}),
);
app.use(createRequestLogger());
const mcpServerUrl = new URL(`http://${CONFIG.host}:${CONFIG.port}`);
const oauthUrls = createOAuthUrls();
const oauthMetadata: OAuthMetadata = {
...oauthUrls,
response_types_supported: ["code"],
};
const tokenVerifier = {
verifyAccessToken: async (token: string) => {
const endpoint = oauthMetadata.introspection_endpoint;
if (!endpoint) {
console.error("[auth] no introspection endpoint in metadata");
throw new Error("No token verification endpoint available in metadata");
}
const params = new URLSearchParams({
token: token,
client_id: CONFIG.auth.clientId,
});
if (CONFIG.auth.clientSecret) {
params.set("client_secret", CONFIG.auth.clientSecret);
}
let response: Response;
try {
response = await fetch(endpoint, {
method: "POST",
headers: {
"Content-Type": "application/x-www-form-urlencoded",
},
body: params.toString(),
});
} catch (e) {
console.error("[auth] introspection fetch threw", e);
throw e;
}
if (!response.ok) {
const txt = await response.text();
console.error("[auth] introspection non-OK", { status: response.status });
try {
const obj = JSON.parse(txt);
console.log(JSON.stringify(obj, null, 2));
} catch {
console.error(txt);
}
throw new Error(`Invalid or expired token: ${txt}`);
}
let data: any;
try {
data = await response.json();
} catch (e) {
const txt = await response.text();
console.error("[auth] failed to parse introspection JSON", {
error: String(e),
body: txt,
});
throw e;
}
if (data.active === false) {
throw new Error("Inactive token");
}
if (!data.aud) {
throw new Error("Resource indicator (aud) missing");
}
const audiences: string[] = Array.isArray(data.aud) ? data.aud : [data.aud];
const allowed = audiences.some((a) => {
try {
return checkResourceAllowed({
requestedResource: a,
configuredResource: mcpServerUrl,
});
} catch {
// Keycloak tokens include non-URL audiences (e.g. "account", "test-client").
// Those are never our resource, so treat them as "no match" instead of crashing.
return false;
}
});
if (!allowed) {
throw new Error(
`None of the provided audiences are allowed. Expected ${mcpServerUrl}, got: ${audiences.join(", ")}`,
);
}
return {
token,
clientId: data.client_id,
scopes: data.scope ? data.scope.split(" ") : [],
expiresAt: data.exp,
};
},
};
app.use(
mcpAuthMetadataRouter({
oauthMetadata,
resourceServerUrl: mcpServerUrl,
scopesSupported: ["mcp:tools"],
resourceName: "MCP Demo Server",
}),
);
const authMiddleware = requireBearerAuth({
verifier: tokenVerifier,
requiredScopes: [],
resourceMetadataUrl: getOAuthProtectedResourceMetadataUrl(mcpServerUrl),
});
const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {};
function createMcpServer() {
const server = new McpServer({
name: "example-server",
version: "1.0.0",
});
server.registerTool(
"add",
{
title: "Addition Tool",
description: "Add two numbers together",
inputSchema: {
a: z.number().describe("First number to add"),
b: z.number().describe("Second number to add"),
},
},
async ({ a, b }) => ({
content: [{ type: "text", text: `${a} + ${b} = ${a + b}` }],
}),
);
server.registerTool(
"multiply",
{
title: "Multiplication Tool",
description: "Multiply two numbers together",
inputSchema: {
x: z.number().describe("First number to multiply"),
y: z.number().describe("Second number to multiply"),
},
},
async ({ x, y }) => ({
content: [{ type: "text", text: `${x} × ${y} = ${x * y}` }],
}),
);
return server;
}
const mcpPostHandler = async (req: express.Request, res: express.Response) => {
const sessionId = req.headers["mcp-session-id"] as string | undefined;
let transport: StreamableHTTPServerTransport;
if (sessionId && transports[sessionId]) {
transport = transports[sessionId];
} else if (!sessionId && isInitializeRequest(req.body)) {
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
onsessioninitialized: (sessionId) => {
transports[sessionId] = transport;
},
});
transport.onclose = () => {
if (transport.sessionId) {
delete transports[transport.sessionId];
}
};
const server = createMcpServer();
await server.connect(transport);
} else {
res.status(400).json({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Bad Request: No valid session ID provided",
},
id: null,
});
return;
}
await transport.handleRequest(req, res, req.body);
};
const handleSessionRequest = async (
req: express.Request,
res: express.Response,
) => {
const sessionId = req.headers["mcp-session-id"] as string | undefined;
if (!sessionId || !transports[sessionId]) {
res.status(400).send("Invalid or missing session ID");
return;
}
const transport = transports[sessionId];
await transport.handleRequest(req, res);
};
app.post("/", authMiddleware, mcpPostHandler);
app.get("/", authMiddleware, handleSessionRequest);
app.delete("/", authMiddleware, handleSessionRequest);
app.listen(CONFIG.port, CONFIG.host, () => {
console.log(`🚀 MCP Server running on ${mcpServerUrl.origin}`);
console.log(`📡 MCP endpoint available at ${mcpServerUrl.origin}`);
console.log(
`🔐 OAuth metadata available at ${getOAuthProtectedResourceMetadataUrl(mcpServerUrl)}`,
);
});
서버를 실행하면 MCP 서버 엔드포인트를 제공해 Visual Studio Code 같은 MCP 클라이언트에 추가할 수 있어요.
TypeScript로 MCP 서버를 구현하는 더 자세한 내용은 TypeScript SDK 문서를 참고하세요.
Python
완전한 Python 프로젝트는 샘플 저장소에서 볼 수 있어요.
Python 시나리오에서는 인증 상호작용을 단순화하기 위해 Python SDK의 MCPServer 클래스에 의존합니다. 이것은 Protected Resource Metadata 문서를 게시하고, 인증되지 않은 요청에 그 문서를 가리키는 WWW-Authenticate 헤더가 있는 401로 답하며, 우리가 제공하는 verifier에 모든 bearer 토큰을 건네줍니다. 인증 주변의 엔드포인트와 토큰 검증 로직 관례는 언어마다 일관되지만, 일부는 프로덕션 시나리오에서 통합하기 더 간단한 방법을 제공합니다.
실제 서버를 작성하기 전에 config.py에서 구성을 설정해야 합니다. 내용은 전적으로 로컬 서버 설정에 기반합니다.
"""Configuration settings for the MCP auth server."""
import os
class Config:
"""Configuration class that loads from environment variables with sensible defaults."""
# Server settings
HOST: str = os.getenv("HOST", "localhost")
PORT: int = int(os.getenv("PORT", "3000"))
# Auth server settings
AUTH_HOST: str = os.getenv("AUTH_HOST", "localhost")
AUTH_PORT: int = int(os.getenv("AUTH_PORT", "8080"))
AUTH_REALM: str = os.getenv("AUTH_REALM", "master")
# OAuth client settings
OAUTH_CLIENT_ID: str = os.getenv("OAUTH_CLIENT_ID", "test-client")
OAUTH_CLIENT_SECRET: str = os.getenv("OAUTH_CLIENT_SECRET", "")
# Scope required on every token
MCP_SCOPE: str = os.getenv("MCP_SCOPE", "mcp:tools")
@property
def server_url(self) -> str:
"""Build the server URL."""
return f"http://{self.HOST}:{self.PORT}"
@property
def auth_base_url(self) -> str:
"""Build the auth server base URL."""
return f"http://{self.AUTH_HOST}:{self.AUTH_PORT}/realms/{self.AUTH_REALM}/"
# Global configuration instance
config = Config()
OAUTH_CLIENT_ID과 OAUTH_CLIENT_SECRET은 앞서 만든 MCP 서버 클라이언트와 연결됩니다. 서버를 시작하기 전에 환경에 설정하세요.
서버 구현은 다음과 같습니다.
import datetime
import logging
from typing import Any
from urllib.parse import urljoin
from pydantic import AnyHttpUrl
from mcp.server import MCPServer
from mcp.server.auth.settings import AuthSettings
from .config import config
from .token_verifier import IntrospectionTokenVerifier
logger = logging.getLogger(__name__)
def create_oauth_urls() -> dict[str, str]:
"""Create OAuth URLs based on configuration (Keycloak-style)."""
auth_base_url = config.auth_base_url
return {
"issuer": auth_base_url,
"introspection_endpoint": urljoin(auth_base_url, "protocol/openid-connect/token/introspect"),
"authorization_endpoint": urljoin(auth_base_url, "protocol/openid-connect/auth"),
"token_endpoint": urljoin(auth_base_url, "protocol/openid-connect/token"),
}
def create_server() -> MCPServer:
"""Create and configure the MCP server."""
oauth_urls = create_oauth_urls()
token_verifier = IntrospectionTokenVerifier(
introspection_endpoint=oauth_urls["introspection_endpoint"],
server_url=config.server_url,
client_id=config.OAUTH_CLIENT_ID,
client_secret=config.OAUTH_CLIENT_SECRET,
)
app = MCPServer(
name="MCP Resource Server",
instructions="Resource Server that validates tokens via Authorization Server introspection",
debug=True,
token_verifier=token_verifier,
auth=AuthSettings(
issuer_url=AnyHttpUrl(oauth_urls["issuer"]),
required_scopes=[config.MCP_SCOPE],
resource_server_url=AnyHttpUrl(config.server_url),
),
)
@app.tool()
async def add_numbers(a: float, b: float) -> dict[str, Any]:
"""
Add two numbers together.
This tool demonstrates basic arithmetic operations with OAuth authentication.
Args:
a: The first number to add
b: The second number to add
"""
result = a + b
return {
"operation": "addition",
"operand_a": a,
"operand_b": b,
"result": result,
"timestamp": datetime.datetime.now().isoformat(),
}
@app.tool()
async def multiply_numbers(x: float, y: float) -> dict[str, Any]:
"""
Multiply two numbers together.
This tool demonstrates basic arithmetic operations with OAuth authentication.
Args:
x: The first number to multiply
y: The second number to multiply
"""
result = x * y
return {
"operation": "multiplication",
"operand_x": x,
"operand_y": y,
"result": result,
"timestamp": datetime.datetime.now().isoformat(),
}
return app
def main() -> int:
"""
Run the MCP Resource Server.
This server:
- Provides RFC 9728 Protected Resource Metadata
- Validates tokens via Authorization Server introspection
- Serves MCP tools requiring authentication
Configuration is loaded from config.py and environment variables.
"""
logging.basicConfig(level=logging.INFO)
oauth_urls = create_oauth_urls()
try:
mcp_server = create_server()
logger.info("Starting MCP Server on %s:%s", config.HOST, config.PORT)
logger.info("Authorization Server: %s", oauth_urls["issuer"])
mcp_server.run(
transport="streamable-http",
host=config.HOST,
port=config.PORT,
streamable_http_path="/",
)
return 0
except Exception:
logger.exception("Server error")
return 1
if __name__ == "__main__":
exit(main())
마지막으로 토큰 검증 로직은 전적으로 token_verifier.py에 위임되어, Keycloak 인트로스펙션 엔드포인트를 사용해 모든 자격 증명 아티팩트의 유효성을 검증할 수 있게 합니다.
"""Token verifier implementation using OAuth 2.0 Token Introspection (RFC 7662)."""
import logging
from typing import Any
import httpx2
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__)
class IntrospectionTokenVerifier(TokenVerifier):
"""Token verifier that uses OAuth 2.0 Token Introspection (RFC 7662)."""
def __init__(
self,
introspection_endpoint: str,
server_url: str,
client_id: str,
client_secret: str,
):
self.introspection_endpoint = introspection_endpoint
self.server_url = server_url
self.client_id = client_id
self.client_secret = client_secret
self.resource_url = resource_url_from_server_url(server_url)
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify token via introspection endpoint."""
if not self.introspection_endpoint.startswith(("https://", "http://localhost", "http://127.0.0.1")):
return None
timeout = httpx2.Timeout(10.0, connect=5.0)
limits = httpx2.Limits(max_connections=10, max_keepalive_connections=5)
async with httpx2.AsyncClient(
timeout=timeout,
limits=limits,
verify=True,
) as client:
try:
form_data = {
"token": token,
"client_id": self.client_id,
}
# Only send client_secret when one is configured
# Public clients authenticate with client_id alone.
if self.client_secret:
form_data["client_secret"] = self.client_secret
headers = {"Content-Type": "application/x-www-form-urlencoded"}
response = await client.post(
self.introspection_endpoint,
data=form_data,
headers=headers,
)
if response.status_code != 200:
return None
data = response.json()
if not data.get("active", False):
return None
if not self._validate_resource(data):
return None
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"),
# AccessToken.resource is `str | None`. Keycloak returns `aud`
# as a *list* here (e.g. ["test-client", "http://localhost:3000",
# "account"]); passing that list straight in raises a pydantic
# ValidationError that the broad `except` below turns into a
# silent 401. We already confirmed this server's resource is a
# valid audience in `_validate_resource`, so record that.
resource=self.resource_url,
subject=data.get("sub"), # RFC 7662 subject (resource owner)
claims=data,
)
except Exception:
logger.exception("Token introspection failed")
return None
def _validate_resource(self, token_data: dict[str, Any]) -> bool:
"""Validate token was issued for this resource server.
Rules:
- Reject if 'aud' missing.
- Accept if any audience entry matches the derived resource URL.
- Supports string or list forms per JWT spec.
"""
if not self.server_url or not self.resource_url:
return False
aud: list[str] | str | None = token_data.get("aud")
if isinstance(aud, list):
return any(self._is_valid_resource(a) for a in aud)
if isinstance(aud, str):
return self._is_valid_resource(aud)
return False
def _is_valid_resource(self, resource: str) -> bool:
"""Check if the given resource matches our server."""
return check_resource_allowed(requested_resource=self.resource_url, configured_resource=resource)
더 자세한 내용은 아래 또는 Python SDK 문서를 참고하세요.
Python MCP Server
서버 루트에 pyproject.toml 파일과 mcp_server 폴더를 두세요. 모든 Python 파일을 mcp_server 폴더에 넣고, pyproject.toml을 다음과 같이 채웁니다.
[project]
name = "mcp-simple-auth"
version = "0.1.0"
description = "A simple MCP server demonstrating OAuth authentication"
requires-python = ">=3.10"
authors = [{ name = "Model Context Protocol a Series of LF Projects, LLC." }]
license = { text = "MIT" }
dependencies = [
"httpx2>=2.5.0",
"mcp>=2.0.0rc1",
"pydantic>=2.0",
]
[project.scripts]
mcp-simple-auth-rs = "mcp_server.server:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["mcp_server"]
[dependency-groups]
dev = ["pyright>=1.1.391", "pytest>=8.3.4", "ruff>=0.8.5"]
그런 다음 아래 명령을 실행해 서버를 시작합니다.
uv sync
uv run mcp-simple-auth-rs
C#
완전한 C# 프로젝트는 샘플 저장소에서 볼 수 있어요.
MCP C# SDK로 MCP 서버에 인증을 설정하려면 표준 ASP.NET Core 빌더 패턴에 기대어 사용할 수 있어요. Keycloak이 제공하는 인트로스펙션 엔드포인트 대신, 토큰 검증에 내장된 ASP.NET Core 기능을 사용할게요.
서버 폴더 루트에 Program.cs, ProtectedMcpServer.csproj 두 파일과 Tools 폴더를 만드세요. Program.cs를 채웁니다.
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.IdentityModel.Tokens;
using ModelContextProtocol.AspNetCore.Authentication;
using ProtectedMcpServer.Tools;
using System.Security.Claims;
var builder = WebApplication.CreateBuilder(args);
var serverUrl = "http://localhost:3000/";
var authorizationServerUrl = "http://localhost:8080/realms/master/";
builder.Services.AddAuthentication(options =>
{
options.DefaultChallengeScheme = McpAuthenticationDefaults.AuthenticationScheme;
options.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;
})
.AddJwtBearer(options =>
{
options.Authority = authorizationServerUrl;
var normalizedServerAudience = serverUrl.TrimEnd('/');
options.TokenValidationParameters = new TokenValidationParameters
{
ValidIssuer = authorizationServerUrl,
ValidAudiences = new[] { normalizedServerAudience, serverUrl },
AudienceValidator = (audiences, securityToken, validationParameters) =>
{
if (audiences == null) return false;
foreach (var aud in audiences)
{
if (string.Equals(aud.TrimEnd('/'), normalizedServerAudience, StringComparison.OrdinalIgnoreCase))
{
return true;
}
}
return false;
}
};
options.RequireHttpsMetadata = false; // Set to true in production
options.Events = new JwtBearerEvents
{
OnTokenValidated = context =>
{
var name = context.Principal?.Identity?.Name ?? "unknown";
var email = context.Principal?.FindFirstValue("preferred_username") ?? "unknown";
Console.WriteLine($"Token validated for: {name} ({email})");
return Task.CompletedTask;
},
OnAuthenticationFailed = context =>
{
Console.WriteLine($"Authentication failed: {context.Exception.Message}");
return Task.CompletedTask;
},
};
})
.AddMcp(options =>
{
options.ResourceMetadata = new()
{
Resource = serverUrl,
ResourceDocumentation = "https://docs.example.com/api/math",
AuthorizationServers = { authorizationServerUrl },
ScopesSupported = ["mcp:tools"]
};
});
builder.Services.AddAuthorization();
builder.Services.AddHttpContextAccessor();
builder.Services.AddMcpServer()
.WithTools<MathTools>()
.WithHttpTransport();
var app = builder.Build();
app.UseAuthentication();
app.UseAuthorization();
app.MapMcp().RequireAuthorization();
Console.WriteLine($"Starting MCP server with authorization at {serverUrl}");
Console.WriteLine($"Using Keycloak server at {authorizationServerUrl}");
Console.WriteLine($"Protected Resource Metadata URL: {serverUrl}.well-known/oauth-protected-resource");
Console.WriteLine("Exposed Math tools: Add, Multiply");
Console.WriteLine("Press Ctrl+C to stop the server");
app.Run(serverUrl);
ProtectedMcpServer.csproj를 채웁니다.
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>net9.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<!-- Identifier for the local secret store, not a secret itself. -->
<UserSecretsId>local-authorization-mcp-server</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="9.0.18" />
<PackageReference Include="ModelContextProtocol" Version="2.0.0" />
<PackageReference Include="ModelContextProtocol.AspNetCore" Version="2.0.0" />
</ItemGroup>
</Project>
Tools 폴더에 MathTools.cs를 만들고 채웁니다.
using System.ComponentModel;
using ModelContextProtocol.Server;
namespace ProtectedMcpServer.Tools;
[McpServerToolType]
public sealed class MathTools
{
[McpServerTool, Description("Add two numbers together.")]
public Task<double> Add(
[Description("First operand")] double a,
[Description("Second operand")] double b)
{
return Task.FromResult(a + b);
}
[McpServerTool, Description("Multiply two numbers together.")]
public Task<double> Multiply(
[Description("First operand")] double a,
[Description("Second operand")] double b)
{
return Task.FromResult(a * b);
}
}
그런 다음 서버 루트에서 실행합니다.
dotnet run
더 자세한 내용은 C# SDK 문서를 참고하세요.
MCP 서버 테스트하기
테스트 목적으로 Visual Studio Code를 사용할게요. 하지만 MCP와 새 인증 사양을 지원하는 어떤 클라이언트든 괜찮습니다.
Cmd + Shift + P를 누르고 **MCP: Add server...**를 선택하세요. HTTP를 선택하고 http://localhost:3000을 입력합니다. Visual Studio Code 안에서 사용할 서버에 고유한 이름을 주세요. mcp.json에서 이제 다음과 같은 항목을 볼 수 있을 것입니다.
"my-mcp-server-18676652": {
"url": "http://localhost:3000",
"type": "http"
}
연결 시 브라우저로 이동하며, Visual Studio Code가 mcp:tools 스코프에 접근하는 것에 동의하라는 프롬프트가 나타납니다.
동의한 후 mcp.json의 서버 항목 바로 위에 도구가 나열된 것을 볼 수 있습니다.
채팅 보기에서 # 기호를 사용해 개별 도구를 호출할 수 있습니다.
흔한 함정과 피하는 방법
포괄적인 보안 안내(공격 벡터, 완화 전략, 구현 모범 사례 포함)는 Security Best Practices를 꼭 읽어 보세요. 몇 가지 핵심 문제를 아래에 짚습니다.
- 토큰 검증이나 인증 로직을 직접 구현하지 마세요. 토큰 검증이나 인증 결정 같은 것에는 시판되고 잘 테스트된 안전한 라이브러리를 사용하세요. 모든 것을 처음부터 한다면 보안 전문가가 아닌 한 잘못 구현할 가능성이 더 큽니다.
- 수명이 짧은 액세스 토큰을 사용하세요. 사용하는 인증 서버에 따라 이 설정이 사용자 지정 가능할 수 있어요. 장수명 토큰을 사용하지 않는 것을 권장합니다. 악의적인 행위자가 훔치면 더 오래 접근을 유지할 수 있기 때문이죠.
- 항상 토큰을 검증하세요. 서버가 토큰을 받았다고 해서 토큰이 유효하거나 서버를 위한 것이라는 뜻은 아닙니다. MCP 서버가 클라이언트로부터 받는 것이 요구된 제약과 일치하는지 항상 확인하세요.
- 토큰을 안전하고 암호화된 저장소에 보관하세요. 어떤 시나리오에서는 서버 측에서 토큰을 캐시해야 할 수 있어요. 그렇다면 저장소가 올바른 접근 통제를 갖추고 서버에 접근하는 악의적인 당사자가 쉽게 유출할 수 없도록 하세요. 또한 견고한 캐시 제거 정책을 구현해 MCP 서버가 만료되거나 그렇지 않으면 유효하지 않은 토큰을 재사용하지 않도록 해야 합니다.
- 프로덕션에서 HTTPS를 강제하세요. 개발 중
localhost를 제외하고 평문 HTTP로 토큰을 받거나 콜백을 리다이렉트하지 마세요. - 최소 권한 스코프. 포괄적인 스코프를 쓰지 마세요. 가능한 곳에서 도구나 기능별로 접근을 나누고, 리소스 서버의 경로/도구별로 필수 스코프를 검증하세요.
- 자격 증명을 로그에 남기지 마세요.
Authorization헤더, 토큰, 코드, 비밀을 절대 로그에 남기지 마세요. 쿼리 문자열과 헤더를 정리하세요. 구조화된 로그에서 민감한 필드를 편집하세요. - 앱과 리소스 서버 자격 증명을 분리하세요. 최종 사용자 흐름에 MCP 서버의 클라이언트 시크릿을 재사용하지 마세요. 모든 비밀은 소스 컨트롤이 아닌 적절한 비밀 매니저에 저장하세요.
- 올바른 챌린지를 반환하세요. 401에서
Bearer,realm,resource_metadata로WWW-Authenticate를 포함해서 클라이언트가 인증 방법을 발견하게 하세요. - DCR(Dynamic Client Registration) 통제. 활성화하면 신뢰된 호스트, 필수 검증, 감사된 등록 같은 조직 특정 제약을 인지하세요. 인증되지 않은 DCR은 누구나 인증 서버에 어떤 클라이언트든 등록할 수 있다는 뜻입니다.
- 멀티테넌트/realm 혼동. 명시적으로 멀티테넌트가 아니면 단일 발급자/테넌트로 고정하세요. 같은 인증 서버가 서명했어도 다른 realm의 토큰을 거부하세요.
- Audience/리소스 표시자 남용.
api같은 일반 audience나 무관한 리소스를 구성하거나 수용하지 마세요. audience/리소스가 구성된 서버와 일치하도록 요구하세요. - 오류 상세 누출. 클라이언트에는 일반 메시지를 반환하고, 내부에는 내부를 노출하지 않으면서 문제 해결을 돕기 위해 상관 ID와 함께 상세한 이유를 기록하세요.
- 세션 식별자 강화.
Mcp-Session-Id를 신뢰할 수 없는 입력으로 취급하고, 절대 인증에 연결하지 마세요. 인증 변경 시 재생성하고 서버 측에서 수명주기를 검증하세요.
관련 표준과 문서
MCP 인증은 다음 잘 확립된 표준 위에 구축됩니다.
- OAuth 2.1: 핵심 인증 프레임워크
- RFC 8414: 인증 서버 메타데이터 발견
- RFC 7591: Dynamic Client Registration
- RFC 9728: Protected Resource Metadata
- RFC 8707: Resource Indicators
추가 세부 사항은 다음을 참고하세요.
이 표준들을 이해하면 인증을 올바르게 구현하고 문제가 생겼을 때 해결하는 데 도움이 됩니다.