check_draft_status

Instructions

Implementation Reference

• mcp_server/skill_store_mcp.py:707-755 (handler)

  The main handler function for the 'check_draft_status' MCP tool. It accepts a claim_token, calls the API endpoint /v1/drafts/status, and returns a formatted status summary (claimed, expired, agent_claimed, etc.). Decorated with @mcp.tool() and @_log_tool.

  @mcp.tool()
  @_log_tool
  def check_draft_status(claim_token: str) -> str:
      """
      Check the status of a draft skill upload using a claim_token. / Draft 스킬 상태 공개 조회.
  
      사용 시점:
        - 사람이 claim_url 을 클릭해서 인증을 끝냈는지 확인
        - contact_email 로 보낸 agent-level verify 메일이 처리됐는지 확인
        - Draft 가 30일 안에 claim 됐는지 / 만료됐는지 확인
  
      Args:
          claim_token: upload_skill_draft 응답의 claim_token
  
      Returns:
          상태 요약 (claimed, expired, agent_verify_email_sent, agent_claimed 등).
      """
      if not claim_token or not claim_token.strip():
          return "❌ claim_token 은 필수입니다."
      import urllib.parse as _up
      result = _get(f"/v1/drafts/status?claim_token={_up.quote(claim_token.strip())}")
      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}"
  
      claimed = bool(result.get("claimed"))
      expired = bool(result.get("expired"))
      agent_claimed = bool(result.get("agent_claimed"))
      icon = "✅" if claimed else ("⏳" if not expired else "⌛")
  
      lines = [
          f"{icon} Draft 상태 — {result.get('skill_name')}",
          f"  skill_id:                 {result.get('skill_id')}",
          f"  draft_agent_author:       {result.get('draft_agent_author')}",
          f"  created_at:               {result.get('created_at')}",
          f"  expires_at:               {result.get('expires_at')}",
          f"  claimed (skill-level):    {claimed}{' at ' + result['claimed_at'] if claimed else ''}",
          f"  expired:                  {expired}",
          f"  verify_email_sent:        {result.get('verify_email_sent')}  (skill-level 1:1 legacy)",
          f"  agent_verify_email_sent:  {result.get('agent_verify_email_sent')}  (agent-level, 2026-04-23)",
          f"  agent_claimed:            {agent_claimed}  (agent identity 이미 계정에 귀속)",
      ]
      if not claimed and not expired and not agent_claimed:
          lines.append("  ⏳ 아직 사람이 claim 하지 않았습니다. claim_url 을 사용자에게 전달했는지 확인하세요.")
      if agent_claimed and not claimed:
          lines.append("  ℹ️ agent-level claim 은 완료됐으나 이 특정 Draft 는 아직 이전 안 됐을 수 있음 — "
                       "vetting 완료 후 자동으로 owner 계정에 귀속됩니다.")
      return "\n".join(lines)

• mcp_server/skill_store_mcp.py:707-708 (registration)

  The @mcp.tool() decorator registers 'check_draft_status' as an MCP tool on the FastMCP server instance.

  @mcp.tool()
  @_log_tool

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

  The _get() helper function is called by check_draft_status to make the GET request to /v1/drafts/status?claim_token=... It returns parsed JSON or an error dict.

  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)}