AIペアアーキテクトのすすめ（第3回）

設計が固まったら次は実装。AWS Kiroを共に考える相棒として活用する手法です。

前回
コードをAIと一緒に育てる：設計から実装、リファクタリングまで
前回までのおさらい
設計をもとにコードを生成させる
kiroが生成した成果物
まとめ ― コード生成はゴールではない
1. 総まとめ

前回

AIペアアーキテクト実践｜Slack通知ツール設計をKiroと相談（第2回）

PythonでSlack日次レポートツールを作る例を通して、AWS Kiroを使った設計相談の実践法を解説。関数分割やディレクトリ構成、再送処理などをAIと検討します。

コードをAIと一緒に育てる：設計から実装、リファクタリングまで

設計が固まったら、次は実装フェーズだ。
ここでもAIは「自動で書く」ではなく、「一緒に考える」相棒になる。

前回までのおさらい

第1回ではAIを“ペアアーキテクト”として設計を相談する考え方を紹介し、第2回ではKiroを使ってSlack通知ツールの設計を詰めました。

今回はその続きとして、実際にコードを生成し、AIと一緒に実装 → テスト → リファクタリングを進めるプロセスをやってみます。

設計をもとにコードを生成させる

設計が明確になった今、初めてAIに「コードを書いて」と依頼します。

レポート内容は固定テキストでサンプル実装。
環境変数は提案の通りでOK。
エラー時はログファイル出力のみでOK.
Pythonは3.12を想定。
cli実行は `python3 daily-report-to-slack.py` の方法で。

了解しました。daily-report-to-slackパッケージを実装していきます。

プロジェクト構成
daily-report-to-slack/
├── daily_report_to_slack/
│ ├── __init__.py
│ ├── client.py
│ └── reporter.py
├── daily-report-to-slack.py # CLI実行ファイル
├── pyproject.toml
├── requirements.txt
├── README.md
└── .gitignore

この構成で実装を開始してもよろしいでしょうか？

実装順序：
1.プロジェクト構造作成
2.Slack APIクライアント（Rate Limit対応付き）
3.レポート生成・投稿機能
4.CLI実行ファイル
5.パッケージ設定ファイル
進めてもよろしいですか？

ここから上記の1～5に沿って1段階ずつレビューをしながらkiroが実装をすすめてくれました。
ここで重要なのは、すぐに受け入れないことです。
作成されたものをレビューしてから次の段階へ進むようにしましょう。
一気に実装が終わってからレビューで指摘すると思いがけないところを変更される場合もあります。

kiroが生成した成果物

daily_report_to_slack.py

#!/usr/bin/env python3
"""
Daily Report to Slack - CLI execution script.

Usage:
    python3 daily-report-to-slack.py [options]

Environment Variables:
    SLACK_BOT_TOKEN    - Slack Bot User OAuth Token (required)
    SLACK_CHANNEL      - Slack channel to post to (required)
    REPORT_TIMEZONE    - Timezone for report (optional, default: UTC)
"""

import argparse
import logging
import sys
from pathlib import Path

# Add the package to Python path
sys.path.insert(0, str(Path(__file__).parent))

from daily_report_to_slack import DailyReporter


def setup_logging(log_level: str = "INFO", log_file: str = None) -> None:
    """
    Setup logging configuration.

    Args:
        log_level: Logging level (DEBUG, INFO, WARNING, ERROR)
        log_file: Optional log file path
    """
    # Configure log format
    log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    date_format = "%Y-%m-%d %H:%M:%S"

    # Setup handlers
    handlers = [logging.StreamHandler(sys.stdout)]

    if log_file:
        # Ensure log directory exists
        log_path = Path(log_file)
        log_path.parent.mkdir(parents=True, exist_ok=True)
        handlers.append(logging.FileHandler(log_file))

    # Configure logging
    logging.basicConfig(
        level=getattr(logging, log_level.upper()),
        format=log_format,
        datefmt=date_format,
        handlers=handlers,
        force=True
    )


def main():
    """Main CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Send daily report to Slack channel",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Environment Variables:
  SLACK_BOT_TOKEN     Slack Bot User OAuth Token (required)
  SLACK_CHANNEL       Slack channel to post to (required)  
  REPORT_TIMEZONE     Timezone for report (optional, default: UTC)

Examples:
  python3 daily-report-to-slack.py
  python3 daily-report-to-slack.py --dry-run
  python3 daily-report-to-slack.py --test-connection
  python3 daily-report-to-slack.py --log-level DEBUG --log-file ./logs/daily-report.log
        """
    )

    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Generate report content but don't send to Slack"
    )

    parser.add_argument(
        "--test-connection",
        action="store_true", 
        help="Test Slack connection and exit"
    )

    parser.add_argument(
        "--preview",
        action="store_true",
        help="Show report preview and exit"
    )

    parser.add_argument(
        "--log-level",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        default="INFO",
        help="Set logging level (default: INFO)"
    )

    parser.add_argument(
        "--log-file",
        help="Log file path (optional, logs to stdout if not specified)"
    )

    parser.add_argument(
        "--token",
        help="Slack Bot User OAuth Token (overrides SLACK_BOT_TOKEN env var)"
    )

    parser.add_argument(
        "--channel", 
        help="Slack channel (overrides SLACK_CHANNEL env var)"
    )

    parser.add_argument(
        "--timezone",
        help="Timezone for report (overrides REPORT_TIMEZONE env var)"
    )

    args = parser.parse_args()

    # Setup logging
    setup_logging(args.log_level, args.log_file)
    logger = logging.getLogger(__name__)

    try:
        logger.info("Starting daily-report-to-slack")

        # Initialize reporter
        reporter = DailyReporter(
            token=args.token,
            channel=args.channel, 
            timezone=args.timezone
        )

        # Handle different modes
        if args.test_connection:
            logger.info("Testing Slack connection...")
            if reporter.test_connection():
                logger.info("✅ Connection test successful")
                sys.exit(0)
            else:
                logger.error("❌ Connection test failed")
                sys.exit(1)

        elif args.preview:
            logger.info("Generating report preview...")
            preview = reporter.get_report_preview()
            print("\n" + "="*50)
            print("REPORT PREVIEW")
            print("="*50)
            print(preview)
            print("="*50)
            sys.exit(0)

        else:
            # Send report (with optional dry-run)
            if args.dry_run:
                logger.info("Running in dry-run mode...")

            success = reporter.send_report(dry_run=args.dry_run)

            if success:
                if args.dry_run:
                    logger.info("✅ Dry-run completed successfully")
                else:
                    logger.info("✅ Daily report sent successfully")
                sys.exit(0)
            else:
                logger.error("❌ Failed to send daily report")
                sys.exit(1)

    except ValueError as e:
        logger.error(f"Configuration error: {e}")
        logger.error("Please check your environment variables or command line arguments")
        sys.exit(1)

    except KeyboardInterrupt:
        logger.info("Operation cancelled by user")
        sys.exit(1)

    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        logger.debug("Full traceback:", exc_info=True)
        sys.exit(1)


if __name__ == "__main__":
    main()

daily_report_to_slack/main.py

"""
Entry point for package execution.
Allows running the package with: python -m daily_report_to_slack
"""

import sys
from pathlib import Path

# Add parent directory to path for CLI script import
parent_dir = Path(__file__).parent.parent
sys.path.insert(0, str(parent_dir))

# Import and run the CLI main function
try:
    from daily_report_to_slack.cli import main
    main()
except ImportError:
    # Fallback to direct script execution
    import subprocess
    script_path = parent_dir / "daily-report-to-slack.py"
    subprocess.run([sys.executable, str(script_path)] + sys.argv[1:])

daily_report_to_slack/init.py

"""
Daily Report to Slack - A simple tool for posting daily reports to Slack channels.
"""

__version__ = "0.1.0"
__author__ = "Your Name"

from .client import SlackClient
from .reporter import DailyReporter

__all__ = ["SlackClient", "DailyReporter"]

daily_report_to_slack/client.py

"""
Slack API client with rate limiting and error handling.
"""

import json
import logging
import random
import time
from typing import Dict, Any, Optional

import requests


class SlackAPIError(Exception):
    """Base exception for Slack API errors."""
    pass


class SlackRateLimitError(SlackAPIError):
    """Exception raised when rate limit is exceeded."""

    def __init__(self, message: str, retry_after: Optional[int] = None):
        super().__init__(message)
        self.retry_after = retry_after


class SlackAuthenticationError(SlackAPIError):
    """Exception raised for authentication errors."""
    pass


class SlackClient:
    """Slack API client with exponential backoff retry logic."""

    def __init__(self, token: str, timeout: int = 30):
        """
        Initialize Slack client.

        Args:
            token: Slack Bot User OAuth Token
            timeout: Request timeout in seconds
        """
        self.token = token
        self.timeout = timeout
        self.base_url = "https://slack.com/api"
        self.logger = logging.getLogger(__name__)

        # Setup session with default headers
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json; charset=utf-8"
        })

    def _exponential_backoff_retry(self, func, max_retries: int = 3, base_delay: float = 1.0):
        """
        Execute function with exponential backoff retry.

        Args:
            func: Function to execute
            max_retries: Maximum number of retries
            base_delay: Base delay in seconds

        Returns:
            Function result

        Raises:
            SlackAPIError: When all retries are exhausted
        """
        for attempt in range(max_retries + 1):
            try:
                return func()
            except SlackRateLimitError as e:
                if attempt == max_retries:
                    self.logger.error(f"Rate limit exceeded after {max_retries} retries")
                    raise

                # Use Retry-After header if available, otherwise exponential backoff
                if e.retry_after:
                    delay = float(e.retry_after)
                    self.logger.warning(f"Rate limited, waiting {delay}s (from Retry-After header)")
                else:
                    delay = base_delay * (2 ** attempt)
                    # Add jitter to avoid thundering herd
                    jitter = random.uniform(0, 0.1 * delay)
                    delay += jitter
                    self.logger.warning(f"Rate limited, waiting {delay:.2f}s (attempt {attempt + 1})")

                time.sleep(delay)
            except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
                if attempt == max_retries:
                    self.logger.error(f"Network error after {max_retries} retries: {e}")
                    raise SlackAPIError(f"Network error: {e}")

                delay = base_delay * (2 ** attempt)
                self.logger.warning(f"Network error, retrying in {delay}s: {e}")
                time.sleep(delay)

    def _make_request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """
        Make HTTP request to Slack API.

        Args:
            method: HTTP method
            endpoint: API endpoint
            data: Request data

        Returns:
            API response as dictionary

        Raises:
            SlackAPIError: For various API errors
        """
        url = f"{self.base_url}/{endpoint}"

        def _request():
            try:
                if method.upper() == "POST":
                    response = self.session.post(url, json=data, timeout=self.timeout)
                else:
                    response = self.session.get(url, params=data, timeout=self.timeout)

                # Handle HTTP status codes
                if response.status_code == 429:
                    retry_after = response.headers.get("Retry-After")
                    raise SlackRateLimitError("Rate limit exceeded", retry_after)
                elif response.status_code == 401:
                    raise SlackAuthenticationError("Invalid token or insufficient permissions")
                elif response.status_code >= 500:
                    raise SlackAPIError(f"Server error: {response.status_code}")
                elif response.status_code >= 400:
                    raise SlackAPIError(f"Client error: {response.status_code} - {response.text}")

                response.raise_for_status()

                # Parse JSON response
                try:
                    result = response.json()
                except json.JSONDecodeError as e:
                    raise SlackAPIError(f"Invalid JSON response: {e}")

                # Check Slack API error
                if not result.get("ok", False):
                    error = result.get("error", "Unknown error")
                    if error == "ratelimited":
                        raise SlackRateLimitError("Rate limit exceeded")
                    elif error in ["invalid_auth", "account_inactive", "token_revoked"]:
                        raise SlackAuthenticationError(f"Authentication error: {error}")
                    else:
                        raise SlackAPIError(f"Slack API error: {error}")

                return result

            except requests.exceptions.Timeout:
                raise SlackAPIError(f"Request timeout after {self.timeout}s")
            except requests.exceptions.ConnectionError as e:
                raise SlackAPIError(f"Connection error: {e}")

        return self._exponential_backoff_retry(_request)

    def post_message(self, channel: str, text: str, **kwargs) -> Dict[str, Any]:
        """
        Post message to Slack channel.

        Args:
            channel: Channel name or ID
            text: Message text
            **kwargs: Additional message parameters

        Returns:
            API response
        """
        data = {
            "channel": channel,
            "text": text,
            **kwargs
        }

        self.logger.info(f"Posting message to channel: {channel}")
        result = self._make_request("POST", "chat.postMessage", data)
        self.logger.info("Message posted successfully")

        return result

    def test_auth(self) -> Dict[str, Any]:
        """
        Test authentication.

        Returns:
            Auth test response
        """
        self.logger.info("Testing authentication")
        result = self._make_request("GET", "auth.test")
        self.logger.info(f"Authentication successful for user: {result.get('user')}")

        return result

daily_report_to_slack/reporter.py

"""
Daily report generator and poster.
"""

import logging
import os
from datetime import datetime
from typing import Optional
from zoneinfo import ZoneInfo

from .client import SlackClient, SlackAPIError


class DailyReporter:
    """Daily report generator and Slack poster."""

    def __init__(self, 
                 token: Optional[str] = None,
                 channel: Optional[str] = None,
                 timezone: Optional[str] = None):
        """
        Initialize daily reporter.

        Args:
            token: Slack Bot User OAuth Token (defaults to SLACK_BOT_TOKEN env var)
            channel: Slack channel (defaults to SLACK_CHANNEL env var)
            timezone: Timezone for report (defaults to REPORT_TIMEZONE env var or UTC)
        """
        self.token = token or os.getenv("SLACK_BOT_TOKEN")
        self.channel = channel or os.getenv("SLACK_CHANNEL")
        self.timezone = timezone or os.getenv("REPORT_TIMEZONE", "UTC")

        self.logger = logging.getLogger(__name__)

        # Validate required parameters
        if not self.token:
            raise ValueError("Slack token is required. Set SLACK_BOT_TOKEN environment variable or pass token parameter.")

        if not self.channel:
            raise ValueError("Slack channel is required. Set SLACK_CHANNEL environment variable or pass channel parameter.")

        # Initialize Slack client
        self.client = SlackClient(self.token)

        self.logger.info(f"DailyReporter initialized for channel: {self.channel}, timezone: {self.timezone}")

    def _get_current_datetime(self) -> datetime:
        """
        Get current datetime in configured timezone.

        Returns:
            Current datetime with timezone
        """
        try:
            tz = ZoneInfo(self.timezone)
            return datetime.now(tz)
        except Exception as e:
            self.logger.warning(f"Invalid timezone '{self.timezone}', using UTC: {e}")
            return datetime.now(ZoneInfo("UTC"))

    def _generate_report_content(self) -> str:
        """
        Generate daily report content.

        Returns:
            Report content as string
        """
        current_time = self._get_current_datetime()
        date_str = current_time.strftime("%Y-%m-%d")
        time_str = current_time.strftime("%H:%M:%S %Z")

        # Sample report content
        report = f"""📊 **Daily Report - {date_str}**

🕐 **Report Time**: {time_str}

📈 **System Status**
• All services: ✅ Operational
• Database: ✅ Healthy
• API Response Time: 150ms avg
• Uptime: 99.9%

📊 **Key Metrics**
• Active Users: 1,234
• New Registrations: 56
• Total Transactions: 789
• Revenue: $12,345

🔍 **Issues & Actions**
• No critical issues reported
• Scheduled maintenance: None
• Next review: {(current_time.replace(hour=9, minute=0, second=0, microsecond=0) + current_time.resolution).strftime("%Y-%m-%d %H:%M")}

---
*Generated automatically by daily-report-to-slack*"""

        return report

    def test_connection(self) -> bool:
        """
        Test Slack connection and authentication.

        Returns:
            True if connection is successful, False otherwise
        """
        try:
            self.logger.info("Testing Slack connection...")
            auth_result = self.client.test_auth()

            user = auth_result.get("user", "Unknown")
            team = auth_result.get("team", "Unknown")

            self.logger.info(f"Connection successful - User: {user}, Team: {team}")
            return True

        except SlackAPIError as e:
            self.logger.error(f"Slack connection failed: {e}")
            return False
        except Exception as e:
            self.logger.error(f"Unexpected error during connection test: {e}")
            return False

    def send_report(self, dry_run: bool = False) -> bool:
        """
        Generate and send daily report to Slack.

        Args:
            dry_run: If True, generate report but don't send to Slack

        Returns:
            True if successful, False otherwise
        """
        try:
            self.logger.info("Generating daily report...")
            report_content = self._generate_report_content()

            if dry_run:
                self.logger.info("DRY RUN - Report content:")
                self.logger.info(f"\n{report_content}")
                return True

            self.logger.info(f"Sending report to channel: {self.channel}")

            # Send message to Slack
            result = self.client.post_message(
                channel=self.channel,
                text=report_content,
                unfurl_links=False,
                unfurl_media=False
            )

            message_ts = result.get("ts")
            self.logger.info(f"Report sent successfully - Message timestamp: {message_ts}")

            return True

        except SlackAPIError as e:
            self.logger.error(f"Failed to send report to Slack: {e}")
            return False
        except Exception as e:
            self.logger.error(f"Unexpected error while sending report: {e}")
            return False

    def send_daily_report(self) -> bool:
        """
        Convenience method to send daily report.
        Alias for send_report().

        Returns:
            True if successful, False otherwise
        """
        return self.send_report()

    def get_report_preview(self) -> str:
        """
        Get preview of report content without sending.

        Returns:
            Report content as string
        """
        return self._generate_report_content()

まとめ ― コード生成はゴールではない

AIが書くコードは出発点にすぎません。本当に価値があるのは、AIとの会話を通じて設計意図を明確にするプロセスじゃないでしょうか。

今回のポイントは：

AIにコードを書かせたら、必ず理由を聞く
リファクタリングの方向を相談する
テスト戦略をAIと共に設計する
拡張を見据えた抽象化を議論する

こうしたサイクルを回すことで、AIは単なる生成ツールではなく、共に設計を育てる相棒になってくれることでしょう。

総まとめ

全3回を通じて見てきたのは、

フェーズ	人間の役割	AIの役割
設計初期	要件を明確化し、前提を伝える	実装方針の比較と提案
実装中	判断・採用・修正	コード生成・理由説明
運用設計	再利用や抽象化を検討	改善・最適化の助言