list_campaigns

Instructions

Implementation Reference

• src/kakao_moment_mcp/tools/campaigns.py:10-79 (handler)

  Core handler: async function list_campaigns() that calls Kakao API /openapi/v4/campaigns, optionally fetches details per campaign (daily_budget, type, goal), and returns a normalized dict with count, items, summary, and raw data.

  async def list_campaigns(
      client: KakaoMomentClient,
      status_filter: str | None = None,
      *,
      with_details: bool = True,
  ) -> dict[str, Any]:
      """캠페인 목록.
  
      Args:
          status_filter: ON / OFF / DELETED 등 카카오 상태 코드. None 이면 전체.
          with_details: True 면 캠페인별 detail 을 동시 호출해 일예산/타입/생성일 보강.
                        카카오 list 엔드포인트는 id/name/config 만 반환하므로 기본 ON.
                        대규모 계정(수백개)에서 부담되면 False 로 끌 수 있음.
      """
      params: dict[str, Any] = {}
      if status_filter:
          params["config"] = status_filter.upper()
      data = await client.get("/openapi/v4/campaigns", params=params or None)
      rows = extract_rows(data)
  
      details: dict[Any, dict[str, Any]] = {}
      if with_details and rows:
          details = await fetch_details(
              client,
              "/openapi/v4/campaigns/{id}",
              [r.get("id") for r in rows],
          )
  
      items: list[dict[str, Any]] = []
      for r in rows:
          merged = {**r, **details.get(r.get("id"), {})}
          type_goal = merged.get("campaignTypeGoal")
          items.append(
              {
                  "id": merged.get("id"),
                  "name": merged.get("name"),
                  "campaign_type": (
                      type_goal.get("campaignType")
                      if isinstance(type_goal, dict)
                      else merged.get("campaignType")
                  ),
                  "goal": (
                      type_goal.get("goal") if isinstance(type_goal, dict) else None
                  ),
                  "status": merged.get("config") or merged.get("status"),
                  "daily_budget": merged.get("dailyBudgetAmount")
                  or merged.get("dailyBudget"),
                  "is_over_budget": merged.get("isDailyBudgetAmountOver"),
                  "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")
      off_count = sum(
          1 for it in items if str(it.get("status", "")).upper() in ("OFF", "PAUSED")
      )
      total_budget = sum(
          float(it.get("daily_budget") or 0) for it in items if it.get("daily_budget")
      )
      summary = (
          f"캠페인 {len(items)}개 (진행 {on_count} · 일시정지 {off_count})"
          f" · 일예산 합계 {int(total_budget):,}원"
      )
      return {
          "count": len(items),
          "items": items,
          "summary": summary,
          "raw": data,
      }

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

  Helper: fetch_details() used to make concurrent detail API calls per campaign ID; extract_rows() to normalize list response formats.

  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}