Developing an API platform to provide AI services to third parties
An API platform for AI services is more than just a set of endpoints, but a complete developer experience: documentation, SDK, sandbox environment, developer portal, monitoring, and support. It is the developer experience that determines adoption rates.
Developer Portal
Key components of the portal:
- Interactive API docs (Swagger UI / Redoc): testing directly in the browser
- API Key management: creation, rotation, revocation of keys
- Usage dashboard: tokens, requests, expenses by period
- Sandbox: testing environment with mock responses and real models
- Webhooks management: subscription to events (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 generation
# Автогенерация 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 Architecture
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 system
@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
)
A well-designed API platform reduces integration time for partners from weeks to days and reduces support calls by 60-70% thanks to high-quality documentation and SDKs.