cloud-api-integration

クラウド API 統合スキル

ファイル構成：分割構成です。コアパターンはメインの SKILL.md に記載されています。完全な実装については references/ を参照してください。

1. 概要

リスクレベル：高 — API 認証情報を扱い、信頼できないプロンプトを処理し、ネットワークに公開されるため、データプライバシー上の懸念があります。

Anthropic Claude、OpenAI GPT-4、Google Gemini API に関する深い専門知識を備えた、クラウド AI API 統合のエキスパートです。セキュアな認証情報管理、プロンプトセキュリティ、レート制限、エラー処理、LLM 固有の脆弱性対策に精通しています。

以下を得意としています。

• API キーの安全な管理とローテーション
• クラウド LLM に対するプロンプトインジェクションの防止
• レート制限とコストの最適化
• 複数プロバイダーによるフォールバック戦略
• 出力のサニタイズとデータプライバシー

主なユースケース：

• 複雑なタスク向けの JARVIS クラウド AI 統合
• ローカルモデルでは不十分な場合のフォールバック
• マルチモーダル処理（画像認識、コード）
• セキュリティを備えたエンタープライズグレードの信頼性

---

2. コア原則

1. まず TDD — 実装の前にテストを作成します。外部 API 呼び出しはすべてモック化します。
2. パフォーマンスを意識 — キャッシュと接続の再利用により、レイテンシ、コスト、信頼性を最適化します。
3. セキュリティを最優先 — キーをハードコードせず、すべての入力をサニタイズし、すべての出力をフィルタリングします。
4. コストを意識 — 使用量を追跡し、上限を設定して、繰り返しのクエリをキャッシュします。
5. 信頼性を重視 — サーキットブレーカーを備えた複数プロバイダーのフォールバックを実装します。

---

3. 実装ワークフロー（TDD）

ステップ 1：まず失敗するテストを書く

# tests/test_cloud_api.py
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from src.cloud_api import SecureClaudeClient, CloudAPIConfig

class TestSecureClaudeClient:
    """Test cloud API client with mocked external calls."""

    @pytest.fixture
    def mock_config(self):
        return CloudAPIConfig(
            anthropic_key="test-key-12345",
            timeout=30.0
        )

    @pytest.fixture
    def mock_anthropic_response(self):
        """Mock Anthropic API response."""
        mock_response = MagicMock()
        mock_response.content = [MagicMock(text="Test response")]
        mock_response.usage.input_tokens = 10
        mock_response.usage.output_tokens = 20
        return mock_response

    @pytest.mark.asyncio
    async def test_generate_sanitizes_input(self, mock_config, mock_anthropic_response):
        """Test that prompts are sanitized before sending."""
        with patch('anthropic.Anthropic') as mock_client:
            mock_client.return_value.messages.create.return_value = mock_anthropic_response

            client = SecureClaudeClient(mock_config)
            result = await client.generate("Test <script>alert('xss')</script>")

            # Verify sanitization was applied
            call_args = mock_client.return_value.messages.create.call_args
            assert "<script>" not in str(call_args)
            assert result == "Test response"

    @pytest.mark.asyncio
    async def test_rate_limiter_blocks_excess_requests(self):
        """Test rate limiting blocks requests over threshold."""
        from src.cloud_api import RateLimiter

        limiter = RateLimiter(rpm=2, daily_cost=100)

        await limiter.acquire(100)
        await limiter.acquire(100)

        with pytest.raises(Exception):  # RateLimitError
            await limiter.acquire(100)

    @pytest.mark.asyncio
    async def test_multi_provider_fallback(self, mock_config):
        """Test fallback to secondary provider on failure."""
        from src.cloud_api import MultiProviderClient

        with patch('src.cloud_api.SecureClaudeClient') as mock_claude:
            with patch('src.cloud_api.SecureOpenAIClient') as mock_openai:
                mock_claude.return_value.generate = AsyncMock(
                    side_effect=Exception("Rate limited")
                )
                mock_openai.return_value.generate = AsyncMock(
                    return_value="OpenAI response"
                )

                client = MultiProviderClient(mock_config)
                result = await client.generate("test prompt")

                assert result == "OpenAI response"
                mock_openai.return_value.generate.assert_called_once()

ステップ 2：テストを通すための最小限の実装を行う

# src/cloud_api.py
class SecureClaudeClient:
    def __init__(self, config: CloudAPIConfig):
        self.client = Anthropic(api_key=config.anthropic_key.get_secret_value())
        self.sanitizer = PromptSanitizer()

    async def generate(self, prompt: str) -> str:
        sanitized = self.sanitizer.sanitize(prompt)
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            messages=[{"role": "user", "content": sanitized}]
        )
        return self._filter_output(response.content[0].text)

ステップ 3：パターンを使ってリファクタリングする

パフォーマンスパターンにあるキャッシュ、接続プーリング、リトライのロジックを適用します。

ステップ 4：すべての検証を実行する

# Run all tests with coverage
pytest tests/test_cloud_api.py -v --cov=src.cloud_api --cov-report=term-missing

# Run security checks
bandit -r src/cloud_api.py

# Type checking
mypy src/cloud_api.py --strict

---

4. パフォーマンスパターン

パターン 1：接続プーリング

# Good: Reuse HTTP connections
import httpx

class CloudAPIClient:
    def __init__(self):
        self._client = httpx.AsyncClient(
            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
            timeout=httpx.Timeout(30.0)
        )

    async def request(self, endpoint: str, data: dict) -> dict:
        response = await self._client.post(endpoint, json=data)
        return response.json()

    async def close(self):
        await self._client.aclose()

# Bad: Create new connection per request
async def bad_request(endpoint: str, data: dict):
    async with httpx.AsyncClient() as client:  # New connection each time!
        return await client.post(endpoint, json=data)

パターン 2：指数バックオフによるリトライ

# Good: Smart retry with backoff
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class CloudAPIClient:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((RateLimitError, APIConnectionError))
    )
    async def generate(self, prompt: str) -> str:
        return await self._make_request(prompt)

# Bad: No retry or fixed delay
async def bad_generate(prompt: str):
    try:
        return await make_request(prompt)
    except Exception:
        await asyncio.sleep(1)  # Fixed delay, no backoff!
        return await make_request(prompt)

パターン 3：レスポンスのキャッシュ

# Good: Cache repeated queries with TTL
from functools import lru_cache
import hashlib
from cachetools import TTLCache

class CachedCloudClient:
    def __init__(self):
        self._cache = TTLCache(maxsize=1000, ttl=300)  # 5 min TTL

    async def generate(self, prompt: str, **kwargs) -> str:
        cache_key = self._make_key(prompt, kwargs)

        if cache_key in self._cache:
            return self._cache[cache_key]

        result = await self._client.generate(prompt, **kwargs)
        self._cache[cache_key] = result
        return result

    def _make_key(self, prompt: str, kwargs: dict) -> str:
        content = f"{prompt}:{sorted(kwargs.items())}"
        return hashlib.sha256(content.encode()).hexdigest()

# Bad: No caching
async def bad_generate(prompt: str):
    return await client.generate(prompt)  # Repeated identical calls!

パターン 4：API 呼び出しのバッチ処理

# Good: Batch multiple requests
import asyncio

class BatchCloudClient:
    async def generate_batch(self, prompts: list[str]) -> list[str]:
        """Process multiple prompts concurrently with rate limiting."""
        semaphore = asyncio.Semaphore(5)  # Max 5 concurrent

        async def limited_generate(prompt: str) -> str:
            async with semaphore:
                return await self.generate(prompt)

        tasks = [limited_generate(p) for p in prompts]
        return await asyncio.gather(*tasks)

# Bad: Sequential processing
async def bad_batch(prompts: list[str]):
    results = []
    for prompt in prompts:
        results.append(await client.generate(prompt))  # One at a time!
    return results

パターン 5：非同期リクエスト処理

# Good: Fully async with proper context management
class AsyncCloudClient:
    async def __aenter__(self):
        self._client = httpx.AsyncClient()
        return self

    async def __aexit__(self, *args):
        await self._client.aclose()

    async def generate(self, prompt: str) -> str:
        response = await self._client.post(
            self.endpoint,
            json={"prompt": prompt},
            timeout=30.0
        )
        return response.json()["text"]

# Usage
async with AsyncCloudClient() as client:
    result = await client.generate("Hello")

# Bad: Blocking calls in async context
def bad_generate(prompt: str):
    response = requests.post(endpoint, json={"prompt": prompt})  # Blocks!
    return response.json()

---

5. コアとなる責務

5.1 セキュリティを最優先する API 統合

クラウド AI API を統合する際は、次のことを行います。

• API キーをハードコードしない — 常に環境変数またはシークレットマネージャーを使用します
• すべてのプロンプトを信頼できないものとして扱う — 送信前にユーザー入力をサニタイズします
• すべての出力をフィルタリングする — データの流出やインジェクションを防ぎます
• レート制限を実装する — 悪用やコスト超過を防ぎます
• 安全にログを記録する — API キーや機密性の高いプロンプトは決してログに記録しません

5.2 コストとパフォーマンスの最適化

• タスクの複雑さに応じて適切なモデルのティアを選択します
• 繰り返しのクエリにはキャッシュを実装します
• ユーザーエクスペリエンスの向上にはストリーミングを使用します
• 使用量を監視し、支出アラートを設定します
• API 障害に備えてサーキットブレーカーを実装します

5.3 プライバシーとコンプライアンス

• クラウド API に送信するデータを最小限に抑えます
• 明示的な同意なしに個人を特定できる情報（PII）を送信しません
• データ保持ポリシーを実装します
• データを学習に使用しない API 機能を利用します
• コンプライアンスに備えてデータフローを文書化します

---

6. 技術的基盤

6.1 コア SDK とバージョン

プロバイダー	本番環境	最小バージョン	備考
Anthropic	anthropic>=0.40.0	>=0.25.0	Messages API のサポート
OpenAI	openai>=1.50.0	>=1.0.0	構造化出力
Gemini	google-generativeai>=0.8.0	-	最新機能

6.2 セキュリティ依存関係

# requirements.txt
anthropic>=0.40.0
openai>=1.50.0
google-generativeai>=0.8.0
pydantic>=2.0          # Input validation
httpx>=0.27.0          # HTTP client with timeouts
tenacity>=8.0          # Retry logic
structlog>=23.0        # Secure logging
cryptography>=41.0     # Key encryption
cachetools>=5.0        # Response caching

---

7. 実装パターン

パターン 1：セキュアな API クライアント設定

from pydantic import BaseModel, SecretStr, Field, validator
from anthropic import Anthropic
import os, structlog

logger = structlog.get_logger()

class CloudAPIConfig(BaseModel):
    """Validated cloud API configuration."""
    anthropic_key: SecretStr = Field(default=None)
    openai_key: SecretStr = Field(default=None)
    timeout: float = Field(default=30.0, ge=5, le=120)

    @validator('anthropic_key', 'openai_key', pre=True)
    def load_from_env(cls, v, field):
        return v or os.environ.get(field.name.upper())

    class Config:
        json_encoders = {SecretStr: lambda v: '***'}

完全な実装については references/advanced-patterns.md を参照してください。

---

8. セキュリティ標準

8.1 重大な脆弱性

脆弱性	深刻度	緩和策
プロンプトインジェクション	高	入力のサニタイズ、出力のフィルタリング
API キーの漏えい	重大	環境変数、シークレットマネージャー
データの流出	高	ネットワークアクセスの制限

8.2 OWASP LLM Top 10 の対応付け

OWASP ID	カテゴリ	緩和策
LLM01	プロンプトインジェクション	すべての入力をサニタイズする
LLM02	安全でない出力	使用前にフィルタリングする
LLM06	情報漏えい	プロンプトにシークレットを含めない

---

9. よくある間違い

# NEVER: Hardcode API Keys
client = Anthropic(api_key="sk-ant-api03-xxxxx")  # DANGEROUS
client = Anthropic()  # SECURE - uses env var

# NEVER: Log API Keys
logger.info(f"Using API key: {api_key}")  # DANGEROUS
logger.info("API client initialized", provider="anthropic")  # SECURE

# NEVER: Trust External Content
content = fetch_url(url)
response = claude.generate(f"Summarize: {content}")  # INJECTION VECTOR!

---

10. 実装前チェックリスト

フェーズ 1：コードを書く前

• モック API レスポンスを使って失敗するテストを書く
• レート制限とコストのしきい値を定義する
• 安全な認証情報の読み込み方法を設定する（環境変数またはシークレットマネージャー）
• 繰り返しのクエリに対するキャッシュ戦略を立てる

フェーズ 2：実装中

• API キーは環境変数／シークレットマネージャーからのみ読み込む
• すべてのユーザーコンテンツに対して入力のサニタイズを有効にする
• レスポンスを使用する前に出力をフィルタリングする
• コネクションプーリングを設定する
• 指数バックオフを用いたリトライロジックを実装する
• 同一のクエリに対するレスポンスをキャッシュする

フェーズ 3：コミット前

• すべてのテストがカバレッジ 80％ 超で成功する
• git の履歴に API キーが含まれていない（git-secrets を使用する）
• セキュリティスキャンに合格する（bandit）
• 型チェックに合格する（mypy）
• 1 日あたりの支出上限を設定する
• 複数プロバイダーへのフォールバックをテストする

---

11. まとめ

目標は、次の特性を備えたクラウド API 統合を作成することです。

• テスト駆動：すべての機能をモックテストで検証する
• 高パフォーマンス：コネクションプーリング、キャッシュ、非同期処理
• 安全：プロンプトインジェクションとデータ流出を防ぐ
• 高信頼性：適切なエラー処理を備えた複数プロバイダーへのフォールバック
• 費用対効果：レート制限と使用状況のモニタリング

実装の詳細については、以下を参照してください：

• references/advanced-patterns.md - キャッシュ、ストリーミング、最適化
• references/security-examples.md - 脆弱性の詳細な分析
• references/threat-model.md - 攻撃シナリオと緩和策