Quote MCP Server

# tools.py """ 명언/인용구 조회를 위한 도구 모음 무료 Quote API 사용 """ import os import logging import requests from cachetools import TTLCache from typing import Optional, Dict, List from datetime import datetime import json import time # 기본 API URL (무료 Quote API 예제) QUOTE_API_URL = "https://api.quotable.io" # Logger logger = logging.getLogger("sample-mcp") level = getattr(logging, os.environ.get("LOG_LEVEL", "INFO").upper(), logging.INFO) logger.setLevel(level) if not logger.handlers: handler = logging.StreamHandler() handler.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")) logger.addHandler(handler) logger.propagate = True # 캐시 설정 quote_cache = TTLCache(maxsize=100, ttl=3600) # 1시간 유지 search_cache = TTLCache(maxsize=200, ttl=1800) # 30분 # 실패한 요청 캐시 (불필요한 재시도 방지, 5분 유지) failure_cache = TTLCache(maxsize=200, ttl=300) # 5분 def get_credentials(arguments: Optional[dict] = None) -> dict: """ 환경 변수에서 API 인증 정보를 가져옵니다. 우선순위: 1) arguments.env, 2) .env 파일 Args: arguments: 도구 호출 인자 Returns: 인증 정보가 담긴 딕셔너리 """ api_key = "" key_source = "none" # 우선순위 1: arguments.env에서 받기 (메인 서버에서 받은 키) if isinstance(arguments, dict) and "env" in arguments: env = arguments["env"] if isinstance(env, dict) and "QUOTE_API_KEY" in env: api_key = env["QUOTE_API_KEY"] key_source = "arguments.env" # 우선순위 2: .env 파일에서 받기 (로컬 개발용) if not api_key: api_key = os.environ.get("QUOTE_API_KEY", "") if api_key: key_source = ".env file" credentials = { "QUOTE_API_KEY": api_key, } # 로깅 (키 마스킹) masked_key = credentials["QUOTE_API_KEY"] if masked_key: masked_key = masked_key[:6] + "***" + f"({len(masked_key)} chars)" logger.debug( "Resolved credentials | QUOTE_API_KEY=%s, source=%s", masked_key or "<empty>", key_source ) return credentials def make_request_with_retry(url: str, params: dict = None, max_retries: int = 3, timeout: int = 30) -> requests.Response: """ 네트워크 요청을 재시도 로직과 함께 수행 Args: url: 요청 URL params: 요청 파라미터 max_retries: 최대 재시도 횟수 timeout: 타임아웃 (초) Returns: Response 객체 Raises: requests.exceptions.RequestException: 모든 재시도 실패 시 """ last_exception = None for attempt in range(max_retries): try: response = requests.get(url, params=params, timeout=timeout) response.raise_for_status() return response except requests.exceptions.Timeout as e: last_exception = e if attempt < max_retries - 1: wait_time = (attempt + 1) * 1 logger.warning("Request timeout (attempt %d/%d), retrying in %ds...", attempt + 1, max_retries, wait_time) time.sleep(wait_time) else: logger.error("Request timeout after %d attempts", max_retries) except requests.exceptions.ConnectionError as e: last_exception = e if attempt < max_retries - 1: wait_time = (attempt + 1) * 1 logger.warning("Connection error (attempt %d/%d), retrying in %ds...", attempt + 1, max_retries, wait_time) time.sleep(wait_time) else: logger.error("Connection error after %d attempts", max_retries) except requests.exceptions.RequestException as e: logger.error("Request failed (non-retryable): %s", str(e)) raise if last_exception: raise last_exception else: raise requests.exceptions.RequestException("모든 재시도가 실패했습니다.") def get_random_quote(arguments: Optional[dict] = None) -> Dict: """ 랜덤 명언을 조회합니다. Args: arguments: 추가 인자 (API 키 등) Returns: 명언 정보 딕셔너리 또는 {"error": "오류 메시지"} """ logger.debug("get_random_quote called") cache_key = "random" if cache_key in quote_cache: logger.debug("Cache hit for random quote") return quote_cache[cache_key] if cache_key in failure_cache: logger.debug("Failure cache hit, skipping") return failure_cache[cache_key] credentials = get_credentials(arguments) # 이 API는 API 키가 필요 없지만, 구조를 보여주기 위해 포함 try: url = f"{QUOTE_API_URL}/random" response = make_request_with_retry(url) data = response.json() result = { "quote": data.get("content", ""), "author": data.get("author", ""), "id": data.get("_id", "") } quote_cache[cache_key] = result logger.debug("API call successful for random quote") return result except requests.exceptions.RequestException as e: error_msg = f"API 요청 실패: {str(e)}" logger.error(error_msg) error_result = {"error": error_msg} failure_cache[cache_key] = error_result return error_result except Exception as e: error_msg = f"예상치 못한 오류: {str(e)}" logger.exception(error_msg) return {"error": error_msg} def search_quotes(keyword: str, limit: int = 10, arguments: Optional[dict] = None) -> Dict: """ 키워드로 명언을 검색합니다. Args: keyword: 검색할 키워드 limit: 결과 개수 arguments: 추가 인자 (API 키 등) Returns: 검색 결과 딕셔너리 또는 {"error": "오류 메시지"} """ logger.debug("search_quotes called | keyword=%r limit=%d", keyword, limit) cache_key = (keyword.lower(), limit) if cache_key in search_cache: logger.debug("Cache hit for search | keyword=%r", keyword) return search_cache[cache_key] if cache_key in failure_cache: logger.debug("Failure cache hit, skipping | keyword=%r", keyword) return failure_cache[cache_key] credentials = get_credentials(arguments) try: url = f"{QUOTE_API_URL}/search/quotes" params = { "query": keyword, "limit": limit } response = make_request_with_retry(url, params) data = response.json() quotes = [] for item in data.get("results", []): quotes.append({ "quote": item.get("content", ""), "author": item.get("author", ""), "id": item.get("_id", "") }) result = { "total": data.get("totalCount", len(quotes)), "quotes": quotes } search_cache[cache_key] = result logger.debug("API call successful for search | keyword=%r total=%d", keyword, len(quotes)) return result except requests.exceptions.RequestException as e: error_msg = f"API 요청 실패: {str(e)}" logger.error(error_msg) error_result = {"error": error_msg} failure_cache[cache_key] = error_result return error_result except Exception as e: error_msg = f"예상치 못한 오류: {str(e)}" logger.exception(error_msg) return {"error": error_msg} def get_quote_by_id(quote_id: str, arguments: Optional[dict] = None) -> Dict: """ ID로 특정 명언을 조회합니다. Args: quote_id: 인용구 ID arguments: 추가 인자 (API 키 등) Returns: 명언 정보 딕셔너리 또는 {"error": "오류 메시지"} """ logger.debug("get_quote_by_id called | quote_id=%r", quote_id) cache_key = quote_id if cache_key in quote_cache: logger.debug("Cache hit for quote | quote_id=%r", quote_id) return quote_cache[cache_key] if cache_key in failure_cache: logger.debug("Failure cache hit, skipping | quote_id=%r", quote_id) return failure_cache[cache_key] credentials = get_credentials(arguments) try: url = f"{QUOTE_API_URL}/quotes/{quote_id}" response = make_request_with_retry(url) data = response.json() result = { "quote": data.get("content", ""), "author": data.get("author", ""), "id": data.get("_id", "") } quote_cache[cache_key] = result logger.debug("API call successful for quote | quote_id=%r", quote_id) return result except requests.exceptions.RequestException as e: error_msg = f"API 요청 실패: {str(e)}" logger.error(error_msg) error_result = {"error": error_msg} failure_cache[cache_key] = error_result return error_result except Exception as e: error_msg = f"예상치 못한 오류: {str(e)}" logger.exception(error_msg) return {"error": error_msg}