get_vetting_result

Instructions

Implementation Reference

• mcp_server/skill_store_mcp.py:834-908 (handler)

  The main handler function for the get_vetting_result MCP tool. It polls the security vetting result for a skill by job_id, supporting both api_key (authenticated) and claim_token (draft uploader) authentication. Returns a formatted string with is_done flag, vetting_status, and findings[].

  @mcp.tool()
  @_log_tool
  def get_vetting_result(job_id: str, api_key: str = "", claim_token: str = "") -> str:
      """
      Get the detailed security vetting report for a skill (poll by job_id, claim_token supported). / 보안 검수 결과 상세 조회.
      업로드 응답의 vetting_job_id 로 검수 결과를 폴링합니다.
      에이전트가 이메일 없이 HTTP만으로 최종 결과를 받는 공식 권장 경로.
  
      ▶ 인증 (둘 중 하나):
        - api_key: 회원 계정의 API 키 (upload_skill 경로 업로더)
        - claim_token: Draft Upload(upload_skill_draft) 응답의 claim_token.
          API 키 없는 에이전트는 이 토큰으로 자신의 검수 결과를 폴링 가능.
  
      반환 메시지에는 is_done 플래그, vetting_status, findings[] 가 포함됩니다.
      is_done=false 면 몇 초 후 다시 호출하세요 (보통 검수는 수 초~수십 초 소요).
  
      Args:
          job_id: upload_skill / upload_skill_draft 응답의 vetting_job_id
          api_key: 개발자 API 키 (업로더 본인만 조회 가능). 없으면 claim_token 필수.
          claim_token: Draft Upload 응답의 claim_token (api_key 대안).
  
      Returns:
          검수 결과 메시지 (is_done 여부 + 결과 포함)
      """
      if not api_key and not claim_token:
          return "❌ api_key 또는 claim_token 중 하나를 반드시 제공하세요."
  
      # claim_token 경로: 인증 헤더 없이 query 파라미터로 전달
      if not api_key and claim_token:
          import urllib.parse as _up
          path = f"/v1/skills/vetting/{job_id}?claim_token={_up.quote(claim_token)}"
          result = _get(path)
          if result.get("status") == "error" or result.get("error_code"):
              code = result.get("error_code") or "ERROR"
              msg = result.get("detail") or result.get("message") or "조회 실패"
              return f"❌ [{code}]: {msg}"
      else:
          result = _get_auth(f"/v1/skills/vetting/{job_id}", api_key)
      if result.get("status") == "error" or result.get("error_code"):
          code = result.get("error_code") or "ERROR"
          msg = result.get("detail") or result.get("message") or "알 수 없는 오류"
          return f"❌ 조회 실패 [{code}]: {msg}"
  
      is_done = bool(result.get("is_done"))
      vs = result.get("vetting_status") or "unknown"
      js = result.get("job_status") or "unknown"
      icon = "✅" if vs in ("approved", "officially_approved") else ("❌" if vs in ("rejected", "officially_rejected") else ("⚠️" if vs == "caution" else "⏳"))
  
      lines = [
          f"{icon} 검수 결과 (is_done={is_done})",
          f"Job ID: {result.get('job_id')}",
          f"Version ID: {result.get('version_id')}",
          f"Job 상태: {js}",
          f"Vetting 상태: {vs}",
      ]
      if result.get("started_at"):
          lines.append(f"시작: {result['started_at']}")
      if result.get("finished_at"):
          lines.append(f"완료: {result['finished_at']}")
      if result.get("summary"):
          lines.append(f"요약: {result['summary']}")
      findings = result.get("findings") or []
      if findings:
          lines.append(f"\n발견 사항 ({len(findings)}건):")
          for i, f in enumerate(findings[:10], 1):
              code = f.get("code") or "-"
              sev = f.get("severity") or ""
              m = f.get("message") or ""
              lines.append(f"  {i}. [{code}] {sev} {m}")
      if result.get("error_msg"):
          lines.append(f"오류: {result['error_msg']}")
      if not is_done:
          lines.append("\n⏳ 아직 진행 중입니다. 몇 초 후 다시 호출하세요.")
  
      return "\n".join(lines)

• mcp_server/skill_store_mcp.py:834-837 (registration)

  Registration via @mcp.tool() decorator on FastMCP instance 'mcp' — this registers the function as an MCP tool named 'get_vetting_result' (derived from the function name).

  @mcp.tool()
  @_log_tool
  def get_vetting_result(job_id: str, api_key: str = "", claim_token: str = "") -> str:
      """

• mcp_server/skill_store_mcp.py:85-95 (helper)

  Helper _get(): Makes an unauthenticated GET request. Used by get_vetting_result for the claim_token auth path (no API key).

  def _get(path: str, params: dict = None) -> dict:
      url = SKILL_STORE_URL + path
      if params:
          url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
      try:
          with urllib.request.urlopen(url, timeout=10) as resp:
              return json.loads(resp.read().decode())
      except urllib.error.HTTPError as e:
          return {"status": "error", "message": f"HTTP {e.code}: {e.reason}"}
      except Exception as e:
          return {"status": "error", "message": str(e)}

• mcp_server/skill_store_mcp.py:113-130 (helper)

  Helper _get_auth(): Makes an authenticated GET request with X-API-KEY header. Used by get_vetting_result for the api_key auth path.

  def _get_auth(path: str, api_key: str, params: dict = None) -> dict:
      """API 키 인증이 필요한 GET 요청."""
      url = SKILL_STORE_URL + path
      if params:
          url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
      req = urllib.request.Request(url, headers={"X-API-KEY": api_key})
      try:
          with urllib.request.urlopen(req, timeout=10) as resp:
              return json.loads(resp.read().decode())
      except urllib.error.HTTPError as e:
          body = e.read().decode()
          try:
              body = json.loads(body).get("message", body)
          except Exception:
              pass
          return {"status": "error", "message": f"HTTP {e.code}: {body}"}
      except Exception as e:
          return {"status": "error", "message": str(e)}

• mcp_server/skill_store_mcp.py:33-45 (helper)

  Decorator _log_tool: Logs each MCP tool call to stdout (used on get_vetting_result).

  def _log_tool(fn):
      """각 MCP tool 호출을 stdout 에 한 줄 기록 — journalctl 에서 grep 가능.
      형식: TOOL_CALL tool=<name> kw=<arg_keys>  (PII 회피 위해 값은 로그 X)
      """
      @_functools_tool.wraps(fn)
      def _wrapper(*args, **kwargs):
          try:
              kw_keys = list(kwargs.keys())
              print(f"TOOL_CALL tool={fn.__name__} kw={kw_keys}", flush=True)
          except Exception:
              pass
          return fn(*args, **kwargs)
      return _wrapper