Claude Code x Teams API で週次KPIレポートを自動集計・自動投稿する完全ガイド

毎週の進捗報告作成が30分からゼロになった

チームリーダーの石川です。毎週月曜、先週のKPIをExcelで集計してTeamsに貼り付けるのに30分かかっていました。Claude Codeでデータ収集から集計・Teamsへの自動投稿まで完全自動化し、月曜の朝には自動でKPIレポートが届いています。

STEP 1：複数システムからKPIを自動収集する

Salesforce・GA4・Jiraなど複数システムからKPIデータを自動収集します。

import requests
from datetime import date, timedelta

def get_salesforce_kpi(last_week):
    sf_url   = "https://your-instance.salesforce.com/services/data/v57.0/query"
    sf_token  = get_sf_token()
    week_start = last_week.strftime("%Y-%m-%dT00:00:00Z")
    week_end   = (last_week + timedelta(days=6)).strftime("%Y-%m-%dT23:59:59Z")
    
    # SOQLで週次データを取得
    queries = {
        "売上合計":     f"SELECT SUM(Amount) total FROM Opportunity WHERE CloseDate >= {week_start[:10]} AND CloseDate <= {week_end[:10]} AND StageName = 'Closed Won'",
        "新規商談数":   f"SELECT COUNT() FROM Opportunity WHERE CreatedDate >= {week_start} AND CreatedDate <= {week_end}",
        "成約率":       f"SELECT SUM(Amount) total FROM Opportunity WHERE CloseDate >= {week_start[:10]} AND StageName = 'Closed Won'",
    }
    
    results = {}
    for kpi, soql in queries.items():
        res = requests.get(sf_url, params={"q": soql},
            headers={"Authorization": f"Bearer {sf_token}"}).json()
        results[kpi] = res["records"][0]["total"] if res.get("records") else 0
    return results

def get_ga4_kpi(week_start, week_end):
    from google.analytics.data_v1beta import BetaAnalyticsDataClient
    from google.analytics.data_v1beta.types import DateRange, Metric, Dimension, RunReportRequest
    
    client = BetaAnalyticsDataClient()
    req = RunReportRequest(
        property=f"properties/{GA4_PROPERTY_ID}",
        date_ranges=[DateRange(start_date=week_start.strftime("%Y-%m-%d"),
                               end_date=week_end.strftime("%Y-%m-%d"))],
        metrics=[Metric(name="sessions"),Metric(name="activeUsers"),
                 Metric(name="conversions"),Metric(name="bounceRate")],
    )
    report = client.run_report(req)
    row = report.rows[0]
    return {"セッション数": int(row.metric_values[0].value),
            "アクティブユーザー": int(row.metric_values[1].value),
            "コンバージョン": int(row.metric_values[2].value),
            "直帰率": float(row.metric_values[3].value)}

today = date.today()
last_monday = today - timedelta(days=today.weekday()+7)
last_sunday  = last_monday + timedelta(days=6)

sf_kpi  = get_salesforce_kpi(last_monday)
ga4_kpi = get_ga4_kpi(last_monday, last_sunday)
print(f"KPI収集完了: SF={sf_kpi}, GA4={ga4_kpi}")

STEP 2：前週比・目標達成率を自動計算する

収集したKPIを前週比・目標達成率とともに集計します。

import pandas as pd

def load_weekly_targets(week_str):
    df = pd.read_excel("targets_2026.xlsx", sheet_name="週次目標")
    row = df[df["週"] == week_str]
    return row.iloc[0].to_dict() if not row.empty else {}

week_str = f"{last_monday.year}W{last_monday.isocalendar()[1]:02d}"
targets  = load_weekly_targets(week_str)

# 前週データ取得
prev_monday = last_monday - timedelta(days=7)
prev_sf     = get_salesforce_kpi(prev_monday)

def calc_comparison(current, previous, target=None):
    wow = (current - previous) / previous * 100 if previous else 0
    vtg = current / target * 100 if target else None
    return {"current": current, "wow": wow, "vs_target": vtg}

kpi_summary = {
    "売上合計":   calc_comparison(sf_kpi["売上合計"],  prev_sf["売上合計"],  targets.get("売上目標")),
    "新規商談数": calc_comparison(sf_kpi["新規商談数"], prev_sf["新規商談数"], targets.get("商談目標")),
    "Webセッション": calc_comparison(ga4_kpi["セッション数"], 0, targets.get("セッション目標")),
    "CV数":       calc_comparison(ga4_kpi["コンバージョン"], 0, targets.get("CV目標")),
}

for kpi, data in kpi_summary.items():
    print(f"{kpi}: {data['current']:,.0f} (前週比{data['wow']:+.1f}% / 目標{data['vs_target']:.0f}%)" if data['vs_target'] else f"{kpi}: {data['current']:,.0f} (前週比{data['wow']:+.1f}%)")

STEP 3：Adaptive Cardでリッチな投稿を作成する

集計したKPIをTeamsのAdaptive Cardでリッチなカード形式で表示します。

import requests

WEBHOOK_URL = "https://company.webhook.office.com/webhookb2/TEAMS/WEBHOOK"

def make_kpi_badge(rate):
    if rate is None: return "⬜ N/A"
    if rate >= 100:  return "🟢 達成"
    if rate >= 80:   return "🟡 進行中"
    return "🔴 未達"

def make_wow_str(wow):
    if wow > 0: return f"▲{wow:+.1f}%"
    if wow < 0: return f"▼{wow:.1f}%"
    return "→ 0.0%"

def post_weekly_kpi_card(kpi_summary, week_str):
    facts = []
    for kpi_name, data in kpi_summary.items():
        badge = make_kpi_badge(data.get("vs_target"))
        wow   = make_wow_str(data["wow"])
        facts.append({"title": f"{badge} {kpi_name}",
                      "value": f"{data['current']:,.0f}　({wow})"})
    
    card = {
        "type": "message",
        "attachments": [{"contentType": "application/vnd.microsoft.card.adaptive",
            "content": {
                "type": "AdaptiveCard", "version": "1.4",
                "body": [
                    {"type": "TextBlock", "size": "Large", "weight": "Bolder",
                     "text": f"📊 週次KPIレポート {week_str}", "color": "Accent"},
                    {"type": "TextBlock", "text": f"集計期間: {last_monday} 〜 {last_sunday}",
                     "size": "Small", "color": "Good"},
                    {"type": "FactSet", "facts": facts},
                    {"type": "ActionSet", "actions": [
                        {"type": "Action.OpenUrl", "title": "詳細レポートを見る",
                         "url": "https://dashboard.company.com/weekly"},
                    ]}
                ]
            }
        }]
    }
    res = requests.post(WEBHOOK_URL, json=card)
    print(f"KPIカード投稿: {'OK' if res.status_code==200 else res.status_code}")

post_weekly_kpi_card(kpi_summary, week_str)

STEP 4：未達KPIの担当者にメンションする

目標未達のKPIの担当者をメンションして対策を促します。

KPI_OWNERS = {
    "売上合計":    "tanaka@co.jp",
    "新規商談数":  "suzuki@co.jp",
    "Webセッション":"yamada@co.jp",
    "CV数":        "nakamura@co.jp",
}

def get_teams_user_id(email, token):
    res = requests.get(f"https://graph.microsoft.com/v1.0/users/{email}",
        headers={"Authorization": f"Bearer {token}"})
    return res.json().get("id","")

def mention_underperformers(kpi_summary, token):
    underperformers = {kpi: data for kpi, data in kpi_summary.items()
                       if data.get("vs_target") is not None and data["vs_target"] < 80}
    
    if not underperformers:
        requests.post(WEBHOOK_URL, json={"text": "🎉 全KPIが目標80%以上達成中です！引き続き頑張りましょう！"})
        return
    
    mentions = []
    entities = []
    for kpi, data in underperformers.items():
        owner = KPI_OWNERS.get(kpi)
        if not owner: continue
        user_id = get_teams_user_id(owner, token)
        mentions.append(f"{owner.split('@')[0]} さん【{kpi}が{data['vs_target']:.0f}%未達】")
        entities.append({"type":"mention","text":f"{owner.split('@')[0]}",
                         "mentioned":{"id":user_id,"name":owner.split('@')[0]}})
    
    if mentions:
        msg = "⚠️ 以下のKPIが目標を大幅に下回っています。今週中に原因分析と対策をご共有ください。

" + "
".join(mentions)
        requests.post(WEBHOOK_URL, json={"body":{"content":msg,"contentType":"html"},"mentions":entities})
        print(f"メンション送信: {len(underperformers)}件の未達KPI")

mention_underperformers(kpi_summary, token)

STEP 5：毎週月曜9時に自動実行する

毎週月曜の朝9時に自動でKPIが集計・投稿される仕組みを完成させます。

import schedule, time

def weekly_kpi_job():
    today = date.today()
    if today.weekday() != 0: return  # 月曜のみ
    
    last_mon = today - timedelta(days=7)
    last_sun = last_mon + timedelta(days=6)
    week_str = f"{last_mon.year}W{last_mon.isocalendar()[1]:02d}"
    
    print(f"週次KPIレポート自動生成: {week_str}")
    try:
        # データ収集
        sf_kpi  = get_salesforce_kpi(last_mon)
        ga4_kpi = get_ga4_kpi(last_mon, last_sun)
        
        # 集計
        targets = load_weekly_targets(week_str)
        prev_sf = get_salesforce_kpi(last_mon - timedelta(days=7))
        kpi_summary = {
            "売上合計":   calc_comparison(sf_kpi["売上合計"],  prev_sf["売上合計"],  targets.get("売上目標")),
            "新規商談数": calc_comparison(sf_kpi["新規商談数"], prev_sf["新規商談数"], targets.get("商談目標")),
            "Webセッション": calc_comparison(ga4_kpi["セッション数"], 0, targets.get("セッション目標")),
        }
        
        # Teamsに投稿
        post_weekly_kpi_card(kpi_summary, week_str)
        mention_underperformers(kpi_summary, token)
        print(f"完了: {week_str}")
    
    except Exception as e:
        # エラー時はSlackにアラート
        requests.post(SLACK_WEBHOOK, json={"text": f"🚨 KPIレポート自動投稿が失敗しました: {e}"})

schedule.every().monday.at("09:00").do(weekly_kpi_job)
while True: schedule.run_pending(); time.sleep(60)

制作実績サンプル

このクオリティの資料が、あなたの業務でも作れます

Aurant Technologies が実際に制作したコンサルティング資料（全82枚）のサンプルです。Claude Code × AI自動化により、従来の1/5のコスト・時間で同等品質を実現します。

タイトル・ダーク背景

3カラム比較・分析

AI活用フロー図解

Before/After比較

構造化コンテンツ

→

全10枚を
見る

御社のテーマでこのレベルの資料を作成できます

制作費・納期・自動化プランは無料相談でご確認いただけます

📊 サンプル全10枚を見る →