BOJ MCP Server

from fastmcp import FastMCP from fastmcp.prompts.prompt import PromptMessage, TextContent from pydantic import BaseModel, Field, ConfigDict, computed_field from typing import Optional, Dict, Any import httpx from contextlib import asynccontextmanager from smithery.decorators import smithery import solvedac_server.search_query as sq TIER_NAMES = ["Bronze", "Silver", "Gold", "Platinum", "Diamond", "Ruby"] TIER_SYMBOLS = ["V", "IV", "III", "II", "I"] DEFAULT_LIMIT = 5 SOLVEDAC_API_BASE_URL = "https://solved.ac/api/v3" state = {"http_client": None} def get_tier_name_from_level(level: int) -> str: """solved.ac의 숫자 레벨을 문자열 티어 이름(예: Platinum V)으로 변환합니다.""" if level == 0: return "Unrated" if 1 <= level <= 30: # (level - 1) // 5 = 0(Bronze) ~ 5(Ruby) tier_index = (level - 1) // 5 # (level - 1) % 5 = 0(V) ~ 4(I) sub_tier_index = (level - 1) % 5 main_tier = TIER_NAMES[tier_index] symbol = TIER_SYMBOLS[sub_tier_index] return f"{main_tier} {symbol}" if level == 31: return "Master" return "Unknown" class UserShowResponse(BaseModel): model_config = ConfigDict(extra="ignore") handle: str tier: int rating: int solvedCount: int @computed_field @property def tier_name(self) -> str: """ API에서 받은 숫자 티어(self.tier)를 헬퍼 함수를 이용해 문자열 이름(예: Platinum V)으로 변환합니다. """ return get_tier_name_from_level(self.tier) async def get_user_info_core(handle: str) -> UserShowResponse: client: httpx.AsyncClient = state.get("http_client") if not client: raise RuntimeError("HTTP Client is not available.") try: resp = await client.get("/user/show", params={"handle": handle}) resp.raise_for_status() return UserShowResponse.model_validate(resp.json()) except httpx.HTTPStatusError as e: s = e.response.status_code if s == 404: raise ValueError( f"사용자 핸들 '{handle}'을(를) 찾을 수 없습니다." ) from e if s == 429: raise RuntimeError( "요청이 많습니다(429). 잠시 후 다시 시도하세요." ) from e if 500 <= s < 600: raise RuntimeError( "solved.ac 서버 오류가 발생했습니다. 잠시 후 다시 시도하세요." ) from e raise except httpx.RequestError as e: raise RuntimeError( f"네트워크 오류로 사용자 정보를 가져오지 못했습니다: {e}" ) from e @asynccontextmanager async def lifespan(app: FastMCP): state["http_client"] = httpx.AsyncClient( base_url=SOLVEDAC_API_BASE_URL, headers={"X-Solvedac-Language": "ko"}, timeout=10.0, follow_redirects=True, ) try: yield finally: if state["http_client"]: await state["http_client"].aclose() state["http_client"] = None @smithery.server() def create_server(): app = FastMCP(name="SolvedAcAPI", lifespan=lifespan) # ---------- 1) 도구 함수: 모델 호출 함수(실행 버튼) ---------- @app.tool( name="solvedac_get_user_info", description="solved.ac 사용자의 레이팅/티어/푼 문제 수 조회" ) async def get_user_info_tool( handle: str = Field(..., description="사용자 핸들") ): return await get_user_info_core(handle) @app.tool( name="solvedac_refresh_tag_catalog", description="solved.ac 태그 카탈로그를 강제로 다시 로드합니다.", ) async def refresh_tag_catalog_tool(): client: httpx.AsyncClient = state.get("http_client") raw = await sq.fetch_tag_list_core(client) cat = sq.build_tag_catalog(raw) state["tag_catalog"] = cat return {"refreshed": True, "count": len(cat.tags)} @app.tool( name="solvedac_search_problems", description=( "난이도/태그/키워드 검색 (예: tier:g5..p5 #dp). " "공식 태그 정규화(옵션) 후, -@handle/-t@handle + sort=random." ), ) async def search_problems_tool( query: str = Field( ..., description="검색 쿼리 (예: 'tier:g5..p5 #dp' 또는 '완전탐색')", ), limit: int = Field( DEFAULT_LIMIT, ge=1, le=20, description="반환 개수" ), handle: Optional[str] = Field( None, description="사용자 solved.ac 핸들" ), max_attempts: int = Field( 3, ge=1, le=5, description="랜덤 정렬 재시도(중복 제거)" ), strict_tags: bool = Field( False, description="모르는 태그는 에러(True)/무시(False)" ), enable_bare_infer: bool = Field( False, description="맨바닥 단어(예: '완전탐색')를 태그로 승격" ), ): q0 = (query or "").strip() # 0) 태그가 실제로 필요할 때만 카탈로그 사용 if sq._needs_tag_catalog(q0, enable_bare_infer): cat = await sq._ensure_tag_catalog(state) # 1) 명시 태그(#dp, tag:dp ...) 정규화 q_norm, unknown, _ = sq.normalize_query_tags( q0, cat.token_to_key, strict=strict_tags ) if strict_tags and unknown: raise ValueError(f"알 수 없는 태그: {', '.join(unknown)}") # 2) 요청 시에만 bare-word 승격 (예: '완전탐색' → '#bruteforce') if enable_bare_infer: inferred = sq.infer_bare_tag_keys(q_norm, cat.token_to_key) if inferred: q_norm = ( f"{q_norm} " + " ".join(f"#{k}" for k in inferred) ) else: # 태그 필요 없음 → 그대로 진행 q_norm = q0 # 3) 푼/시도 제외 필터 부착 q = sq._augment_query_with_user_filters(q_norm, handle) # 4) 서버 랜덤 정렬 + 중복 제거 + 재시도 uniq: Dict[int, Any] = {} last_resp = None for _ in range(max_attempts): resp = await sq.search_problems_core( state, query=q, page=1, sort="random" ) last_resp = resp for it in (resp.items or []): pid = getattr(it, "problemId", None) or getattr(it, "id", None) if pid is not None and pid not in uniq: uniq[pid] = it if len(uniq) >= limit: break if len(uniq) >= limit: break if last_resp is None: raise RuntimeError("검색 응답이 없습니다.") items = list(uniq.values())[:limit] last_resp.items = items last_resp.count = len(items) # ✅ 반환 개수로 세팅 (테스트가 기대하는 계약) return last_resp # ---------- 2) 리소스 함수: 모델이 참고할 자료(자료실 주소) ---------- @app.resource( "solvedac://tags/catalog", description="solved.ac 공식 태그 목록을 로드하여 서버 메모리에 보관합니다.", ) async def solvedac_tag_catalog_resource(): client: httpx.AsyncClient = state.get("http_client") if not client: raise RuntimeError("HTTP Client is not available.") cat = state.get("tag_catalog") if cat is None: raw = await sq.fetch_tag_list_core(client) cat = sq.build_tag_catalog(raw) state["tag_catalog"] = cat return {"count": len(cat.tags), "keys": [t.key for t in cat.tags]} @app.resource( "solvedac://tags/catalog", description="solved.ac 공식 태그 목록을 로드하여 서버 메모리에 보관합니다.", ) async def solvedac_tag_catalog_resource(): cat = await sq._ensure_tag_catalog(state) # ← 내부 헬퍼 사용 return {"count": len(cat.tags), "keys": [t.key for t in cat.tags]} @app.resource("solvedac://users/{handle}", description="특정 solved.ac 사용자의 기본 정보(레이팅, 티어, 푼 문제 수)를 조회합니다.") async def get_user_info( handle: str = Field(..., description="조회하려는 사용자의 solved.ac 핸들/아이디"), ) -> UserShowResponse: return await get_user_info_core(handle) @app.resource("solvedac://problems/search/{stub}", description="난이도, 태그, 키워드 등으로 solved.ac 문제를 검색합니다. (예: query='tier:s5..g5 tag:dp')") async def search_problems( query: str = Field(..., description="문제 검색 쿼리"), page: int = Field(1, ge=1, description="페이지 번호(1부터 시작)"), stub: str = Field("_", description="템플릿 제약 대응용 더미 세그먼트(무시됨)"), ) -> sq.ProblemSearchResponse: return await sq.search_problems_core(state, query=query, page=page) # ---------- 3) 프롬프트 함수: 모델이 참고할 행동(행동 절차 카드) ---------- @app.prompt( name="solvedac.search-workflow", description="자연어 조건을 solved.ac 검색 쿼리로 변환하고, 해당 쿼리로 문제 후보를 검토합니다.", tags={"solvedac", "search"}, ) def search_workflow_prompt( natural_request: str = Field( ..., description="예: '실버~골드 사이 DP 5문제, 비슷한 태그는 제외'" ), page: int = Field(1, ge=1, description="검색 페이지"), ) -> list[PromptMessage]: sys = PromptMessage( role="assistant", content=TextContent( type="text", text=( "You are a Solved.ac search assistant.\n" "1) Convert the user's request into a precise Solved.ac query string " "(e.g., `tier:g5..p5 tag:dfs -tag:greedy`).\n" "2) Do NOT browse the web.\n" "3) Call the MCP TOOL `solvedac_search_problems` with {query, page}.\n" "4) Rank top 5 by suitability and show: problemId, titleKo, level." ), ), ) usr = PromptMessage( role="user", content=TextContent( type="text", text=( f"요청(자연어): {natural_request}\n" f"페이지: {page}\n" f"규칙: 쿼리를 먼저 제시하고, 이어서 리소스를 호출해 결과를 평가하세요." ), ), ) return [sys, usr] get_user_info_for_test = get_user_info_core search_problems_for_test = sq.search_problems_core search_workflow_prompt_for_test = sq.search_workflow_prompt_core return app