Разработка API-платформы для предоставления AI-сервисов третьим сторонам

Разработка API-платформы для предоставления AI-сервисов третьим сторонам

API-платформа для AI-сервисов — это не просто набор endpoints, а полноценная developer experience: документация, SDK, sandbox среда, developer portal, monitoring и поддержка. Именно developer experience определяет adoption скорость.

Developer Portal

Ключевые компоненты портала:

• Interactive API docs (Swagger UI / Redoc): тестирование прямо в браузере
• API Key management: создание, ротация, отзыв ключей
• Usage dashboard: токены, запросы, расходы по периодам
• Sandbox: тестовая среда с mock responses и реальными моделями
• Webhooks management: подписка на события (job completed, billing alert)

# FastAPI с автогенерацией OpenAPI документации
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI(
    title="AI Services API",
    version="2.0.0",
    description="Comprehensive AI inference and processing API",
    terms_of_service="https://api.company.com/terms",
    contact={"email": "[email protected]"},
    license_info={"name": "Commercial"},
)

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        description=app.description,
        routes=app.routes,
    )
    # Добавление примеров запросов
    openapi_schema["paths"]["/v1/completions"]["post"]["requestBody"]["content"][
        "application/json"]["examples"] = {
        "simple": {
            "summary": "Simple text completion",
            "value": {"model": "gpt-4o-mini", "prompt": "Hello, world!"}
        }
    }
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Multi-language SDK генерация

# Автогенерация SDK из OpenAPI spec через openapi-generator
# Поддерживает: Python, JavaScript/TypeScript, Go, Java, C#, Ruby

# Сгенерированный Python SDK:
from ai_platform import AIClient

client = AIClient(api_key="sk-...")

# Text generation
response = client.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    max_tokens=500
)

# Async поддержка
async with client.AsyncAIClient(api_key="sk-...") as async_client:
    response = await async_client.completions.create(...)

# Автоматические retries, exponential backoff
client = AIClient(
    api_key="sk-...",
    max_retries=3,
    timeout=30.0
)

Rate Limiting архитектура

from fastapi import Request
import redis.asyncio as aioredis

class TokenBucketRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def check(self, api_key: str, limit: int,
                    window_seconds: int = 60) -> tuple[bool, dict]:
        now = time.time()
        key = f"ratelimit:{api_key}:{int(now // window_seconds)}"

        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, window_seconds * 2)
        current_count, _ = await pipe.execute()

        remaining = max(0, limit - current_count)
        reset_at = int(now // window_seconds + 1) * window_seconds

        return current_count <= limit, {
            "X-RateLimit-Limit": limit,
            "X-RateLimit-Remaining": remaining,
            "X-RateLimit-Reset": reset_at
        }

Webhooks система

@app.post("/v1/webhooks")
async def register_webhook(request: WebhookCreateRequest,
                            api_key = Depends(authenticate)):
    webhook = await webhook_store.create({
        'customer_id': api_key.customer_id,
        'url': request.url,
        'events': request.events,  # ['job.completed', 'billing.limit_approaching']
        'secret': secrets.token_hex(32)  # Для проверки подписи
    })
    return {"webhook_id": webhook.id, "secret": webhook.secret}

async def deliver_webhook(webhook_id: str, event: str, payload: dict):
    webhook = await webhook_store.get(webhook_id)
    signature = hmac.new(
        webhook.secret.encode(),
        json.dumps(payload).encode(),
        hashlib.sha256
    ).hexdigest()

    async with httpx.AsyncClient() as client:
        response = await client.post(
            webhook.url,
            json=payload,
            headers={
                "X-Webhook-Signature": f"sha256={signature}",
                "X-Webhook-Event": event
            },
            timeout=30.0
        )

Хорошо разработанная API-платформа сокращает время интеграции для партнёров с недель до дней и снижает количество обращений в поддержку на 60-70% благодаря качественной документации и SDK.