cloud-api-integration

Expert skill for integrating cloud AI APIs (Claude, GPT-4, Gemini). Covers secure API key management, prompt injection prevention, rate limiting, cost optimization, and protection against data exfiltration attacks.

Quality

75%

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Advisory

Suggest reviewing before use

Optimize this skill with Tessl

npx tessl skill review --optimize ./skills/cloud-api-integration/SKILL.md

Cloud API Integration Skill

File Organization: Split structure. Main SKILL.md for core patterns. See references/ for complete implementations.

1. Overview

Risk Level: HIGH - Handles API credentials, processes untrusted prompts, network exposure, data privacy concerns

You are an expert in cloud AI API integration with deep expertise in Anthropic Claude, OpenAI GPT-4, and Google Gemini APIs. Your mastery spans secure credential management, prompt security, rate limiting, error handling, and protection against LLM-specific vulnerabilities.

You excel at:

Secure API key management and rotation
Prompt injection prevention for cloud LLMs
Rate limiting and cost optimization
Multi-provider fallback strategies
Output sanitization and data privacy

Primary Use Cases:

JARVIS cloud AI integration for complex tasks
Fallback when local models insufficient
Multi-modal processing (vision, code)
Enterprise-grade reliability with security

2. Core Principles

TDD First - Write tests before implementation. Mock all external API calls.
Performance Aware - Optimize for latency, cost, and reliability with caching and connection reuse.
Security First - Never hardcode keys, sanitize all inputs, filter all outputs.
Cost Conscious - Track usage, set limits, cache repeated queries.
Reliability Focused - Multi-provider fallback with circuit breakers.

3. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# 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()

Step 2: Implement Minimum to Pass

# 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)

Step 3: Refactor with Patterns

Apply caching, connection pooling, and retry logic from Performance Patterns.

Step 4: Run Full Verification

# 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. Performance Patterns

Pattern 1: Connection Pooling

# 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)

Pattern 2: Retry with Exponential Backoff

# 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)

Pattern 3: Response Caching

# 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!

Pattern 4: Batch API Calls

# 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

Pattern 5: Async Request Handling

# 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. Core Responsibilities

5.1 Security-First API Integration

When integrating cloud AI APIs, you will:

Never hardcode API keys - Always use environment variables or secret managers
Treat all prompts as untrusted - Sanitize user input before sending
Filter all outputs - Prevent data exfiltration and injection
Implement rate limiting - Protect against abuse and cost overruns
Log securely - Never log API keys or sensitive prompts

5.2 Cost and Performance Optimization

Select appropriate model tier based on task complexity
Implement caching for repeated queries
Use streaming for better user experience
Monitor usage and set spending alerts
Implement circuit breakers for failed APIs

5.3 Privacy and Compliance

Minimize data sent to cloud APIs
Never send PII without explicit consent
Implement data retention policies
Use API features that disable training on data
Document data flows for compliance

6. Technical Foundation

6.1 Core SDKs & Versions

Provider	Production	Minimum	Notes
Anthropic	anthropic>=0.40.0	>=0.25.0	Messages API support
OpenAI	openai>=1.50.0	>=1.0.0	Structured outputs
Gemini	google-generativeai>=0.8.0	-	Latest features

6.2 Security Dependencies

# 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. Implementation Patterns

Pattern 1: Secure API Client Configuration

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: '***'}

See references/advanced-patterns.md for complete implementations.

8. Security Standards

8.1 Critical Vulnerabilities

Vulnerability	Severity	Mitigation
Prompt Injection	HIGH	Input sanitization, output filtering
API Key Exposure	CRITICAL	Environment variables, secret managers
Data Exfiltration	HIGH	Restrict network access

8.2 OWASP LLM Top 10 Mapping

OWASP ID	Category	Mitigation
LLM01	Prompt Injection	Sanitize all inputs
LLM02	Insecure Output	Filter before use
LLM06	Info Disclosure	No secrets in prompts

9. Common Mistakes

# 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. Pre-Implementation Checklist

Phase 1: Before Writing Code

Write failing tests with mocked API responses
Define rate limits and cost thresholds
Set up secure credential loading (env vars or secrets manager)
Plan caching strategy for repeated queries

Phase 2: During Implementation

API keys loaded from environment/secrets manager only
Input sanitization active on all user content
Output filtering before using responses
Connection pooling configured
Retry logic with exponential backoff
Response caching for identical queries

Phase 3: Before Committing

All tests pass with >80% coverage
No API keys in git history (use git-secrets)
Security scan passes (bandit)
Type checking passes (mypy)
Daily spending limits configured
Multi-provider fallback tested

11. Summary

Your goal is to create cloud API integrations that are:

Test-Driven: All functionality verified with mocked tests
Performant: Connection pooling, caching, async operations
Secure: Protected against prompt injection and data exfiltration
Reliable: Multi-provider fallback with proper error handling
Cost-effective: Rate limiting and usage monitoring

For complete implementation details, see:

references/advanced-patterns.md - Caching, streaming, optimization
references/security-examples.md - Full vulnerability analysis
references/threat-model.md - Attack scenarios and mitigations

Repository: martinholovsky/claude-skills-generator
Commit: 1086ef2

Last updated: about 2 months ago
Created: about 2 months ago

Is this your skill?

If you maintain this skill, you can claim it as your own. Once claimed, you can manage eval scenarios, bundle related skills, attach documentation or rules, and ensure cross-agent compatibility.