CLOVA Studio 운영자10 Posted November 19 공유하기 Posted November 19 들어가며 (1부) MCP 실전 쿡북: LangChain에서 네이버 검색 도구 연결하기와 (2부) MCP 실전 쿡북: 세션 관리와 컨텍스트 유지 전략에서는 MCP 서버의 기본 구현과 클라이언트로 MCP 서버를 연결하여 활용하는 방법을 다루었습니다. 이번 3부에서는 FastMCP에서 제공하는 인증 시스템을 활용해 MCP 서버를 구축하는 방법을 다룹니다. OAuth 인증이 통합된 MCP 서버 방식에서는 사용자가 처음 한 번만 네이버 로그인을 진행하면, 서버가 자동으로 토큰을 받아오게 됩니다. 사용자는 토큰이 어떻게 발급되고 유지되는지 신경 쓸 필요 없이, 단순 로그인만으로 자신의 네이버 캘린더나 카페 등 로그인 기반 서비스를 바로 사용할 수 있습니다. 토큰이 만료되더라도 MCP 서버가 자동으로 검증 및 재발급 절차를 처리하므로, 로그인 과정이 반복되지 않습니다. 이를 통해 사용자 계정 기반 외부 서비스와 HyperCLOVA X 모델 간의 안전한 연동이 가능해지고, 개인화된 기능을 더 쉽게 활용할 수 있습니다. 본 쿡북에서는 네이버 로그인 인증을 직접 구현해보고, 추가로 네이버 캘린더까지 연동한 실습 시나리오를 통해 OAuth 인증 기반 MCP 서버의 구축 방법을 단계별로 안내합니다. MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 링크 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 ✔︎ (4부) 운영과 확장을 위한 다양한 팁 링크 1. 사전 준비 사항 본 섹션에서는 인증 기반 MCP 서버를 구현하기 전에 필요한 사전 준비 사항에 대해 안내합니다. Ngrok 설치 및 등록 Ngrok은 로컬 서버를 외부에서 접근할 수 있도록 해주는 터널링 도구입니다. 본 예제에서는 서버가 로컬 환경에서 실행되기 때문에 외부 접근을 가능하게 하기 위해 Ngrok을 사용했습니다. 실제 서비스 환경에서는 Ngrock 등록 없이 운영 중인 도메인 주소를 그대로 사용해도 됩니다. 먼저 공식 사이트에서 회원가입 후, 안내에 따라 ngrok을 설치하고 발급받은 인증 토큰을 등록해야 합니다. 설치와 설정이 완료되면 터미널에서 다음 명령어를 실행하여, 로컬 8000 포트를 https://xxx.ngrok-free.app/ 형태의 공개 URL로 매핑할 수 있습니다. ngrok http 8000 발급 받은 공개 URL은 이후 환경 변수 설정에 함께 입력합니다. 네이버 개발자 센터 설정 네이버 개발자 센터 > Application > 내 애플리케이션 > Application 등록에서 다음 항목을 설정합니다. 애플리케이션 이름을 설정합니다. 사용 API: '검색','네이버 로그인','캘린더'를 추가합니다. 로그인 오픈 API 서비스 환경 'PC 웹' 환경을 추가합니다. 서비스 URL: MCP 서버의 엔드포인트 주소를 입력합니다.(ex. https://xxx.ngrok-free.app 또는 https://your-domain.com) 네이버 로그인 Callback URL: 서버 주소 뒤에 '/auth/callback'을 추가합니다.(ex. https://xxx.ngrok-free.app/auth/callback 또는 https://your-domain.com/auth/callback) 비로그인 오픈 API 서비스 환경 서비스 URL: MCP 서버의 엔드포인트 주소를 입력합니다.(ex. https://xxx.ngrok-free.app 또는 https://your-domain.com) 페이지 하단의 수정 버튼을 클릭해 설정을 완료합니다. API 호출에 필요한 클라이언트 아이디와 클라이언트 시크릿 정보를 확인합니다. 클라이언트 아이디와 클라이언트 시크릿 확인을 참고해 주세요. 네이버 개발자 센터 > 내 애플리케이션 > 멤버 관리에서 테스트 ID를 등록합니다. 애플리케이션이 '개발 중' 상태이면 멤버 관리 메뉴에서 등록한 아이디만 네이버 로그인을 이용할 수 있습니다. 관리자 아이디 외 다른 아이디로 테스트를 진행할 경우 멤버 관리 메뉴에서 테스트 ID 등록을 진행합니다. 애플리케이션이 '승인'된 상태이면 모든 아이디로 네이버 로그인을 이용할 수 있습니다. Quote 네이버 API 사용 시 네이버 개발자 센터의 이용 약관을 준수해야 합니다. 약관 준수 의무를 위반하여 발생한 모든 법적 책임은 사용자에게 있습니다. 네이버 API를 사용하는 방법에 대한 자세한 설명은 네이버 개발자 센터 API 공통 가이드를 참고해 주십시오. 프로젝트 구성 프로젝트의 전체 파일 구조는 다음과 같습니다. Python 버전은 3.10이상, 3.13 미만입니다. mcp_cookbook_part3/ ├── remote_auth_server.py # FastMCP 서버(네이버 OAuth) ├── fastmcp_client.py ├── naver.py # 네이버 OAuth Provider 구현 └── .env 환경 변수 설정 환경 변수를 설정합니다. 루트 디렉터리에 .env 파일을 생성한 뒤, 앞서 발급받은 API Key를 다음과 같이 입력하고 저장합니다. 이때 따옴표 없이 값을 작성해야 하며, VS Code에서 실행할 경우 설정에서 Use Env File 옵션이 활성화되어 있는지 확인하세요. #.env CLOVA_STUDIO_API_KEY=YOUR_API_KEY NAVER_CLIENT_ID=YOUR_CLIENT_ID NAVER_CLIENT_SECRET=YOUR_CLIENT_SECRET BASE_URL=https://xxx.ngrok-free.app # 또는 https://your-domain.com 패키지 설치 프로젝트에 필요한 패키지 목록은 아래 다운로드 링크에서 확인할 수 있습니다. 해당 내용을 복사해 루트 디렉터리에 requirements.txt 파일로 저장하세요. requirements.txt 다운로드 루트 디렉터리에서 터미널을 실행하여 다음과 같이 예제 실행에 필요한 패키지를 설치합니다. 가상환경 설치를 권장합니다. # 1. 파이썬 가상환경 생성 python -m venv .venv # 2. 가상환경 활성화 (macOS/Linux) source .venv/bin/activate # (Windows) # .venv/Scripts/activate.ps1 # 3. 패키지 설치 pip install -r requirements.txt 실행 방법 본 쿡북에서는 실행을 위해 총 3개의 터미널이 필요합니다. (Ngrok, 서버, 클라이언트) Ngrok 실행 # 1. 사전 준비 사항에서 설정한 포트로 실행 ngrok http 8000 인증 기반 MCP 서버 실행 # 2. 가상환경 활성화 source .venv/bin/activate # 3. 인증 서버 실행 python remote_auth_server.py FastMCP 클라이언트 실행 # 4. 가상환경 활성화 source .venv/bin/activate # 5. 클라이언트 실행 python fastmcp_client.py 2. MCP 인증 요구 사항 MCP 인증을 구현하기에 앞서, 인증의 기본 개념과 MCP 서버가 준수해야 할 인증 표준 가이드, 그리고 IdP 활용 방안을 소개합니다. 인증(Authentication)과 인가(Authorization) MCP 서버와 같은 외부 클라이언트 요청을 받는 시스템에서는 인증(Authentication)와 인가(Authorization) 체계를 적용해 사용자별 접근 권한을 안전하게 관리하는 것이 중요합니다. 인증(Authentication): 사용자의 신원(ex. 아이디·비밀번호, 소셜 로그인, SSO 등)을 검증하여, 요청자가 실제로 등록된 사용자임을 확인하는 과정입니다. 인가(Authorization): 인증이 완료된 사용자가 접근하려는 리소스나 기능에 대해 권한이 있는지를 판별하는 과정입니다. 이때, 사용자 신원의 확인은 OIDC와 같은 인증 계층이 담당하며, 권한 관리 및 위임 절차는 OAuth 2.0 같은 인가 프레임워크를 통해 진행됩니다. 이 두 기술은 MCP 서버의 핵심 보안 구조를 구성합니다. OIDC를 활용한 인증 계층 OIDC(OpenID Connect)는 OAuth 2.0 프로토콜 위에 구축된 인증 계층입니다. OIDC 인증의 핵심은 ID 토큰(ID Token)을 발급하는 것입니다. 이 토큰은 JWT 형식으로, 사용자의 식별자, 발급 시간, 만료 시간 등의 신원 정보를 담고 있습니다. 클라이언트는 이 ID 토큰을 통해 사용자의 신원을 확인할 수 있게 됩니다. 또한 OIDC는 UserInfo 엔드포인트를 제공해서 사용자 프로필 정보(ex. 이름, 이메일 등)를 조회할 수 있게 합니다. OAuth 2.0 인가 프레임워크 인가를 구현하는 데 가장 널리 사용되는 표준 프레임워크는 OAuth 2.0입니다. 핵심 개념은 클라이언트가 사용자를 대신해 리소스에 접근할 수 있도록 권한을 안전하게 위임하는 구조를 제공한다는 점입니다. 이 프레임워크는 다음 두 구성 요소를 중심으로 동작합니다. 인증 서버(Authorization Server): OIDC 환경에서 사용자의 신원을 확인하고, 클라이언트에게 허용된 스코프(Scope)에 따라 액세스 토큰(Access Token)을 발급할지 결정합니다. 리소스 서버(Resource Server): 실제 리소스(ex. MCP 서버)를 보유하고 있으며, 클라이언트가 전달한 액세스 토큰을 검증해 요청된 작업을 허용할지 결정합니다. MCP 인증 표준 가이드 MCP 공식 가이드(2025-06-18)에서는 견고한 인증 및 인가 환경 구축을 위해 다음 4가지 OAuth 표준을 준수하도록 명시하고 있습니다. OAuth 2.1 IETF DRAFT(draft-ietf-oauth-v2-1-13) OAuth 2.0의 보안 취약점으로 식별된 부분을 해결하고, 모범 사례로 인정된 확장 기능을 필수 사항으로 통합하여 정리한 개정판입니다. 일례로, 모든 클라이언트에 대해 PKCE(Proof Key for Code Exchange)를 필수로 적용하도록 합니다. 자세한 변경 사항은 OAuth 2.1 공식 문서에서 확인할 수 있습니다. 참고로, 현재 OAuth 2.1은 정식 RFC가 아닌 Internet Draft 단계로, 아직 표준화 과정이 진행 중입니다. 그럼에도 불구하고 업계에서는 사실상 가장 안전한 구현 지침으로 받아들여지고 있으며, MCP 서버와 같은 외부 접근 환경에서는 OAuth 2.1을 적용하는 것이 권장됩니다. OAuth 2.0 Authorization Server Metadata(RFC8414) 클라이언트가 OAuth 인증 서버와 상호작용하는 데 필요한 정보를 자동으로 얻을 수 있게 해주는 메타데이터 형식에 대한 표준입니다. 이 메타데이터에는 인증 엔드포인트, 토큰 엔드포인트, 지원하는 스코프, 지원하는 그랜트 타입(Grant Type) 등의 정보가 포함됩니다. 이 표준을 사용하면 MCP 인증 서버의 위치와 기능을 클라이언트가 동적으로 검색하고 구성할 수 있습니다. OAuth 2.0 Dynamic Client Registration Protocol(RFC7591) Dynamic Client Registration(DCR)은 클라이언트가 인증 서버에 자동으로 등록될 수 있게 해주는 표준 방식입니다. 기존에는 클라이언트(애플리케이션)을 사용하기 전, 관리자가 인증 서버에 필요한 정보를 수동으로 등록하고 client_id를 받아야 했지만, 이 표준을 사용하면 이 절차가 간소화됩니다. 클라이언트는 등록 엔드포인트에 자신의 클라이언트 메타데이터를 전송하기만 하면, 서버가 자동으로 client_id와 client_secret을 발급합니다. 즉, MCP 기반 애플리케이션의 자동화된 배포 및 확장을 지원하여, 새로운 클라이언트가 MCP 환경에 빠르게 통합될 수 있도록 합니다. OAuth 2.0 Protected Resource Metadata(RFC9728) 리소스 서버가 자신이 어떤 인증 및 인가 요구 사항을 가지고 있는지 메타데이터로 공개할 수 있도록 하는 표준입니다. 메타데이터에는 토큰 형식, 허용되는 스코프, 특별한 보안 요구 사항이 있는지 등의 정보가 포함됩니다. 이 표준을 통해 클라이언트나 인증 서버는 리소스 서버의 요구 사항을 동적으로 인식할 수 있습니다. IdP와 OAuth Proxy의 활용 MCP 서버에 OAuth 기반 인증 체계를 구축할 때, 기존의 IdP(Identity Provider)를 활용하는 것이 일반적입니다. IdP는 사용자의 신원을 확인하고 인증 서비스를 제공하는 주체입니다. 대표적으로 구글, 네이버와 같은 소셜 로그인 서비스, 그리고 Stytch, WorkOS, Auth0과 같은 기업용 SSO(Single Sign-On) 솔루션이 있습니다. 현재 OAuth 2.1은 정식 RFC 표준으로 확정되지 않은 단계이기 때문에, 대부분의 IdP는 공식적으로 OAuth 2.0을 지원합니다. 따라서 실제 구현 시에는 MCP 표준에서 요구하는 PKCE와 같은 핵심 보안 메커니즘을 지원하는 IdP로 채택하는 것이 좋습니다. 만약 선택한 IdP가 최신 보안 기능을 제공하지 않거나, DCR 또는 메타데이터 기반 동적 구성 기능을 지원하지 않는다면, FastMCP의 OAuth Proxy 같은 어댑터를 연결해 사용할 수 있습니다. FastMCP의 OAuth Proxy는 MCP 규격과 호환되지 않는 외부 로그인 서비스를 MCP 클라이언트가 기대하는 DCR 기반 인터페이스로 변환해 줍니다. 즉, 클라이언트는 MCP 표준 요청을 보내면, OAuth Proxy가 이를 IdP가 처리할 수 있는 형태로 중개하고, 응답은 다시 MCP 형식으로 재구성합니다. 3. OAuth Proxy 기반 인증 커스터마이징 이번 섹션에서는 FastMCP의 기본 코드를 기반으로 네이버 인증을 구현하는 방법을 설명합니다. OAuth Proxy 동작 흐름 FastMCP에서 제공하는 OAuth Proxy는 MCP 클라이언트와 외부 인증 서버 사이에서 어댑터처럼 동작합니다. MCP 클라이언트에는 MCP 표준 인증 서버로 보이도록 /register와 /.well-known/... 엔드포인트를 제공해주고, 네이버 인증 서버에는 미리 등록된 일반 OAuth 클라이언트로 동작하여 요청을 중계합니다. 즉, OAuth Proxy는 인증 서버가 MCP 표준을 지원하지 않아도, 클라이언트 입장에서는 MCP 표준 인증 흐름이 그대로 유지되는 것처럼 보이게 하는 역할을 합니다. PKCE(Proof Key for Code Exchange)란? PKCE는 OAuth 인증 과정에서 인가 코드가 탈취되는 것을 방지하는 보안 메커니즘입니다. 인증을 시작할 때 무작위 보안 코드(code_challenge)를 생성하여 전달하고, 토큰을 요청할 때 원본 검증 코드(code_verifier)를 함께 제출합니다. 인증 서버는 검증 코드가 처음 받은 보안 코드와 일치하는지 확인하여, 동일한 클라이언트가 요청했음을 증명합니다. 이 방식은 중간에 인가 코드를 가로채더라도 검증 코드 없이는 토큰을 발급받을 수 없어 안전합니다. OAuth Proxy의 역할 중 MCP 표준 인증 흐름을 위해 이중 PKCE 검증을 수행합니다. 첫 번째는 MCP 클라이언트와 Proxy 간의 PKCE 검증이고, 두 번째는 Proxy와 실제 OAuth Provider 간의 PKCE 검증입니다. 이를 통해 양쪽 구간 모두에서 보안이 강화됩니다. NaverProvider 구현 다음은 FastMCP에서 공식 제공하는 GoogleProvider를 커스터마이징 하여 NaverProvider를 구현한 스크립트입니다. 레퍼런스 코드는 여기에서 확인할 수 있습니다. # naver.py from __future__ import annotations import httpx from key_value.aio.protocols import AsyncKeyValue from pydantic import AnyHttpUrl, SecretStr, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict from fastmcp.server.auth import TokenVerifier from fastmcp.server.auth.auth import AccessToken from fastmcp.server.auth.oauth_proxy import OAuthProxy from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes from fastmcp.utilities.logging import get_logger from fastmcp.utilities.types import NotSet, NotSetT logger = get_logger(__name__) class NaverProviderSettings(BaseSettings): """Settings for Naver OAuth provider.""" model_config = SettingsConfigDict( env_prefix="NAVER_", env_file=ENV_FILE, extra="ignore", ) client_id: str | None = None client_secret: SecretStr | None = None base_url: AnyHttpUrl | str | None = None issuer_url: AnyHttpUrl | str | None = None redirect_path: str | None = None required_scopes: list[str] | None = None timeout_seconds: int | None = None allowed_client_redirect_uris: list[str] | None = None jwt_signing_key: str | None = None @field_validator("required_scopes", mode="before") @classmethod def _parse_scopes(cls, v): return parse_scopes(v) class NaverTokenVerifier(TokenVerifier): """Token verifier for Naver OAuth tokens. 네이버의 사용자 프로필 API를 호출하여 검증해야 합니다. 이 검증기는 네이버 API 사양에 따라 토큰 검증과 사용자 정보 추출을 처리합니다. 주요 특징: - 네이버 프로필 API (/v1/nid/me)를 사용한 토큰 검증 - resultcode/message 구조를 가진 네이버 특유의 응답 형식 처리 - 네이버에서 제공하는 포괄적인 사용자 프로필 데이터 추출 """ def __init__( self, *, required_scopes: list[str] | None = None, timeout_seconds: int = 10, ): """네이버 토큰 검증기를 초기화합니다. Args: required_scopes: 필수 OAuth 스코프. timeout_seconds: API 호출을 위한 HTTP 요청 타임아웃 (기본값: 10초) """ super().__init__(required_scopes=required_scopes) self.timeout_seconds = timeout_seconds async def verify_token(self, token: str) -> AccessToken | None: """네이버 OAuth 토큰을 검증하고 사용자 정보를 추출합니다. 이 메서드는 네이버 프로필 API를 호출하여 토큰을 검증하고, 성공할 경우 사용자 클레임이 포함된 AccessToken을 반환합니다. Args: token: 검증할 액세스 토큰 Returns: 유효한 경우 사용자 클레임이 포함된 AccessToken, 무효한 경우 None """ try: async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: # 네이버 프로필 API 호출 response = await client.get( "https://openapi.naver.com/v1/nid/me", headers={ "Authorization": f"Bearer {token}", "User-Agent": "FastMCP-Naver-OAuth", }, ) # HTTP 상태 코드 확인 if response.status_code != 200: logger.debug( "Naver profile API returned HTTP %d for token verification", response.status_code, ) return None # JSON 응답 파싱 try: api_response = response.json() except Exception as e: logger.debug("Failed to parse Naver API response as JSON: %s", e) return None # 네이버 프로필 API 결과 코드 확인 result_code = api_response.get("resultcode") if result_code != "00": logger.debug( "Naver API returned error code %s: %s", result_code, api_response.get("message", "Unknown error"), ) return None # 'response' 필드에서 사용자 프로필 데이터 추출 user_profile = api_response.get("response") if not user_profile or not isinstance(user_profile, dict): logger.debug("Naver API response missing user profile data") return None # 필수 사용자 ID 검증 user_id = user_profile.get("id") if not user_id: logger.debug("Naver user profile missing required 'id' field") return None # 스코프 검증 # 네이버의 프로필 API는 스코프 정보를 반환하지 않음 # 프로필 API 접근 성공으로 스코프 유효성을 간접 검증 print("required_scopes:",self.required_scopes) validated_scopes = self.required_scopes or ["openid"] if self.required_scopes: if "openid" in self.required_scopes: logger.debug("Naver openid scope validated via successful profile access") else: logger.warning( "Naver scope validation limited - profile API doesn't return scope info" ) # 네이버 프로필 데이터로부터 사용자 클레임 구성 user_claims = { "sub": user_id, "iss": "naver", } # 네이버에서 제공하는 사용자 프로필 필드들 profile_fields = { "email": user_profile.get("email"), "name": user_profile.get("name"), "nickname": user_profile.get("nickname"), "profile_image": user_profile.get("profile_image"), "age": user_profile.get("age"), "gender": user_profile.get("gender"), "birthday": user_profile.get("birthday"), "birthyear": user_profile.get("birthyear"), "mobile": user_profile.get("mobile"), } # 값이 있는 필드만 클레임에 추가 for field, value in profile_fields.items(): if value is not None and value != "": user_claims[field] = value # 원본 프로필 데이터 보존 user_claims["naver_profile"] = user_profile # AccessToken 생성 # Note: 네이버는 프로필 API에서 토큰 만료 정보를 제공하지 않음 # 토큰 교환 시점에서 expires_in 정보를 별도로 추적해야 함 access_token = AccessToken( token=token, client_id="naver-verified", # 네이버 검증된 토큰 식별자 scopes=validated_scopes, expires_at=None, # 토큰 교환 시 설정됨 claims=user_claims, ) logger.debug( "Naver token verified successfully for user %s (nickname: %s)", user_id, user_profile.get("nickname", "unknown") ) return access_token except httpx.RequestError as e: logger.debug("Network error during Naver token verification: %s", e) return None except Exception as e: logger.error( "Unexpected error during Naver token verification: %s", e, exc_info=True ) return None class NaverProvider(OAuthProxy): """FastMCP용 네이버 OAuth 프로바이더. 이 프로바이더는 모든 FastMCP 서버에 네이버 OAuth 보호 기능을 쉽게 추가할 수 있게 합니다. 네이버 OAuth 앱 자격 증명과 기본 URL만 제공하면 바로 사용할 수 있습니다. 주요 기능: - 네이버에 대한 OAuth Proxy - 네이버 사용자 정보 API를 통한 자동 토큰 검증 - 네이버 프로필 API로부터 사용자 정보 추출 - PKCE, DCR 인터페이스, 메타데이터 제공 등 네이버 로그인이 MCP 표준을 간접적으로 준수하도록 기능 제공 사용 예시: ```python from fastmcp import FastMCP from fastmcp.server.auth.providers.naver import NaverProvider auth = NaverProvider( client_id="your_naver_client_id", client_secret="your_naver_client_secret", base_url="https://my-server.com" ) mcp = FastMCP("My App", auth=auth) ``` """ def __init__( self, *, client_id: str | NotSetT = NotSet, client_secret: str | NotSetT = NotSet, base_url: AnyHttpUrl | str | NotSetT = NotSet, issuer_url: AnyHttpUrl | str | NotSetT = NotSet, redirect_path: str | NotSetT = NotSet, required_scopes: list[str] | NotSetT = NotSet, timeout_seconds: int | NotSetT = NotSet, allowed_client_redirect_uris: list[str] | NotSetT = NotSet, client_storage: AsyncKeyValue | None = None, jwt_signing_key: str | bytes | NotSetT = NotSet, require_authorization_consent: bool = True, ): """네이버 OAuth 프로바이더를 초기화합니다. Args: client_id: 네이버 개발자센터에서 발급받은 OAuth 클라이언트 ID. client_secret: 네이버 개발자센터에서 발급받은 OAuth 클라이언트 시크릿. base_url: OAuth 엔드포인트가 접근 가능한 공개 URL (마운트 경로 포함). issuer_url: OAuth 메타데이터를 위한 Issuer URL (기본값: base_url). 경로 하위에 마운트할 때 discovery 중 404 오류를 방지하기 위해 루트 레벨 URL을 사용하세요. redirect_path: 네이버 OAuth 앱에 설정된 리다이렉트 경로 (기본값: "/auth/callback"). required_scopes: 필수 네이버 스코프 (기본값: ["openid"]). 일반적인 스코프: - "openid" - OpenID Connect용 (기본값) - "profile" - 프로필 정보 접근 - "email" - 이메일 접근 timeout_seconds: 네이버 API 호출을 위한 HTTP 요청 타임아웃 (기본값: 10초). allowed_client_redirect_uris: MCP 클라이언트용 허용된 리다이렉트 URI 패턴 목록. None(기본값)인 경우 모든 URI 허용, 빈 목록인 경우 URI 허용 안함. client_storage: OAuth 상태를 위한 스토리지 백엔드 (클라이언트 등록, 암호화된 토큰). None인 경우 데이터 디렉토리(`platformdirs`에서 파생)에 DiskStore가 생성됩니다. 디스크 스토어는 JWT 서명 키에서 파생된 키를 사용하여 암호화됩니다. jwt_signing_key: FastMCP JWT 토큰 서명을 위한 시크릿 (문자열 또는 바이트). 바이트가 제공되면 그대로 사용됩니다. 문자열이 제공되면 32바이트 키로 파생됩니다. 제공되지 않으면 업스트림 클라이언트 시크릿이 PBKDF2를 사용하여 32바이트 키로 파생됩니다. require_authorization_consent: 클라이언트 인증 전 사용자 동의 필요 여부 (기본값: True). True인 경우 네이버로 리다이렉트되기 전에 사용자에게 동의 화면이 표시됩니다. False인 경우 사용자 확인 없이 직접 인증이 진행됩니다. 보안 경고: 로컬 개발 또는 테스트 환경에서만 비활성화하세요. """ settings = NaverProviderSettings.model_validate( { k: v for k, v in { "client_id": client_id, "client_secret": client_secret, "base_url": base_url, "issuer_url": issuer_url, "redirect_path": redirect_path, "required_scopes": required_scopes, "timeout_seconds": timeout_seconds, "allowed_client_redirect_uris": allowed_client_redirect_uris, "jwt_signing_key": jwt_signing_key, }.items() if v is not NotSet } ) # 필수 설정 검증 if not settings.client_id: raise ValueError( "client_id is required - set via parameter or NAVER_CLIENT_ID" ) if not settings.client_secret: raise ValueError( "client_secret is required - set via parameter or NAVER_CLIENT_SECRET" ) # 기본값 적용 timeout_seconds_final = settings.timeout_seconds or 10 required_scopes_final = settings.required_scopes or ['openid'] allowed_client_redirect_uris_final = settings.allowed_client_redirect_uris # 네이버 토큰 검증기 생성 token_verifier = NaverTokenVerifier( required_scopes=required_scopes_final, timeout_seconds=timeout_seconds_final, ) # SecretStr에서 문자열 추출 client_secret_str = ( settings.client_secret.get_secret_value() if settings.client_secret else "" ) # 네이버 엔드포인트로 OAuth Proxy 초기화 super().__init__( upstream_authorization_endpoint="https://nid.naver.com/oauth2/authorize", upstream_token_endpoint="https://nid.naver.com/oauth2/token", upstream_client_id=settings.client_id, upstream_client_secret=client_secret_str, token_verifier=token_verifier, base_url=settings.base_url, redirect_path=settings.redirect_path, issuer_url=settings.issuer_url or settings.base_url, allowed_client_redirect_uris=allowed_client_redirect_uris_final, client_storage=client_storage, jwt_signing_key=settings.jwt_signing_key, require_authorization_consent=require_authorization_consent, token_endpoint_auth_method="client_secret_post", # 본문 요청에 포함하여 전달 ) logger.info( "Initialized Naver OAuth provider for client %s with scopes: %s", settings.client_id, required_scopes_final, ) 위 스크립트를 만들기 위해 레퍼런스 코드에서 토큰 검증 방식과 API 엔드 포인트 구성, 두 가지 항목에 커스터마이징을 적용하였습니다. 각 항목의 구체적인 차이점은 이어서 설명하도록 하겠습니다. 이 과정을 참고하면, 서비스 환경에 맞게 기업별 SSO나 소셜 로그인 등 다양한 인증 방식을 확장 적용할 수 있습니다. 토큰 검증 방식의 차이 각 서비스의 API 명세가 다르기 때문에 토큰의 유효성을 확인하는 방법에 차이가 있습니다. GoogleTokenVerifier 다음은 Google의 tokeninfo API(/oauth2/v1/tokeninfo)를 호출한 예시입니다. 해당 API는 토큰 자체의 정보(만료 시간, 스코프 등)를 직접 반환해 줍니다. 필요에 따라 userinfo API를 추가로 호출하여 사용자 프로필을 가져올 수 있습니다. Quote class GoogleTokenVerifier(TokenVerifier): """Token verifier for Google OAuth tokens. Google OAuth tokens are opaque (not JWTs), so we verify them by calling Google's tokeninfo API to check if they're valid and get user info. """ def __init__( self, *, required_scopes: list[str] | None = None, timeout_seconds: int = 10, """Initialize the Google token verifier. Args: required_scopes: Required OAuth scopes (e.g., ['openid', 'https://www.googleapis.com/auth/userinfo.email']) timeout_seconds: HTTP request timeout """ super().__init__(required_scopes=required_scopes) self.timeout_seconds = timeout_seconds async def verify_token(self, token: str) -> AccessToken | None: """Verify Google OAuth token by calling Google's tokeninfo API.""" try: async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: # Use Google's tokeninfo endpoint to validate the token response = await client.get( "https://www.googleapis.com/oauth2/v1/tokeninfo", params={"access_token": token}, headers={"User-Agent": "FastMCP-Google-OAuth"}, ) if response.status_code != 200: logger.debug( "Google token verification failed: %d", response.status_code, ) return None token_info = response.json() # Check if token is expired expires_in = token_info.get("expires_in") if expires_in and int(expires_in) <= 0: logger.debug("Google token has expired") return None # Extract scopes from token info scope_string = token_info.get("scope", "") token_scopes = [ scope.strip() for scope in scope_string.split(" ") if scope.strip() ] # Check required scopes if self.required_scopes: token_scopes_set = set(token_scopes) required_scopes_set = set(self.required_scopes) if not required_scopes_set.issubset(token_scopes_set): logger.debug( "Google token missing required scopes. Has %d, needs %d", len(token_scopes_set), len(required_scopes_set), ) return None # Get additional user info if we have the right scopes user_data = {} if "openid" in token_scopes or "profile" in token_scopes: try: userinfo_response = await client.get( "https://www.googleapis.com/oauth2/v2/userinfo", headers={ "Authorization": f"Bearer {token}", "User-Agent": "FastMCP-Google-OAuth", }, ) if userinfo_response.status_code == 200: user_data = userinfo_response.json() except Exception as e: logger.debug("Failed to fetch Google user info: %s", e) # Calculate expiration time expires_at = None if expires_in: expires_at = int(time.time() + int(expires_in)) # Create AccessToken with Google user info access_token = AccessToken( token=token, client_id=token_info.get( "audience", "unknown" ), # Use audience as client_id scopes=token_scopes, expires_at=expires_at, claims={ "sub": user_data.get("id") or token_info.get("user_id", "unknown"), "email": user_data.get("email"), "name": user_data.get("name"), "picture": user_data.get("picture"), "given_name": user_data.get("given_name"), "family_name": user_data.get("family_name"), "locale": user_data.get("locale"), "google_user_data": user_data, "google_token_info": token_info, }, ) logger.debug("Google token verified successfully") return access_token except httpx.RequestError as e: logger.debug("Failed to verify Google token: %s", e) return None except Exception as e: logger.debug("Google token verification error: %s", e) return None 구글 tokeninfo API 응답 예시 Quote { "expires_in": 3599, "scope": "openid email", "user_id": "123456" } 구글 userinfo API 응답 예시 Quote { "id": "108123456789012345678", "email": "user@example.com", "verified_email": true, "name": "홍길동", "given_name": "길동", "family_name": "홍", "picture": "https://lh3.googleusercontent.com/a/default-user=s96-c", "locale": "ko" } NaverTokenVerifier 다음은 네이버의 사용자 프로필 API(/v1/nid/me)를 호출한 예시입니다. 해당 API는 토큰 자체의 정보보다는 사용자 프로필을 바로 반환합니다. 따라서, Google과 달리 네이버는 토큰 검증 과정에서 토큰 자체의 정보(만료 시간, 스코프 등)을 알 수 없습니다. Quote class NaverTokenVerifier(TokenVerifier): """Token verifier for Naver OAuth tokens. 네이버의 사용자 프로필 API를 호출하여 검증해야 합니다. 이 검증기는 네이버 API 사양에 따라 토큰 검증과 사용자 정보 추출을 처리합니다. 주요 특징: - 네이버 프로필 API (/v1/nid/me)를 사용한 토큰 검증 - resultcode/message 구조를 가진 네이버 특유의 응답 형식 처리 - 네이버에서 제공하는 포괄적인 사용자 프로필 데이터 추출 """ def __init__( self, *, required_scopes: list[str] | None = None, timeout_seconds: int = 10, """네이버 토큰 검증기를 초기화합니다. Args: required_scopes: 필수 OAuth 스코프. timeout_seconds: API 호출을 위한 HTTP 요청 타임아웃 (기본값: 10초) """ super().__init__(required_scopes=required_scopes) self.timeout_seconds = timeout_seconds async def verify_token(self, token: str) -> AccessToken | None: """네이버 OAuth 토큰을 검증하고 사용자 정보를 추출합니다. 이 메서드는 네이버 프로필 API를 호출하여 토큰을 검증하고, 성공할 경우 사용자 클레임이 포함된 AccessToken을 반환합니다. Args: token: 검증할 액세스 토큰 Returns: 유효한 경우 사용자 클레임이 포함된 AccessToken, 무효한 경우 None """ try: async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: # 네이버 프로필 API 호출 response = await client.get( "https://openapi.naver.com/v1/nid/me", headers={ "Authorization": f"Bearer {token}", "User-Agent": "FastMCP-Naver-OAuth", }, ) # HTTP 상태 코드 확인 if response.status_code != 200: logger.debug( "Naver profile API returned HTTP %d for token verification", response.status_code, ) return None # JSON 응답 파싱 try: api_response = response.json() except Exception as e: logger.debug("Failed to parse Naver API response as JSON: %s", e) return None # 네이버 프로필 API 결과 코드 확인 result_code = api_response.get("resultcode") if result_code != "00": logger.debug( "Naver API returned error code %s: %s", result_code, api_response.get("message", "Unknown error"), ) return None # 'response' 필드에서 사용자 프로필 데이터 추출 user_profile = api_response.get("response") if not user_profile or not isinstance(user_profile, dict): logger.debug("Naver API response missing user profile data") return None # 필수 사용자 ID 검증 user_id = user_profile.get("id") if not user_id: logger.debug("Naver user profile missing required 'id' field") return None # 스코프 검증 # 네이버의 프로필 API는 스코프 정보를 반환하지 않음 # 프로필 API 접근 성공으로 스코프 유효성을 간접 검증 print("required_scopes:",self.required_scopes) validated_scopes = self.required_scopes or ["openid"] if self.required_scopes: if "openid" in self.required_scopes: logger.debug("Naver openid scope validated via successful profile access") else: logger.warning( "Naver scope validation limited - profile API doesn't return scope info" ) # 네이버 프로필 데이터로부터 사용자 클레임 구성 user_claims = { "sub": user_id, "iss": "naver", } # 네이버에서 제공하는 사용자 프로필 필드들 profile_fields = { "email": user_profile.get("email"), "name": user_profile.get("name"), "nickname": user_profile.get("nickname"), "profile_image": user_profile.get("profile_image"), "age": user_profile.get("age"), "gender": user_profile.get("gender"), "birthday": user_profile.get("birthday"), "birthyear": user_profile.get("birthyear"), "mobile": user_profile.get("mobile"), } # 값이 있는 필드만 클레임에 추가 for field, value in profile_fields.items(): if value is not None and value != "": user_claims[field] = value # 원본 프로필 데이터 보존 user_claims["naver_profile"] = user_profile # AccessToken 생성 # Note: 네이버는 프로필 API에서 토큰 만료 정보를 제공하지 않음 # 토큰 교환 시점에서 expires_in 정보를 별도로 추적해야 함 access_token = AccessToken( token=token, client_id="naver-verified", # 네이버 검증된 토큰 식별자 scopes=validated_scopes, expires_at=None, # 토큰 교환 시 설정됨 claims=user_claims, ) logger.debug( "Naver token verified successfully for user %s (nickname: %s)", user_id, user_profile.get("nickname", "unknown") ) return access_token except httpx.RequestError as e: logger.debug("Network error during Naver token verification: %s", e) return None except Exception as e: logger.error( "Unexpected error during Naver token verification: %s", e, exc_info=True ) return None 네이버 사용자 프로필 API 응답 예시 Quote { "resultcode": "00", "message": "success", "response": { "id": "123456", "email": "user@example.com" } } API 엔드포인트 및 인증 방식의 차이 OAuthProxy가 어떤 서버와 통신할지를 정의하는 API 엔드포인트와 인증 설정값이 서로 다릅니다. GoogleProvider API 엔드포인트: accounts.google.com, oauth2.googleapis.com 등 Google의 공식 주소를 사용합니다. 인증 방식: token_endpoint_auth_method를 따로 지정하지 않아, 기본 값인 client_secret_basic을 사용합니다.(ID/Secret을 HTTP 헤더에 담아 전송) super().__init__( upstream_authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth", upstream_token_endpoint="https://oauth2.googleapis.com/token", ... ) NaverProvider API 엔드포인트: nid.naver.com으로 시작하는 네이버의 공식 주소를 사용합니다. 인증 방식: 네이버 API 명세에 따라 token-endpoint-auth_method="client_secret_post"를 명시적으로 설정했습니다.(ID/Secret을 HTTP 요청 본문에 담아 전송) super().__init__( upstream_authorization_endpoint="https://nid.naver.com/oauth2/authorize", upstream_token_endpoint="https://nid.naver.com/oauth2/token", ... token_endpoint_auth_method="client_secret_post" ) Quote 네이버 로그인을 연동할 때 2가지의 엔드포인트가 존재하며, 본 예제에서는 PKCE를 지원하는 OIDC API를 사용하였습니다. OIDC API는 OAuth 2.0 API와 유사하지만 경로와 요청 파라미터, 응답 값 일부가 다르기 때문에 활용에 참고하시길 바랍니다. OAuth 2.0 API 네이버 로그인 인증 요청: https://nid.naver.com/oauth2.0/authorize 접근토큰 발급 요청: https://nid.naver.com/oauth2.0/token OIDC(Open ID Connect) API 네이버 로그인 인증 요청: https://nid.naver.com/oauth2/authorize 접근토큰 발급 요청: https://nid.naver.com/oauth2/token 4. OAuth 인증이 통합된 MCP 서버 구현 앞서 구현한 naver.py의 NaverProvider 클래스를 사용하여 MCP 서버에 네이버 로그인 인증을 적용합니다. 클라이언트가 이 MCP 서버에 접근하면 자동으로 네이버 로그인 페이지로 리다이렉트되며, 사용자가 권한을 승인하면 서버를 사용할 수 있게 됩니다. 이번 예제에서 다루는 MCP 서버에는 총 세 가지의 도구를 등록합니다. get_current_time: 현재 시간을 반환하는 도구로, 정확한 시간 계산을 돕습니다. web_search: 네이버 검색을 수행하는 도구입니다. create_calendar_schedule: 네이버 캘린더에 일정을 추가하는 도구로, 해당 도구를 사용하기 위해서는 인증이 필요합니다. 다음은 네이버 인증이 적용된 MCP 서버 구현 코드입니다. # remote_auth_server.py import os import re import uuid import httpx from datetime import datetime from fastmcp import FastMCP from naver import NaverProvider # 커스터마이징한 naver.py from dotenv import load_dotenv load_dotenv() NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID") NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET") BASE_URL = os.getenv("BASE_URL") # 네이버 OAuth Provider 초기화 auth = NaverProvider( client_id=NAVER_CLIENT_ID, client_secret=NAVER_CLIENT_SECRET, base_url=BASE_URL, ) mcp = FastMCP(name="Naver Remote MCP", auth=auth) # Naver OAuth 인증 적용 # 현재 시간 도구 정의 @mcp.tool( name="get_current_time", description="현재 시간을 반환하는 도구입니다. 한국 시간(KST, UTC+9) 기준으로 현재 날짜와 시간을 제공합니다." ) # openAI API는 빈 properties를 허용하지 않으므로 더미 파라미터 설정 async def get_current_time(format: str = "default") -> dict: """ 현재 시간을 조회합니다. Args: format (str): 사용하지 않음 (호환성을 위한 더미 파라미터) Returns: { "datetime": str, # ISO 8601 형식 (YYYY-MM-DDTHH:MM:SS) "date": str, # 현재 날짜 (YYYY-MM-DD) "time": str, # 현재 시간 (HH:MM:SS) "weekday": str, # 요일 (한글) "timestamp": int # Unix timestamp } """ import pytz # 한국 시간대 설정 kst = pytz.timezone('Asia/Seoul') now = datetime.now(kst) weekdays = ['월요일', '화요일', '수요일', '목요일', '금요일', '토요일', '일요일'] return { "datetime": now.strftime("%Y-%m-%dT%H:%M:%S"), "date": now.strftime("%Y-%m-%d"), "time": now.strftime("%H:%M:%S"), "weekday": weekdays[now.weekday()], "timestamp": int(now.timestamp()) } # 네이버 검색 도구 정의 @mcp.tool( name="web_search", description="네이버 웹 검색을 수행하는 도구입니다." ) async def web_search(query: str, display: int = 10, start: int = 1, sort: str = "sim") -> dict: """ 네이버 웹 검색을 수행합니다. Args: query (str): 검색어. 사용자의 요청을 반영한 구체적이고 문맥이 드러나는 검색 질의(ex. 2025년 개봉 예정인 SF 영화). 시점, 조건, 문맥 등을 포함할 것. display (int): 출력 건수 (1~100) start (int): 검색 시작 위치 (1~1000) sort (str): 정렬 옵션 ("sim" | "date") Returns: { "query": str, "total": int, "items": [{"title": str, "link": str, "description": str}] } """ url = "https://openapi.naver.com/v1/search/webkr.json" params = {"query": query, "display": display, "start": start, "sort": sort} headers = { "X-Naver-Client-Id": NAVER_CLIENT_ID, "X-Naver-Client-Secret": NAVER_CLIENT_SECRET, } async with httpx.AsyncClient() as client: r = await client.get(url, headers=headers, params=params) r.raise_for_status() data = r.json() results = [] for item in data.get("items", []): results.append({ "title": re.sub(r"<.*?>", "", item.get("title") or "").strip(), "link": item.get("link"), "description": re.sub(r"<.*?>", "", item.get("description") or "").strip(), }) return { "query": query, "total": data.get("total", 0), "items": results, } # 네이버 캘린더 일정 추가 도구 정의 @mcp.tool( name="create_calendar_schedule", description="네이버 캘린더에 새로운 일정을 추가하는 도구입니다." ) async def create_calendar_schedule( title: str, start_datetime: str, end_datetime: str, description: str = "", location: str = "", ) -> dict: """ 캘린더에 일정을 추가합니다. Args: title (str): 일정의 제목을 추가합니다. start_datetime (str): 일정이 시작하는 시간을 설정합니다. end_datetime (str): 일정이 끝나는 시간을 설정합니다. description (str): 일정에 대한 설명을 추가합니다. location (str): 일정의 장소를 추가합니다. Returns: { "success": bool, "message": str, "response": str # (성공 시) } 또는 실패 시 { "success": bool, "error_code": int, # (예외 발생 시 "error": str) "message": str } """ from fastmcp.server.dependencies import get_access_token import urllib.request import urllib.parse from datetime import datetime as dt try: token = get_access_token() access_token = token.token header = "Bearer " + access_token url = "https://openapi.naver.com/calendar/createSchedule.json" # 날짜 형식 변환 네이버 캘린더가 지원하는 형식: "YYYYMMDDTHHMMSS" def parse_datetime(dt_str): # 이미 올바른 형식이면 그대로 반환 if 'T' in dt_str and len(dt_str) == 15: return dt_str # "YYYY-MM-DD HH:MM:SS" 형식 변환 try: parsed = dt.strptime(dt_str.replace('T', ' ').replace('-', '').replace(':', ''), '%Y%m%d %H%M%S') return parsed.strftime('%Y%m%dT%H%M%S') except: pass # "YYYY-MM-DD HH:MM:SS" 형식 try: parsed = dt.strptime(dt_str, '%Y-%m-%d %H:%M:%S') return parsed.strftime('%Y%m%dT%H%M%S') except: pass return dt_str start_datetime = parse_datetime(start_datetime) end_datetime = parse_datetime(end_datetime) # URL 인코딩 cal_sum = urllib.parse.quote(title) cal_des = urllib.parse.quote(description) if description else "" cal_loc = urllib.parse.quote(location) if location else "" # 고유 ID 생성 uid = str(uuid.uuid4())[:14] # iCalendar 문자열 생성 schedule_ical = "BEGIN:VCALENDAR\n" schedule_ical += "VERSION:2.0\n" schedule_ical += "PRODID:Naver Calendar\n" schedule_ical += "CALSCALE:GREGORIAN\n" schedule_ical += "BEGIN:VTIMEZONE\n" schedule_ical += "TZID:Asia/Seoul\n" schedule_ical += "BEGIN:STANDARD\n" schedule_ical += "DTSTART:19700101T000000\n" schedule_ical += "TZNAME:GMT%2B09:00\n" schedule_ical += "TZOFFSETFROM:%2B0900\n" schedule_ical += "TZOFFSETTO:%2B0900\n" schedule_ical += "END:STANDARD\n" schedule_ical += "END:VTIMEZONE\n" schedule_ical += "BEGIN:VEVENT\n" schedule_ical += "SEQUENCE:0\n" schedule_ical += "CLASS:PUBLIC\n" schedule_ical += "TRANSP:OPAQUE\n" schedule_ical += f"UID:{uid}\n" schedule_ical += f"DTSTART;TZID=Asia/Seoul:{start_datetime}\n" schedule_ical += f"DTEND;TZID=Asia/Seoul:{end_datetime}\n" schedule_ical += f"SUMMARY:{cal_sum} \n" if description: schedule_ical += f"DESCRIPTION:{cal_des} \n" if location: schedule_ical += f"LOCATION:{cal_loc} \n" now = datetime.now().strftime("%Y%m%dT%H%M%SZ") schedule_ical += f"CREATED:{now}\n" schedule_ical += f"LAST-MODIFIED:{now}\n" schedule_ical += f"DTSTAMP:{now}\n" schedule_ical += "END:VEVENT\n" schedule_ical += "END:VCALENDAR" data = "calendarId=defaultCalendarId&scheduleIcalString=" + schedule_ical request = urllib.request.Request(url, data=data.encode("utf-8")) request.add_header("Authorization", header) response = urllib.request.urlopen(request) rescode = response.getcode() if rescode == 200: response_body = response.read() result = response_body.decode('utf-8') return { "success": True, "message": "일정이 성공적으로 추가되었습니다.", "response": result } else: return { "success": False, "error_code": rescode, "message": f"일정 추가에 실패했습니다." } except Exception as e: return { "success": False, "error": str(e), "message": "일정 추가 중 오류가 발생했습니다." } if __name__ == "__main__": mcp.run(transport="streamable-http", path="/mcp") Quote 아래 JSON Configuration을 통해 Claude나 Cursor에서 실행시킨 인증 기반 MCP서버에 접근할 수 있습니다. { "mcpServers": { "naver-remote-mcp": { "command": "npx", "args": ["-y", "mcp-remote", "https://xxx.ngrok-free.app/mcp"] } } 5. FastMCP 기반 클라이언트 구현 본 프로젝트에서 FastMCP의 OAuth Proxy를 기반으로 한 NaverProvider를 사용하여 인증이 적용된 MCP 서버를 구성했습니다. FastMCP 클라이언트는 NaverProvider의 OAuth 인증 플로우(브라우저 로그인, PKCE, 토큰 관리 등) 를 완전히 지원하므로, 이 서버와의 통신을 위해서는 FastMCP 기반 클라이언트 구현이 필요합니다. 다음의 클라이언트 스크립트를 실행한 뒤 인증 과정을 거치면 '~/.fastmcp/oauth-mcp-client-cache' 경로에 토큰이 발급됩니다. 그리고 클라이언트는 해당 토큰을 가지고 MCP 서버에 접근할 수 있게 됩니다. 모델은 HCX-005를 사용하였고, 만약 HCX-007 모델을 도구와 함께 사용할 경우에는 reasoning_effort="none"을 반드시 설정해야 합니다. # fastmcp_client.py import os import json import glob import asyncio from typing import List from fastmcp import Client # https://gofastmcp.com/clients/client#the-fastmcp-client from langchain.agents import create_agent from langchain.messages import SystemMessage, HumanMessage, AIMessage from langchain_naver import ChatClovaX from langchain_core.tools import StructuredTool from pydantic import create_model from dotenv import load_dotenv load_dotenv() async def main(clova_api_key: str, server_url: str): """ FastMCP 클라이언트를 실행합니다. 인증 기반 MCP 서버와 연결하여 인증 절차를 진행하고 서버에 등록된 도구를 연동하여 최종 응답을 생성합니다. Args: clova_api_key: CLOVA Studio API Key. server_url: MCP 서버의 URL. """ model = ChatClovaX(model="HCX-005", api_key=clova_api_key) # FastMCP 클라이언트 생성 및 OAuth 인증 async with Client(server_url, auth="oauth") as auth_client: print("✓ Authenticated with OAuth!") # MCP 도구를 LangChain 형식으로 변환 structured_tools = await load_tools(auth_client) agent = create_agent(model, structured_tools) state = { "messages": [ SystemMessage(content=( "당신은 비서입니다. 사용자가 원하는 것을 도와주세요\n" "당신은 현재 시간을 알 수 있고, 웹 검색을 하거나 캘린더에 일정을 추가 할 수 있습니다\n" "사용자 요청에 대해 현재 시간이 필요한 경우에는 정확한 시간을 활용해 주세요\n" "예를 들어, 사용자가 '지금 몇 시야?' 라고 요청하면 현재 시간을 알려줍니다\n " "사용자가 '내일 일정 만들어줘','3일 뒤로 일정 잡아줘' 라고 요청하면 현재 시간을 기준으로 날짜를 확인해 활용해주세요\n" "각 도구는 반드시 한 번만 호출하세요. 일정 추가(create_calendar_schedule)는 절대 중복 호출하지 마세요\n" )) ] } print("\nAI: 안녕하세요. 저는 AI 어시스턴트입니다. 원하시는 요청을 입력해 주세요. (종료하려면 '종료'를 입력하세요.)") while True: user_input = input("\n\nUser: ") if user_input.lower() in ["종료", "exit"]: print("AI: 대화를 종료합니다. 이용해주셔서 감사합니다.") break state["messages"].append(HumanMessage(content=user_input)) # astream_events를 사용하여 스트리밍으로 결과 처리 try: final_answer = "" async for event in agent.astream_events(state, version="v1"): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: print(chunk.content, end="", flush=True) final_answer += chunk.content elif kind == "on_tool_start": print(f"\n[도구 선택]: {event['name']}\n[도구 호출]: {event['data'].get('input')}") elif kind == "on_tool_end": print(f"[도구 응답]: {event['data'].get('output')}\n") # 스트리밍이 끝나면 최종 답변을 AIMessage로 만들어 상태에 추가 state["messages"].append(AIMessage(content=final_answer)) except Exception as e: print(f"\nAI: 요청을 처리하는 중에 오류가 발생했습니다. 오류: {e}") pass async def load_tools(client: Client) -> List[StructuredTool]: """ FastMCP 클라이언트로부터 도구를 로드하여 LangChain StructuredTool로 변환합니다. Args: client: FastMCP Client 인스턴스 (인증된 세션 유지) Returns: List[StructuredTool]: LangChain 호환 도구 목록 """ # MCP 서버에서 도구 목록 가져오기 tools = await client.list_tools() structured_tools = [] for tool in tools: # 도구의 입력 스키마 추출 schema = tool.inputSchema or {} props = schema.get("properties", {}) # 파라미터 없는 도구 제외 if not props: continue # Pydantic 모델 동적 생성 - MCP 스키마를 Pydantic 필드로 변환 field_definitions = {} for key, prop in props.items(): # JSON Schema 타입을 Python 타입으로 변환 field_type = _get_python_type(prop.get("type", "string")) # 모든 필드를 required로 설정 (... = Ellipsis) field_definitions[key] = (field_type, ...) # 동적으로 Pydantic 모델 클래스 생성 # 예: "web_search" -> "web_searchInput" 클래스 InputModel = create_model( f"{tool.name}Input", **field_definitions ) # 각 도구마다 고유한 async 함수 생성 # tool_name과 mcp_client를 캡처하여 각 도구마다 고유한 함수 생성 def create_async_func(tool_name=tool.name, mcp_client=client): async def func(**kwargs): # 실제 MCP 서버의 도구 호출 result = await mcp_client.call_tool(tool_name, kwargs) return result return func # LangChain StructuredTool 생성 # coroutine 파라미터를 사용하여 async 함수 지원 structured_tool = StructuredTool.from_function( coroutine=create_async_func(), name=tool.name, description=tool.description or "", args_schema=InputModel, ) structured_tools.append(structured_tool) return structured_tools def _get_python_type(json_type: str) -> type: """JSON Schema 타입을 Python 타입으로 변환합니다.""" type_mapping = { "string": str, "integer": int, "number": float, "boolean": bool, "array": list, "object": dict, } return type_mapping.get(json_type, str) if __name__ == "__main__": CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") BASE_URL = os.getenv("BASE_URL") SERVER_URL = BASE_URL + "/mcp/" asyncio.run(main(CLOVA_STUDIO_API_KEY, SERVER_URL)) 6. 시나리오 실행 본 섹션에서는 위에서 구현한 클라이언트를 실행했을 때 나타나는 단계를 살펴봅니다. 네이버 로그인 및 권한 부여 1. FastMCP 클라이언트를 실행하면 다음과 같은 화면이 나타납니다. 2. Allow Access 버튼을 눌러주면, 네이버 로그인 페이지로 넘어가집니다. 3. 로그인을 완료하면 권한 부여 페이지가 나타납니다. 캘린더 일정 추가 도구를 사용하려면 캘린더 일정 등록 권한에 반드시 동의해야 합니다. Quote 선택 권한을 해제 했을 때 수정하는 방법 네이버 로그인 페이지에 접속한 다음 로그인을 합니다. 이력 관리 > 연결된 서비스 관리를 확인합니다. 연결한 MCP 서버를 클릭하여 서비스 동의 철회를 진행합니다. 클라이언트를 다시 실행하여 권한 부여를 다시 진행합니다. 4. 권한 부여가 완료되면 FastMCP의 인증 완료 페이지로 리다이렉트됩니다. 대화 실행 결과 아래 예시는 사용자가 두 차례 질의를 이어가는 과정에서, 인증 기반 MCP 서버의 web_search 도구를 호출하여 장소를 검색하고, get_current_time 도구와 create_calendar_schedule 도구를 호출하여 해당 장소에 방문하는 일정을 추가합니다. 해당 과정에서 누적된 대화 히스토리가 활용되는 흐름을 보여줍니다. AI: 안녕하세요. 저는 AI 어시스턴트입니다. 원하시는 요청을 입력해 주세요. (종료하려면 '종료'를 입력하세요.) User: 예술의 전당이 어디있어? [도구 선택]: web_search [도구 호출]: {'query': '예술의 전당 위치', 'display': 1, 'start': 1, 'sort': 'date'} [도구 응답]: content='CallToolResult(content=[TextContent(type=\'text\', text=\'{"query":"예술의 전당 위치","total":3145637,"items":[{"title":"오시는 길·주차안내 | 방문·이용","link":"https://www.sac.or.kr/site/main/welcome/welcome?parkingInfo=y","description":"주소 도로명 : 서울시 서초구 남부순환로 2406 예술의전당 (우 06757), 지번 : 서울시 서초구 서초동 700번지, 건물명 : 예술의전당(오페라하우스, 음악당, 한가람미술관,한가람디자인미술관, 서울서예박물관)"}]}\', annotations=None, meta=None)], structured_content={\'query\': \'예술의 전당 위치\', \'total\': 3145637, \'items\': [{\'title\': \'오시는 길·주차안내 | 방문·이용\', \'link\': \'https://www.sac.or.kr/site/main/welcome/welcome?parkingInfo=y\', \'description\': \'주소 도로명 : 서울시 서초구 남부순환로 2406 예술의전당 (우 06757), 지번 : 서울시 서초구 서초동 700번지, 건물명 : 예술의전당(오페라하우스, 음악당, 한가람미술관,한가람디자인미술관, 서울서예박물관)\'}]}, data={\'query\': \'예술의 전당 위치\', \'total\': 3145637, \'items\': [{\'title\': \'오시는 길·주차안내 | 방문·이용\', \'link\': \'https://www.sac.or.kr/site/main/welcome/welcome?parkingInfo=y\', \'description\': \'주소 도로명 : 서울시 서초구 남부순환로 2406 예술의전당 (우 06757), 지번 : 서울시 서초구 서초동 700번지, 건물명 : 예술의전당(오페라하우스, 음악당, 한가람미술관,한가람디자인미술관, 서울서예박물관)\'}]}, is_error=False)' name='web_search' tool_call_id='call_DW7Ti30f1EeSZ5tXBZAVebsR' 예술의 전당은 **서울시 서초구 남부순환로 2406**에 위치해 있으며, 우편번호는 06757입니다. 건물명은 오페라하우스, 음악당, 한가람미술관, 한가람디자인미술관, 서울서예박물관이 포함되어 있습니다. 보다 자세한 정보와 주차 안내는 공식 웹사이트인 [https://www.sac.or.kr](https://www.sac.or.kr)에서 확인하실 수 있습니다. User: 3일 뒤 오후 3시에 방문 일정 잡아줘. [도구 선택]: get_current_time [도구 호출]: {'format': 'yyyy-MM-dd HH:mm'} [도구 응답]: content='CallToolResult(content=[TextContent(type=\'text\', text=\'{"datetime":"2025-10-15T14:48:06","date":"2025-10-15","time":"14:48:06","weekday":"수요일","timestamp":1760507286}\', annotations=None, meta=None)], structured_content={\'datetime\': \'2025-10-15T14:48:06\', \'date\': \'2025-10-15\', \'time\': \'14:48:06\', \'weekday\': \'수요일\', \'timestamp\': 1760507286}, data={\'datetime\': \'2025-10-15T14:48:06\', \'date\': \'2025-10-15\', \'time\': \'14:48:06\', \'weekday\': \'수요일\', \'timestamp\': 1760507286}, is_error=False)' name='get_current_time' tool_call_id='call_mLl81eYuX5SIsUINUOCFjJ2q' [도구 선택]: create_calendar_schedule [도구 호출]: {'title': '예술의 전당 방문', 'start_datetime': '2025-10-18 15:00', 'end_datetime': '2025-10-18 16:00', 'description': '예술의 전당 방문 일정', 'location': '서울시 서초구 남부순환로 2406'} [도구 응답]: content='CallToolResult(content=[TextContent(type=\'text\', text=\'{"success":true,"message":"일정이 성공적으로 추가되었습니다.","response":"{\\\\"result\\\\":\\\\"success\\\\",\\\\"code\\\\":200,\\\\"returnValue\\\\":{\\\\"calendarId\\\\":\\\\"5946777\\\\",\\\\"processType\\\\":\\\\"create\\\\",\\\\"icalUid\\\\":\\\\"f518004f-459f-\\\\"}}"}\', annotations=None, meta=None)], structured_content={\'success\': True, \'message\': \'일정이 성공적으로 추가되었습니다.\', \'response\': \'{"result":"success","code":200,"returnValue":{"calendarId":"5946777","processType":"create","icalUid":"f518004f-459f-"}}\'}, data={\'success\': True, \'message\': \'일정이 성공적으로 추가되었습니다.\', \'response\': \'{"result":"success","code":200,"returnValue":{"calendarId":"5946777","processType":"create","icalUid":"f518004f-459f-"}}\'}, is_error=False)' name='create_calendar_schedule' tool_call_id='call_CfxP1vBw9mmGkuheiM1KzsJb' 네, 3일 뒤인 **2025년 10월 18일 오후 3시**부터 1시간 동안 예술의 전당 방문 일정이 잡혔습니다. 해당 일정은 사용자님의 캘린더에 저장되었으며, 주소는 서울시 서초구 남부순환로 2406입니다. 즐거운 관람 되세요! 마무리 3부에서는 FastMCP에서 제공하는 인증 시스템을 MCP 서버에 통합하는 과정을 살펴보았습니다. NaverProvider를 예시로 커스터마이징 한 인증 클래스를 구현하고, FastMCP의 OAuth Proxy를 통해 인증 과정을 자동화함으로써, HyperCLOVA X 모델과 안전하게 연동되는 개인화 MCP 서버를 완성했습니다. 마지막 4부에서는 운영과 확장에 대한 팁을 다룹니다. MCP Inspector를 활용한 서버 디버깅, Flowise를 이용한 시각적 테스트, 그리고 직접 서버를 작성하고 확장하는 실전 노하우를 소개합니다. MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 링크 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 ✔︎ (4부) 운영과 확장을 위한 다양한 팁 링크 링크 복사 다른 사이트에 공유하기 More sharing options...
Recommended Posts
게시글 및 댓글을 작성하려면 로그인 해주세요.
로그인