list_creatives

Instructions

Implementation Reference

• src/kakao_moment_mcp/tools/creatives.py:21-87 (handler)

  Core handler function that implements the list_creatives tool logic. Fetches creatives from Kakao Moment API, optionally fetches details, and returns normalized items with status summary.

  async def list_creatives(
      client: KakaoMomentClient,
      adgroup_id: str | None = None,
      *,
      with_details: bool = True,
  ) -> dict[str, Any]:
      """소재 목록.
  
      Args:
          adgroup_id: 특정 광고그룹의 소재만 필터. 카카오모먼트는 adgroup_id 가
                      없으면 KakaoMomentException(-813) 을 던지므로 필수에 가깝습니다.
          with_details: True 면 소재별 detail 을 동시 호출해 포맷/심사상태/이미지/생성일 보강.
      """
      if not adgroup_id:
          raise ValueError(
              "list_creatives 는 adgroup_id 가 필요합니다. "
              "list_adgroups 로 ID 를 먼저 얻어 전달하세요."
          )
      params: dict[str, Any] = {"adGroupId": adgroup_id}
      data = await client.get("/openapi/v4/creatives", params=params)
      rows = extract_rows(data)
  
      details: dict[Any, dict[str, Any]] = {}
      if with_details and rows:
          details = await fetch_details(
              client,
              "/openapi/v4/creatives/{id}",
              [r.get("id") for r in rows],
          )
  
      items: list[dict[str, Any]] = []
      for r in rows:
          merged = {**r, **details.get(r.get("id"), {})}
          items.append(
              {
                  "id": merged.get("id"),
                  "name": merged.get("name"),
                  "adgroup_id": merged.get("adGroupId") or adgroup_id,
                  "status": merged.get("config") or merged.get("status"),
                  "creative_type": merged.get("format")
                  or merged.get("creativeFormat")
                  or merged.get("creativeType"),
                  "preview_url": _preview_url(merged),
                  "landing_url": merged.get("pcLandingUrl"),
                  "review_status": merged.get("reviewStatus"),
                  "creative_status": merged.get("creativeStatus"),
                  "title": merged.get("title"),
                  "description": merged.get("description"),
                  "rejected_reason": merged.get("rejectedReason") or None,
                  "status_description": merged.get("statusDescription"),
                  "created_at": merged.get("createdDate"),
              }
          )
  
      on_count = sum(1 for it in items if str(it.get("status", "")).upper() == "ON")
      review_ok = sum(
          1
          for it in items
          if str(it.get("review_status", "")).upper() in ("APPROVED", "OK")
      )
      summary = f"소재 {len(items)}개 (진행 {on_count} · 심사완료 {review_ok})"
      return {
          "count": len(items),
          "items": items,
          "summary": summary,
          "raw": data,
      }

• src/kakao_moment_mcp/tools/_common.py:12-61 (helper)

  Helper function extract_rows() used to parse API response rows, and fetch_details() used for concurrent detail fetching.

  def extract_rows(payload: Any) -> list[dict[str, Any]]:
      """카카오모먼트 응답에서 row 리스트를 안전하게 추출.
  
      카카오는 엔드포인트별로 응답 형태가 달라서 여러 경우를 다 시도한다:
        - {"content": [...], ...}              (페이지네이션 응답)
        - {"data": [...]}                     (단순 리스트)
        - {"data": {...}}                      (단일 객체)
        - [...]                                (배열 직접)
        - {...}                                (객체 직접)
      """
      if isinstance(payload, list):
          return [r for r in payload if isinstance(r, dict)]
      if not isinstance(payload, dict):
          return []
      for key in ("content", "data"):
          v = payload.get(key)
          if isinstance(v, list):
              return [r for r in v if isinstance(r, dict)]
      # data 가 단일 객체이거나 keys 가 그대로 row 인 경우
      if "data" in payload and isinstance(payload["data"], dict):
          return [payload["data"]]
      return []
  
  
  async def fetch_details(
      client: KakaoMomentClient,
      path_template: str,
      ids: list[Any],
  ) -> dict[Any, dict[str, Any]]:
      """ID 리스트를 받아 path_template.format(id=...) 로 detail 을 동시 조회한다.
  
      카카오모먼트의 list 엔드포인트는 daily_budget/bid_amount 등을 포함하지 않으므로
      list 한 뒤 각 항목의 detail 을 N+1 로 받아 보강한다. 동시 호출은 카카오 레이트리밋
      완화를 위해 _DETAIL_CONCURRENCY 로 제한.
  
      실패한 ID 는 dict 에서 누락된다(빈 dict). 호출 측은 .get(id, {}) 패턴으로 안전 접근.
      """
      sem = asyncio.Semaphore(_DETAIL_CONCURRENCY)
  
      async def one(item_id: Any) -> tuple[Any, dict[str, Any]]:
          async with sem:
              try:
                  data = await client.get(path_template.format(id=item_id))
              except Exception:  # noqa: BLE001 — detail 실패는 silent (raw fallback)
                  return item_id, {}
          body = data.get("data") if isinstance(data, dict) and "data" in data else data
          return item_id, body if isinstance(body, dict) else {}
  
      results = await asyncio.gather(*(one(i) for i in ids if i is not None))
      return {k: v for k, v in results}

• src/kakao_moment_mcp/tools/creatives.py:10-10 (helper)

  Helper function _preview_url() extracts a displayable preview URL from the merged creative detail response.

  def _preview_url(merged: dict[str, Any]) -> str | None: