Project 10: Deploy a Complete OpenAI SDK Application to Production

This project takes Project 1's chatbot — ChatSession, the retry wrapper, and the streaming loop — and carries it the rest of the way to a production-ready deployment, synthesizing deployment and operations patterns from Unit 26, the security practices from Unit 23, and the observability, cost, and performance techniques from Unit 25. Where earlier projects each introduced one capability, this lesson is entirely about the layer that sits around a working application before it is safe and observable enough to run in front of real users.

Scope and Design Decisions

The starting point is assumed to be Project 1's ChatSession and stream_reply function, already working locally. This lesson adds four things a working prototype does not have: a secured API boundary, request-level authentication, structured logging and cost tracking, and a deployable service definition. It does not cover infrastructure choice (which cloud provider, which container platform) in depth — that decision is context-specific and Unit 26 covers the trade-offs; the focus here is what needs to be true of the application itself regardless of where it runs.

Four decisions define this project:

1. Secrets never enter application code or logs. The API key and any other credentials are loaded from environment variables at startup and are never written to a log line, an error message, or a response body.
2. Every request is authenticated before it reaches the model. A publicly reachable chatbot endpoint with no request-level authentication is an open invitation to have your API budget consumed by someone else's traffic.
3. Cost and latency are logged per request, not inferred after the fact. Production incidents involving runaway spend or degraded latency are diagnosed far faster when the data was captured at request time rather than reconstructed later from provider billing dashboards.
4. The service fails safely and visibly. An unhandled error should return a clean error response and emit a log entry, never leak a stack trace or an API key to the client.

Wrapping the Chatbot as a Secured HTTP Service

import os
import time
import logging
import hashlib
import hmac
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("chatbot_service")

app = FastAPI()

API_KEY_HASH = os.environ["CHATBOT_API_KEY_HASH"]  # sha256 hex digest, never the raw key

def verify_request_key(provided_key: str) -> bool:
    computed_hash = hashlib.sha256(provided_key.encode()).hexdigest()
    # Constant-time comparison prevents a timing side-channel from leaking
    # how many leading characters of the hash matched.
    return hmac.compare_digest(computed_hash, API_KEY_HASH)

@app.middleware("http")
async def require_api_key(request: Request, call_next):
    provided = request.headers.get("x-api-key", "")
    if not verify_request_key(provided):
        logger.warning("Rejected request with invalid API key from %s", request.client.host)
        raise HTTPException(status_code=401, detail="Invalid or missing API key")
    return await call_next(request)

Storing API_KEY_HASH rather than the raw client-facing key means that even if the service's environment configuration were exposed, an attacker would have a hash, not a usable credential — the same principle as storing password hashes rather than passwords, applied here to service-to-service authentication. hmac.compare_digest is used instead of == specifically because Python's default string comparison short-circuits on the first mismatched character, which creates a timing side-channel: an attacker measuring response times could incrementally guess a hash byte by byte. This is a small detail with an outsized security consequence, and it is exactly the kind of thing Unit 23 emphasizes checking for in any comparison involving secrets.

Note: Real deployments typically use a proper API gateway, OAuth, or a managed authentication service rather than a hand-rolled header check; this middleware illustrates the underlying principle (verify before processing, compare secrets in constant time) in a form small enough to read in full, not a recommendation to hand-roll auth for a real production system.

Wiring in the Chatbot With Cost and Latency Logging

from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY from the environment automatically

# Reuse ChatSession, call_with_retry, and stream_reply from Project 1 unmodified.

PRICE_PER_1K_INPUT_TOKENS = 0.003
PRICE_PER_1K_OUTPUT_TOKENS = 0.012

def estimate_cost(usage) -> float:
    input_cost = (usage.input_tokens / 1000) * PRICE_PER_1K_INPUT_TOKENS
    output_cost = (usage.output_tokens / 1000) * PRICE_PER_1K_OUTPUT_TOKENS
    return round(input_cost + output_cost, 6)

@app.post("/chat")
async def chat_endpoint(request: Request):
    body = await request.json()
    user_text = body.get("message", "")
    session_id = body.get("session_id", "anonymous")

    session = _get_or_create_session(session_id)
    start_time = time.monotonic()

    try:
        response = client.responses.create(
            model=session.model,
            input=session.to_input_list() + [{"role": "user", "content": user_text}],
        )
    except Exception:
        logger.exception("Chat completion failed for session %s", session_id)
        raise HTTPException(status_code=502, detail="The assistant is temporarily unavailable.")

    latency_ms = round((time.monotonic() - start_time) * 1000, 1)
    cost = estimate_cost(response.usage)

    logger.info(
        "session=%s latency_ms=%s cost_usd=%s input_tokens=%s output_tokens=%s",
        session_id, latency_ms, cost, response.usage.input_tokens, response.usage.output_tokens,
    )

    session.add_user_message(user_text)
    session.add_assistant_message(response.output_text)
    return {"reply": response.output_text, "latency_ms": latency_ms, "cost_usd": cost}

_SESSIONS: dict[str, "ChatSession"] = {}

def _get_or_create_session(session_id: str):
    from project1_chatbot import ChatSession  # reused from Project 1
    if session_id not in _SESSIONS:
        _SESSIONS[session_id] = ChatSession(system_prompt="You are a helpful assistant.")
    return _SESSIONS[session_id]

estimate_cost computes a per-request dollar figure directly from response.usage, logged alongside latency on every single request rather than only sampled or reconstructed later from a monthly bill. This is the difference between noticing a cost spike within minutes of a bad deployment and noticing it at the end of the billing cycle — cheap to add, and one of the highest-leverage observability practices from Unit 25 for any LLM-backed service, since per-request cost varies far more than in typical web services and can spike suddenly from something as simple as a prompt change that grows the system prompt.

The try/except around the model call logs the full exception server-side with logger.exception (which includes the stack trace) but returns only a generic, safe message to the client — this is the fail-safely principle in code: detailed diagnostic information stays in the logs where an operator can see it, and the client never receives anything that could reveal internal implementation details or, worse, a fragment of a stack trace containing a partially-redacted secret.

Health Checks and Graceful Startup Validation

@app.on_event("startup")
async def validate_configuration():
    required_vars = ["OPENAI_API_KEY", "CHATBOT_API_KEY_HASH"]
    missing = [v for v in required_vars if not os.environ.get(v)]
    if missing:
        raise RuntimeError(f"Missing required environment variables: {missing}")
    logger.info("Configuration validated successfully at startup")

@app.get("/health")
async def health_check():
    return {"status": "ok"}

validate_configuration runs once at process startup and fails loudly — raising, which prevents the service from starting at all — rather than allowing the service to come up and fail confusingly on the first real request. This is a deliberate trade-off: a service that starts successfully but is silently misconfigured produces much harder-to-diagnose incidents than one that refuses to start with a clear error message naming exactly which variable is missing. /health is a minimal liveness endpoint, the kind a container orchestrator or load balancer polls to decide whether to route traffic to this instance; a more thorough version might also verify connectivity to the OpenAI API itself, at the cost of a small amount of latency on every health check.

Containerizing the Service

# Dockerfile content (not executable Python, shown for reference):
#
# FROM python:3.12-slim
# WORKDIR /app
# COPY requirements.txt .
# RUN pip install --no-cache-dir -r requirements.txt
# COPY . .
# ENV PYTHONUNBUFFERED=1
# EXPOSE 8000
# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

def load_secrets_from_environment() -> dict:
    """Confirms the deployment pattern: secrets arrive via environment
    variables injected by the orchestration platform's secret manager,
    never baked into the image or committed to source control."""
    return {
        "openai_api_key": os.environ["OPENAI_API_KEY"],
        "api_key_hash": os.environ["CHATBOT_API_KEY_HASH"],
    }

The Dockerfile shown as a comment (since it is not Python, it cannot go in an executable block, but it belongs alongside the code it packages) never copies a .env file or hardcodes a key — secrets are injected at container runtime by whatever secret manager the deployment platform provides (a cloud provider's secrets service, Kubernetes secrets, or a platform like the one Unit 26 walks through). PYTHONUNBUFFERED=1 ensures log output is flushed immediately rather than buffered, which matters because a container's stdout is typically what the orchestration platform's log collector reads — buffered output can make logs appear to "disappear" until the buffer flushes, which is exactly the kind of thing to catch before it causes a confusing debugging session in production.

Testing the Security and Cost Logic Without a Live Server

def test_verify_request_key_accepts_correct_key():
    global API_KEY_HASH
    real_key = "test-secret-key-123"
    API_KEY_HASH = hashlib.sha256(real_key.encode()).hexdigest()
    assert verify_request_key(real_key) is True
    print("PASS: correct API key is accepted")

def test_verify_request_key_rejects_wrong_key():
    global API_KEY_HASH
    API_KEY_HASH = hashlib.sha256("correct-key".encode()).hexdigest()
    assert verify_request_key("wrong-key") is False
    print("PASS: incorrect API key is rejected")

class FakeUsage:
    def __init__(self, input_tokens, output_tokens):
        self.input_tokens = input_tokens
        self.output_tokens = output_tokens

def test_estimate_cost_computes_expected_value():
    usage = FakeUsage(input_tokens=1000, output_tokens=500)
    cost = estimate_cost(usage)
    expected = round((1000 / 1000) * 0.003 + (500 / 1000) * 0.012, 6)
    assert cost == expected
    print("PASS: cost estimate matches expected per-token pricing calculation")

test_verify_request_key_accepts_correct_key()
test_verify_request_key_rejects_wrong_key()
test_estimate_cost_computes_expected_value()

test_verify_request_key_rejects_wrong_key and its accepting counterpart validate the authentication boundary using plain hashing with no HTTP server, no middleware stack, and no network call — the security-critical logic (does this credential match) is fully isolated and testable, which is exactly what should be true of any authentication check, since it is the piece of the system where an untested edge case has the highest consequence. estimate_cost is tested against a FakeUsage object with known token counts and a hand-computed expected value, which protects against a future refactor accidentally breaking the pricing formula unnoticed — a change that would otherwise only surface as an unexplained discrepancy in a monthly bill weeks later.

Extending This Project

Add per-session-ID rate limiting on top of the API key check to prevent a single compromised credential from consuming the entire request budget, and add distributed tracing (OpenTelemetry) around the model call so latency and cost can be correlated with specific upstream requests in a multi-service deployment.

Common Mistakes

• Comparing API keys or hashes with a standard ==. This is vulnerable to a timing attack that can leak information about the correct value byte by byte. Always use a constant-time comparison such as hmac.compare_digest for secret comparisons.
• Letting the service start successfully with missing configuration. A service that starts but fails mysteriously on the first request is much harder to diagnose than one that refuses to start with an explicit error naming the missing variable.
• Logging only aggregate cost from a monthly billing dashboard. By the time a cost anomaly shows up there, it has often been running for weeks. Log cost per request so a spike is visible within minutes of a bad deployment.

Best Practices

• Load all secrets from the environment, never from source code or committed files. Combine this with a startup check that fails fast and clearly if anything required is missing.
• Log latency and cost on every request, not as an afterthought. This is the single highest-leverage observability addition for an LLM-backed service, given how variable per-request cost can be.
• Return generic, safe error messages to clients while logging full detail server-side. A client should never see a stack trace, an internal exception message, or anything that could reveal implementation details or partial secrets.