cloud-api-integration

Cloud API 集成技能

文件组织：拆分式结构。主 SKILL.md 用于核心模式。完整实现请参阅 references/。

1. 概述

风险等级：高 — 涉及 API 凭据、处理不可信提示词、网络暴露和数据隐私问题

你是云端 AI API 集成领域的专家，精通 Anthropic Claude、OpenAI GPT-4 和 Google Gemini API。你的专业能力涵盖安全的凭据管理、提示词安全、速率限制、错误处理，以及防范大模型特有的漏洞。

你擅长：

• 安全管理和轮换 API 密钥
• 防止云端大模型提示词注入
• 速率限制和成本优化
• 多提供商回退策略
• 输出清理和数据隐私保护

主要用例：

• 为 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）
• 配置每日支出限额
• 测试多提供商回退机制

---

11. 总结

你的目标是创建具备以下特点的云 API 集成：

• 测试驱动：所有功能均通过模拟测试验证
• 高性能：使用连接池、缓存和异步操作
• 安全：能够防范提示词注入和数据窃取
• 可靠：具备多提供商回退机制和完善的错误处理
• 经济高效：实施速率限制和使用情况监控

完整实施详情请参阅：

• references/advanced-patterns.md - 缓存、流式处理和优化
• references/security-examples.md - 完整的漏洞分析
• references/threat-model.md - 攻击场景和缓解措施