search_blog
Search and retrieve blog results from Naver using specific keywords, with options to navigate pages and customize display settings for precise and structured blog discovery.
Instructions
Searches for blogs on Naver using the given keyword. The page parameter allows for page navigation.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| display | No | ||
| page | No | ||
| query | Yes | ||
| sort | No | sim |
Implementation Reference
- server.py:359-362 (registration)MCP tool registration for 'search_blog' using @mcp.tool decorator, specifying the tool name and description.@mcp.tool( name="search_blog", description="Searches for blogs on Naver using the given keyword. The page parameter allows for page navigation." )
- server.py:363-377 (handler)The handler function implementing the core logic of 'search_blog': calculates pagination, prepares API parameters, and calls the shared _make_api_call to fetch and format Naver blog search results.async def search_blog(query: str, display: int = 10, page: int = 1, sort: str = "sim") -> str: """ Searches for blogs on Naver using the given keyword. The page parameter allows for page navigation. Args: query (str): The keyword to search for display (int, optional): The number of results to display. Default is 10. page (int, optional): The starting page number. Default is 1. sort (str, optional): 정렬 기준. 기본값은 "sim" (유사도). """ start = calculate_start(page, display) display = min(display, 100) # display 최대값 제한 params = {"query": query, "display": display, "start": start, "sort": sort} return await _make_api_call("blog.json", params, BlogResult, "Blog")
- server.py:123-123 (schema)Pydantic schema (BlogResult model) used for validating and parsing the Naver blog search API response.class BlogResult(SearchResultBase): items: List[BlogItem]
- server.py:348-355 (helper)Helper function to compute the 'start' parameter for Naver API pagination, used by search_blog.def calculate_start(page: int, display: int) -> int: """Calculates the start value for the API call based on the page number and display count.""" if page < 1: page = 1 start = (page - 1) * display + 1 # 네이버 API의 start 최대값(1000) 제한 고려 return min(start, 1000)
- server.py:158-345 (helper)Shared helper function that performs the actual Naver API call, parses response using the provided model (e.g., BlogResult), handles errors, and formats output text. Called by search_blog.async def _make_api_call( endpoint: str, params: Dict[str, Any], result_model: BaseModel, search_type_name: str # 동적 프롬프트 생성을 위한 검색 타입 이름 추가 ) -> str: """ Calls the Naver search API and parses the result, returning the result in text format. """ if not HEADERS: logger.error("네이버 API 인증 정보가 설정되지 않았습니다.") error_resp = ErrorResponse(error="인증 정보 미설정", details="NAVER_CLIENT_ID 또는 NAVER_CLIENT_SECRET 환경 변수를 확인하세요.") return "오류 발생:\n" + f"오류: {error_resp.error}\n세부사항: {error_resp.details}" url = f"{NAVER_API_BASE_URL}{endpoint}" prompt_string = "처리 중 오류 발생:" # 기본 오류 프롬프트 try: async with httpx.AsyncClient(timeout=10.0) as client: logger.info(f"네이버 API 호출 시작 - URL: {url}, Params: {params}") response = await client.get(url, headers=HEADERS, params=params) response.raise_for_status() # HTTP 오류 시 예외 발생 data = response.json() logger.info(f"API 응답 성공 (상태 코드: {response.status_code})") try: # Pydantic 모델로 파싱 및 유효성 검사 result = result_model.model_validate(data) logger.info(f"데이터 파싱 성공 (모델: {result_model.__name__})") # 동적 Prompt 생성 (SearchResultBase 상속 모델인 경우) if isinstance(result, SearchResultBase): start_index = result.start end_index = result.start + len(result.items) - 1 prompt_string = f"네이버 {search_type_name} 검색 결과 (총 {result.total:,}건 중 {start_index}~{end_index}번째):" # 결과를 구조화된 텍스트 형식으로 변환 text_result = f"{prompt_string}\n\n" # 결과 항목 형식화 for i, item in enumerate(result.items, 1): text_result += f"### 결과 {i}\n" # 일반적인 항목 처리 (대부분의 모델에 공통) if hasattr(item, 'title'): # HTML 태그 제거 title = item.title.replace('<b>', '').replace('</b>', '') text_result += f"제목(title): {title}\n" if hasattr(item, 'link'): text_result += f"링크(link): {item.link}\n" if hasattr(item, 'description') and item.description: # HTML 태그 제거 desc = item.description.replace('<b>', '').replace('</b>', '') text_result += f"설명(description): {desc}\n" # 모델별 특수 필드 처리 if isinstance(item, BlogItem): text_result += f"블로거명(bloggername): {item.bloggername}\n" text_result += f"블로그 링크(bloggerlink): {item.bloggerlink}\n" if item.postdate: text_result += f"작성일(postdate): {item.postdate}\n" elif isinstance(item, NewsItem): if item.originallink: text_result += f"원본 링크(originallink): {item.originallink}\n" if item.pubDate: text_result += f"발행일(pubDate): {item.pubDate}\n" elif isinstance(item, BookItem) or isinstance(item, ShopItem): if hasattr(item, 'image') and item.image: text_result += f"이미지(image): {item.image}\n" if hasattr(item, 'author') and item.author: text_result += f"저자(author): {item.author}\n" if hasattr(item, 'price') and item.price: text_result += f"가격(price): {item.price}\n" if hasattr(item, 'discount') and item.discount: text_result += f"할인가(discount): {item.discount}\n" if hasattr(item, 'publisher') and item.publisher: text_result += f"출판사(publisher): {item.publisher}\n" if hasattr(item, 'pubdate') and item.pubdate: text_result += f"출판일(pubdate): {item.pubdate}\n" if hasattr(item, 'isbn') and item.isbn: text_result += f"ISBN(isbn): {item.isbn}\n" elif isinstance(item, ShopItem): if hasattr(item, 'image') and item.image: text_result += f"이미지(image): {item.image}\n" if hasattr(item, 'lprice') and item.lprice: text_result += f"최저가(lprice): {item.lprice}\n" if hasattr(item, 'hprice') and item.hprice: text_result += f"최고가(hprice): {item.hprice}\n" if hasattr(item, 'mallName') and item.mallName: text_result += f"쇼핑몰명(mallName): {item.mallName}\n" if hasattr(item, 'brand') and item.brand: text_result += f"브랜드(brand): {item.brand}\n" if hasattr(item, 'maker') and item.maker: text_result += f"제조사(maker): {item.maker}\n" if hasattr(item, 'category1') and item.category1: text_result += f"카테고리1(category1): {item.category1}\n" if hasattr(item, 'category2') and item.category2: text_result += f"카테고리2(category2): {item.category2}\n" if hasattr(item, 'category3') and item.category3: text_result += f"카테고리3(category3): {item.category3}\n" if hasattr(item, 'category4') and item.category4: text_result += f"카테고리4(category4): {item.category4}\n" elif isinstance(item, LocalItem): if item.category: text_result += f"카테고리(category): {item.category}\n" if item.telephone: text_result += f"전화번호(telephone): {item.telephone}\n" if item.address: text_result += f"주소(address): {item.address}\n" if item.roadAddress: text_result += f"도로명주소(roadAddress): {item.roadAddress}\n" if item.mapx: text_result += f"지도 X좌표(mapx): {item.mapx}\n" if item.mapy: text_result += f"지도 Y좌표(mapy): {item.mapy}\n" elif isinstance(item, ImageItem): if item.thumbnail: text_result += f"썸네일(thumbnail): {item.thumbnail}\n" if item.sizeheight: text_result += f"높이(sizeheight): {item.sizeheight}\n" if item.sizewidth: text_result += f"너비(sizewidth): {item.sizewidth}\n" elif isinstance(item, EncycItem): if item.thumbnail: text_result += f"썸네일(thumbnail): {item.thumbnail}\n" elif isinstance(item, CafeArticleItem): if item.cafename: text_result += f"카페명(cafename): {item.cafename}\n" if item.cafeurl: text_result += f"카페 링크(cafeurl): {item.cafeurl}\n" text_result += "\n" return text_result elif isinstance(result, AdultResult): prompt_string = f"네이버 {search_type_name} 확인 결과:" if result.adult == 0: return f"{prompt_string} 일반 검색어" else: return f"{prompt_string} 성인 검색어" elif isinstance(result, ErrataResult): print(f"ErrataResult: {result}") prompt_string = f"네이버 {search_type_name} 확인 결과:" if result.errata == "": return f"{prompt_string} 오타 없음" else: return f"{prompt_string} {result.errata}" else: # 예상치 못한 결과 타입 prompt_string = f"네이버 {search_type_name} 처리 결과:" # 결과를 JSON 형식의 문자열로 변환 result_json = json.dumps(result.model_dump(), ensure_ascii=False) return f"{prompt_string}\n{result_json}" except ValidationError as e: logger.error(f"Pydantic 유효성 검사 오류: {e}") error_resp = ErrorResponse(error="응답 데이터 형식 오류", details=str(e)) return f"{prompt_string}\n오류: {error_resp.error}\n세부사항: {error_resp.details}" except httpx.HTTPStatusError as e: logger.error(f"API HTTP 상태 오류: {e.response.status_code} - {e.response.text}", exc_info=True) error_resp = ErrorResponse( error=f"API 오류 ({e.response.status_code})", details=e.response.text, status_code=e.response.status_code ) return f"{prompt_string}\n오류: {error_resp.error}\n세부사항: {error_resp.details}" except httpx.RequestError as e: logger.error(f"네트워크 요청 오류: {e}", exc_info=True) error_resp = ErrorResponse(error="네트워크 오류", details=f"네이버 API 서버 연결 실패: {e}") return f"{prompt_string}\n오류: {error_resp.error}\n세부사항: {error_resp.details}" except Exception as e: logger.exception(f"예상치 못한 오류 발생: {e}") # exc_info=True와 동일 error_resp = ErrorResponse(error="서버 내부 오류", details=str(e)) return f"{prompt_string}\n오류: {error_resp.error}\n세부사항: {error_resp.details}"