16.7 — Project: AI API Gateway/Platform

This project builds the external, developer-facing AI API platform designed in Part 11.9 into a real, runnable system, incorporating Part 7.9's internal-gateway patterns hardened for external traffic. Read Part 11.9 first.

1. Customer Scenario

A company (Part 11.9's scenario) wants to expose internal AI capabilities (summarization, extraction, classification) as a metered, billed, public developer API — facing genuinely external, less-trusted, paying traffic rather than internal services.

2. Requirements

Recap of Part 11.9: strict external-facing security (every request untrusted, Part 9.5), accurate per-call billing metering, enforced per-key rate limiting as both a fairness and cost-abuse control, industry-standard API ergonomics.

3. Architecture

Part 11.9's architecture: a hardened external gateway (authentication, validation, rate limiting) in front of internal AI capability services, with an atomic serve-and-meter billing pipeline.

4. Technology Selection

• FastAPI with strict Pydantic validation (Part 1.3/1.9) on every external-facing endpoint.
• Redis (Part 1.6) for per-key rate limiting/quota, given its speed for this high-frequency check.
• A dedicated, transactional billing/metering database (Part 1.4) distinct from general operational logging, given billing's direct financial-trust stakes (Part 11.9, section 8).

5. Folder Structure

ai-api-platform/
├── src/
│   ├── gateway/
│   │   ├── auth.py                 # API-key authentication (Part 9.4/9.5)
│   │   ├── rate_limiter.py         # per-key, Redis-backed (Part 1.6/9.5)
│   │   └── validation.py           # strict external-input validation (Part 1.9/9.1)
│   ├── capabilities/
│   │   ├── summarization.py        # Part 3.1/3.2
│   │   ├── extraction.py           # Part 3.2
│   │   └── classification.py       # Part 3.1
│   ├── billing/
│   │   ├── metering.py             # atomic serve-and-meter logging (Part 11.9, section 6)
│   │   └── reconciliation.py       # served-vs-billed discrepancy monitoring (Part 8.4)
│   └── docs/
│       └── openapi_customization.py  # developer-facing API documentation (Part 1.3)
├── tests/
│   ├── unit/
│   ├── security/
│   │   └── test_external_input_hardening.py  # adversarial malformed/malicious payloads
│   └── billing/
│       └── test_metering_accuracy.py  # verifies every served call is correctly billed
├── infra/
│   └── k8s/
│       └── gateway-deployment.yml  # independently scaled from capability services (Part 7.4/11.9)
└── requirements.txt

6. Implementation (key excerpt)

The atomic serve-and-meter pattern, directly implementing Part 11.9's core billing-accuracy requirement:

async def handle_api_call(request, capability_service, metering_store) -> dict:
    """
    Serves an API call and records its billing metering as one atomic
    operation, preventing the discrepancy risk of two separately-failable steps.
    """
    async with metering_store.transaction() as txn:
        result = await capability_service.process(request)
        await txn.record_usage(
            api_key=request.api_key, endpoint=request.endpoint,
            tokens_used=result.usage.total_tokens, timestamp=datetime.now(timezone.utc),
        )
        # Commit only after BOTH the serve and the metering record succeed —
        # a failure in either rolls back the transaction, never leaving a
        # served-but-unbilled or billed-but-unserved discrepancy.
    return result

7. Testing

test_external_input_hardening.py runs a suite of deliberately malformed, oversized, and adversarial payloads (Part 1.9/9.1) against every endpoint, verifying strict validation rejects them before reaching any internal capability service. test_metering_accuracy.py verifies every successfully served call produces exactly one billing record, with no discrepancy under simulated partial-failure conditions.

Rendered in full below, test_metering_accuracy.py is this project's highest-priority test file — a billing discrepancy is a direct financial-trust failure (section 9), not a cosmetic bug — covering the atomic serve-and-meter transaction from section 6 under three conditions: the happy path, a mid-transaction capability-service failure, and concurrent requests against the same API key's quota (the classic race condition that lets a client burst past its limit if quota checks aren't atomic):

"""tests/billing/test_metering_accuracy.py

Verifies the atomic serve-and-meter pipeline (section 6) never produces a
served-but-unbilled or billed-but-unserved discrepancy, and that per-key
rate limiting holds even under concurrent bursts.
"""

import asyncio

import pytest
import pytest_asyncio

from src.gateway.rate_limiter import RedisRateLimiter
from src.billing.metering import handle_api_call


API_KEY = "sk-test-acme-001"


class _FakeCapabilityService:
    """A stand-in capability service whose `process` call can be told to
    fail, to test the transaction's rollback behavior under partial failure."""

    def __init__(self, should_fail: bool = False, tokens_used: int = 120):
        self.should_fail = should_fail
        self.tokens_used = tokens_used
        self.calls = 0

    async def process(self, request):
        self.calls += 1
        if self.should_fail:
            raise RuntimeError("simulated downstream failure")

        class _Result:
            usage = type("Usage", (), {"total_tokens": self.tokens_used})()

        return _Result()


@pytest_asyncio.fixture
async def metering_store(pg_pool):
    """Provides a transactional metering store backed by a real test database,
    so transaction rollback behavior is exercised against real commit semantics
    rather than a mock that can't reproduce partial-failure edge cases.

    Args:
        pg_pool: A connection pool fixture for the test billing database.

    Returns:
        MeteringStore: A metering store wired to the test database.
    """
    from src.billing.metering import MeteringStore

    store = MeteringStore(pg_pool)
    await store.reset_for_test(API_KEY)
    return store


@pytest.mark.asyncio
async def test_successful_call_produces_exactly_one_billing_record(metering_store):
    """The happy path: a successfully served call must produce exactly one
    usage record, with the correct token count."""
    request = type("Req", (), {"api_key": API_KEY, "endpoint": "/v1/summarize"})()
    service = _FakeCapabilityService(should_fail=False, tokens_used=340)

    await handle_api_call(request, service, metering_store)

    records = await metering_store.get_usage_records(API_KEY)
    assert len(records) == 1
    assert records[0]["tokens_used"] == 340


@pytest.mark.asyncio
async def test_downstream_failure_leaves_no_billing_record(metering_store):
    """If the capability service fails mid-call, the transaction must roll
    back completely — no billing record for a call that was never actually
    served, which would otherwise be a direct customer-facing billing dispute."""
    request = type("Req", (), {"api_key": API_KEY, "endpoint": "/v1/summarize"})()
    service = _FakeCapabilityService(should_fail=True)

    with pytest.raises(RuntimeError):
        await handle_api_call(request, service, metering_store)

    records = await metering_store.get_usage_records(API_KEY)
    assert records == []  # served-but-unbilled AND billed-but-unserved are both excluded


@pytest.mark.asyncio
async def test_rate_limiter_blocks_concurrent_burst_past_quota(redis_client):
    """Verifies the Redis-backed limiter (section 4) enforces quota atomically
    under concurrent requests — a non-atomic check-then-increment would let
    a burst of simultaneous requests each pass the check before any of them
    finish incrementing, letting a client exceed its limit."""
    limiter = RedisRateLimiter(redis_client, limit=10, window_seconds=60)

    results = await asyncio.gather(
        *[limiter.allow_request(API_KEY) for _ in range(25)]
    )

    allowed_count = sum(1 for allowed in results if allowed)
    assert allowed_count == 10  # exactly the configured limit, never more, despite 25 concurrent attempts


@pytest.mark.asyncio
async def test_quota_resets_after_window_expires(redis_client):
    """A key that has exhausted its quota must be allowed again once the
    rate-limit window rolls over — verifying the limiter doesn't
    permanently lock out a key past a single window."""
    limiter = RedisRateLimiter(redis_client, limit=1, window_seconds=1)

    assert await limiter.allow_request(API_KEY) is True
    assert await limiter.allow_request(API_KEY) is False  # quota exhausted within the window

    await asyncio.sleep(1.1)
    assert await limiter.allow_request(API_KEY) is True  # window expired, quota restored

The concurrent-burst test (test_rate_limiter_blocks_concurrent_burst_past_quota) is the one most worth taking seriously in review: a rate limiter that passes every sequential test but uses a non-atomic Redis GET-then-SET internally will still fail it, which is exactly the class of bug section 12's "Redis-backed, low-latency check" choice needs to guard against in practice, not just in intent.

8. Evaluation

Per-capability accuracy/quality evaluation (Part 8.1) as a product-quality signal for developers deciding whether to adopt the API, alongside latency/uptime SLA tracking (Part 11.9, section 4) as the published reliability commitment.

9. Security

This is the highest-external-exposure project in this handbook — strict input validation on every field (Part 1.9/9.1), enforced rate limiting as a security control against cost-abuse (Part 9.5), and API-key rotation/revocation capability (Part 9.4's token-management discipline, applied to API keys) treated as baseline, non-optional requirements.

10. Observability

Per-API-key usage/latency/error-rate metrics (Part 8.4) serve both operational monitoring and developer-facing usage dashboards; billing-reconciliation monitoring (section 7) as a dedicated, high-priority alert category given its direct financial-trust implications.

11. Deployment

Gateway and internal capability services scale independently (Part 7.4/7.5/11.9) — a traffic spike from one large legitimate customer shouldn't be conflated with abuse, requiring quota logic that distinguishes tier-appropriate high usage from actual anomalous behavior.

12. Scaling

Rate limiting and quota enforcement must scale with the gateway itself (Part 7.5) without becoming a bottleneck — a Redis-backed, low-latency check (Part 1.6) is chosen specifically for this reason over a slower, more complex alternative.

13. Cost Considerations

The underlying per-call LLM/compute cost (Part 7.10) must be accurately modeled and priced with margin into external tiers — a pricing structure built without this discipline risks being unprofitable at real usage volume (Part 11.9, section 12's exact warning).

14. Failure Scenarios

A malicious/adversarial external payload → caught by gateway-level validation, never reaching internal services (Part 9.1/9.5). A billing/metering discrepancy → caught by reconciliation monitoring (section 7/10) before it compounds into a customer-facing billing dispute. A rate-limit bypass attempt → caught by enforced, Redis-backed limiting plus anomaly monitoring for unusual key-usage patterns suggesting compromise.

15. Improvements

Add a sandboxed "playground" tier with tight, free-tier rate limits letting prospective developers try the API before committing to a paid plan — a common, effective adoption-driving pattern for developer-facing API products, implemented using the same rate-limiting infrastructure already built.

16. Business Metrics

API adoption growth (new keys activated, calls per active key over time) as a leading indicator; revenue per API key and gross margin per capability (accounting for actual underlying LLM cost, Part 15) as the core lagging business metrics justifying continued product investment.

---

Key takeaways

• This project warrants the strictest security posture in this handbook, given genuinely external, less-trusted, paying traffic — every input is untrusted, and rate limiting functions as a hard security control, not just fairness.
• Atomic serve-and-meter logging is the specific technical mechanism that prevents billing discrepancies, a direct financial-trust requirement distinct from general operational logging.
• Pricing this product correctly requires accurately modeling the actual underlying per-call cost (Part 7.10) — a business-model risk as real as any technical one.

Recommended next chapter

08-full-fde-engagement-first-meeting-to-production.md