Why this pattern
– WordPress is great at routing and rendering, not long-running I/O.
– AI calls are slow, variable, and expensive; they need retries, quotas, and tracing.
– The solution: push jobs to an external worker and accept results via signed webhooks.

Architecture (high level)
– Client (WP admin or theme) submits an AI request to a WP REST route.
– WordPress writes a job row (pending), enqueues to an external queue (or HTTP to a worker gateway).
– Worker (Python/Node) pulls the job, calls the AI provider, then POSTs a signed webhook back to WordPress.
– WordPress verifies the signature, stores result, and invalidates relevant cache.
– Frontend polls or uses SSE/WS via a lightweight proxy for updates.

Database schema (custom table)
– wp_ai_jobs
– id (bigint PK)
– user_id (bigint)
– status (enum: pending, running, succeeded, failed)
– input_hash (char(64)) for idempotency
– request_json (longtext)
– result_json (longtext, nullable)
– error_text (text, nullable)
– created_at, updated_at (datetime)
– idempotency_key (varchar(64), unique)
– webhook_ts (datetime, nullable)

Create the table on plugin activation
– dbDelta with utf8mb4, proper indexes:
– INDEX status_created (status, created_at)
– UNIQUE idempotency_key (idempotency_key)
– INDEX input_hash (input_hash)

Plugin structure (minimal)
– ai-integration/
– ai-integration.php (bootstrap, routes, activation)
– includes/
– class-ai-controller.php (REST endpoints)
– class-ai-webhook.php (webhook verifier)
– class-ai-repo.php (DB access)
– class-ai-queue.php (enqueue out to worker)
– helpers.php (crypto, validation)
– Do not store secrets in options; put them in wp-config.php.

Secrets and config (wp-config.php)
– define(‘AI_WORKER_URL’, ‘https://worker.example.com/jobs’);
– define(‘AI_WEBHOOK_SECRET’, ‘base64-32-bytes’);
– define(‘AI_JWT_PRIVATE_KEY’, ‘—–BEGIN PRIVATE KEY—–…’);
– define(‘AI_QUEUE_TIMEOUT’, 2); // seconds for outbound enqueue

REST endpoint: create job (POST /wp-json/ai/v1/jobs)
– Validate capability (logged-in or signed public token).
– Build idempotency_key from client or hash(input_json + user_id + model).
– Insert row (pending).
– Enqueue to worker:
– POST to AI_WORKER_URL with signed JWT (kid, iat, exp, sub=user_id, jti=idempotency_key).
– Timeout <= 2s. If enqueue fails, leave job pending; a retry worker (Action Scheduler) can re-enqueue.
– Return { job_id, status: "pending" }.

Example: tiny enqueue
– Headers: Authorization: Bearer
– Body: { job_id, idempotency_key, request: {…}, callback_url: “https://site.com/wp-json/ai/v1/webhook” }

Webhook endpoint: receive result (POST /wp-json/ai/v1/webhook)
– Require HMAC-SHA256 signature header: X-AI-Signature: base64(hmac(secret, body))
– Require idempotency_key and job_id in body.
– Verify:
– Constant-time compare HMAC.
– Check timestamp drift <= 2 minutes (X-AI-Timestamp).
– Enforce replay guard: cache "webhook:{jti}" in Redis for 10m.
– Update row (status to succeeded/failed, set result_json or error_text, webhook_ts).
– Return 204.

Minimal verification (PHP)
– $sig = base64_decode($_SERVER['HTTP_X_AI_SIGNATURE'] ?? '');
– $calc = hash_hmac('sha256', $rawBody, AI_WEBHOOK_SECRET, true);
– hash_equals($sig, $calc) or wp_die('invalid sig', 403);

Frontend polling pattern
– Client gets job_id, then polls GET /wp-json/ai/v1/jobs/{id} every 1–2s (cap at 30s).
– Cache-control: private, max-age=0. Use ETag from updated_at to 304 unchanged.
– Optional: stream via SSE proxied through PHP only if your infra supports long-lived requests without PHP-FPM worker starvation.

Idempotency and dedupe
– On create:
– If idempotency_key exists, return existing job.
– Also check input_hash + user_id within time window to reduce duplicates from flaky clients.

Rate limiting
– Per-user sliding window: e.g., 60 jobs/10m.
– Use wp_cache (Redis/Memcached). Key: rl:{user}:{minute-epoch}. Increment and check.
– On limit exceed, 429 with Retry-After.

Background retries
– Action Scheduler job scans pending/running older than N minutes:
– Re-enqueue if no worker ack.
– Mark failed if exceeded retry budget; store error_text.

Security checklist
– Do not accept webhooks without HMAC and timestamp.
– JWT to worker uses short exp (<=60s). Sign with ES256 or RS256; rotate keys quarterly.
– Sanitize and escape all fields when rendering.
– Disable file edits in prod; restrict wp-admin to known IPs if possible.
– Log minimal PII; encrypt sensitive request_json fields at rest if needed (sodium_crypto_secretbox).

Performance considerations
– Never call AI providers inside a WP page render path.
– Outbound enqueue must be non-blocking (<2s). Use Requests::post with short timeouts and no redirects.
– Store only necessary parts of result_json; large blobs to object storage (S3) with signed URLs.
– Use indexes to keep dashboard queries fast; paginate admin list by created_at DESC.
– Cache job summaries with wp_cache_set on read path; invalidate on webhook.

Worker reference (Python, outline)
– Pull from queue, call provider with circuit breaker and retry/backoff (e.g., 100ms→2s jitter).
– On completion, POST result to callback_url with:
– Headers: X-AI-Signature, X-AI-Timestamp
– Body: { job_id, idempotency_key, status, result_json, usage: {tokens, ms} }
– Keep results small; upload big artifacts elsewhere first.

Minimal job table index DDL
– INDEX status_created (status, created_at)
– INDEX user_created (user_id, created_at)
– UNIQUE idempotency_key (idempotency_key)

Observability
– Add a request_id to all flows; return it to client.
– Store provider latency, tokens, and error codes in result_json. Useful for cost/perf dashboards.
– Emit Server-Timing headers on job reads: worker;dur=123,provider;dur=456.

Admin UI ideas
– List jobs with filters (status, user, model).
– Re-enqueue button (capability checked).
– Export CSV of usage by date/user.

Deployment checklist
– HTTPS everywhere; verify real client IP behind any CDN.
– Set AI_WEBHOOK_SECRET via environment, not version control.
– Protect webhook with allowlist of worker IPs if static.
– Enable object cache. Prefer Redis with persistence.
– Load test: 200 req/s create → ensure PHP-FPM pool and DB connections stay healthy.
– Back up the table and rotate old rows to cold storage monthly.

What to avoid
– Synchronous AI calls in templates.
– Storing provider keys in options.
– Webhooks without signature or timestamp.
– Unbounded job payload sizes.

This pattern scales from small sites to high-traffic publishers, keeps your PHP requests fast, and centralizes reliability and security where they belong: in the worker and webhook boundary.

This post shows a real, production-ready pattern for running AI inference from WordPress without exposing API keys to the browser. We’ll build a secure REST endpoint, add caching and rate limits, handle retries and webhooks, and log everything for observability.

Use case examples:
– Generate product descriptions or summaries from authenticated admin screens
– Enrich form submissions (classify, route, extract fields)
– Power a custom block or dashboard tool that fetches AI results server-side

High-level architecture
– Frontend (block/admin page) → WP REST endpoint (server) → AI provider (OpenAI/Anthropic/etc.)
– Caching at the WordPress layer (transient or object cache)
– Rate limiting per user/site to prevent abuse
– Background jobs for long-running tasks via Action Scheduler
– Optional webhooks from AI provider back to WordPress
– Audit logs stored in a custom table with PII minimization

Prerequisites
– WordPress 6.4+
– PHP 8.1+
– Persistent object cache (Redis/Memcached) recommended
– Action Scheduler plugin or equivalent job runner
– Environment configuration for secrets (wp-config.php or environment variables)

1) Minimal plugin scaffold
Create wp-content/plugins/ai-secure-proxy/ai-secure-proxy.php

prefix . ‘aisp_logs’;
$charset = $wpdb->get_charset_collate();
$sql = “CREATE TABLE IF NOT EXISTS $table (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
user_id BIGINT UNSIGNED NULL,
route VARCHAR(191) NOT NULL,
req_hash CHAR(64) NOT NULL,
tokens_in INT UNSIGNED DEFAULT 0,
tokens_out INT UNSIGNED DEFAULT 0,
duration_ms INT UNSIGNED DEFAULT 0,
status_code INT DEFAULT 0,
error TEXT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
) $charset;”;
require_once ABSPATH . ‘wp-admin/includes/upgrade.php’;
dbDelta($sql);
}

public function register_routes() {
register_rest_route(self::NAMESPACE, ‘/infer’, [
‘methods’ => WP_REST_Server::CREATABLE,
‘permission_callback’ => [$this, ‘can_access’],
‘callback’ => [$this, ‘handle_infer’],
‘args’ => [
‘prompt’ => [‘required’ => true, ‘type’ => ‘string’, ‘minLength’ => 1, ‘maxLength’ => 8000],
‘model’ => [‘required’ => false, ‘type’ => ‘string’, ‘default’ => ‘gpt-4o-mini’],
‘cache’ => [‘required’ => false, ‘type’ => ‘boolean’, ‘default’ => true]
]
]);

register_rest_route(self::NAMESPACE, ‘/webhook’, [
‘methods’ => WP_REST_Server::CREATABLE,
‘permission_callback’ => ‘__return_true’,
‘callback’ => [$this, ‘handle_webhook’],
]);
}

public function can_access(WP_REST_Request $req) {
// Require logged-in user with capability. For public use, implement signed HMAC header instead.
return current_user_can(self::CAPABILITY);
}

private function rate_key($user_id) {
return self::OPT_PREFIX . ‘rate_’ . $user_id . ‘_’ . gmdate(‘YmdH’);
}

private function is_rate_limited($user_id) {
$key = $this->rate_key($user_id);
$count = (int) wp_cache_get($key, ”, false, $found);
if (!$found) $count = (int) get_option($key, 0);
return $count >= self::RATE_LIMIT;
}

private function bump_rate($user_id) {
$key = $this->rate_key($user_id);
$count = (int) get_option($key, 0) + 1;
update_option($key, $count, false);
wp_cache_set($key, $count, ”, 3600);
}

public function handle_infer(WP_REST_Request $req) {
$user_id = get_current_user_id();
if ($this->is_rate_limited($user_id)) {
return new WP_Error(‘rate_limited’, ‘Rate limit exceeded’, [‘status’ => 429]);
}

$prompt = trim($req->get_param(‘prompt’));
$model = sanitize_text_field($req->get_param(‘model’));
$use_cache = (bool) $req->get_param(‘cache’);

// Hash to cache against normalized inputs
$req_hash = hash(‘sha256’, json_encode([‘m’=>$model, ‘p’=>$prompt]));

$cache_key = self::OPT_PREFIX . ‘c_’ . $req_hash;
if ($use_cache) {
$cached = wp_cache_get($cache_key);
if ($cached !== false) {
$this->log($user_id, ‘/infer’, $req_hash, 0, 0, 1, 200, null);
return new WP_REST_Response([‘cached’ => true, ‘result’ => $cached], 200);
}
}

$start = microtime(true);
$resp = $this->call_ai_provider($model, $prompt);
$duration_ms = (int) round((microtime(true) – $start) * 1000);

if (is_wp_error($resp)) {
$this->log($user_id, ‘/infer’, $req_hash, 0, 0, $duration_ms, 500, $resp->get_error_message());
return $resp;
}

$result = [
‘text’ => $resp[‘text’] ?? ”,
‘tokens_in’ => $resp[‘tokens_in’] ?? 0,
‘tokens_out’ => $resp[‘tokens_out’] ?? 0,
‘model’ => $model,
];

if ($use_cache && !empty($result[‘text’])) {
wp_cache_set($cache_key, $result, ”, self::CACHE_TTL);
}

$this->bump_rate($user_id);
$this->log($user_id, ‘/infer’, $req_hash, (int)$result[‘tokens_in’], (int)$result[‘tokens_out’], $duration_ms, 200, null);

return new WP_REST_Response([‘cached’ => false, ‘result’ => $result], 200);
}

private function call_ai_provider($model, $prompt) {
// Secrets via environment or wp-config
$api_key = getenv(‘OPENAI_API_KEY’) ?: (defined(‘OPENAI_API_KEY’) ? OPENAI_API_KEY : ”);
if (!$api_key) return new WP_Error(‘config_error’, ‘Missing AI API key’, [‘status’ => 500]);

$body = [
‘model’ => $model,
‘input’ => $prompt,
];

$attempts = 0;
$max_attempts = 3;
$last_err = null;

while ($attempts 15,
‘headers’ => [
‘Authorization’ => ‘Bearer ‘ . $api_key,
‘Content-Type’ => ‘application/json’
],
‘body’ => wp_json_encode($body),
]);

if (is_wp_error($response)) {
$last_err = $response;
} else {
$code = wp_remote_retrieve_response_code($response);
$data = json_decode(wp_remote_retrieve_body($response), true);

// Simple provider-agnostic parse
if ($code >= 200 && $code $text, ‘tokens_in’ => $tokens_in, ‘tokens_out’ => $tokens_out];
}

// Retry on 429/5xx with exponential backoff
if (in_array($code, [429, 500, 502, 503, 504], true)) {
$last_err = new WP_Error(‘ai_retry’, ‘Transient AI error: ‘ . $code);
} else {
return new WP_Error(‘ai_error’, ‘AI provider error: ‘ . $code, [‘status’ => $code, ‘data’ => $data]);
}
}

// Backoff
usleep((int) (pow(2, $attempts) * 200000)); // 200ms, 400ms, 800ms
}

return $last_err ?: new WP_Error(‘ai_error’, ‘Unknown AI error’, [‘status’ => 500]);
}

public function handle_webhook(WP_REST_Request $req) {
// Optional: verify HMAC signature from provider
$shared = getenv(‘AI_WEBHOOK_SECRET’) ?: ”;
$sig = $req->get_header(‘x-aisig’);
if ($shared && $sig) {
$calc = hash_hmac(‘sha256’, $req->get_body(), $shared);
if (!hash_equals($calc, $sig)) {
return new WP_Error(‘forbidden’, ‘Invalid signature’, [‘status’ => 403]);
}
}
// Process event (store job result, update post meta, etc.)
do_action(‘aisp_webhook_received’, $req->get_json_params());
return new WP_REST_Response([‘ok’ => true], 200);
}

private function log($user_id, $route, $req_hash, $tin, $tout, $dur_ms, $code, $error) {
global $wpdb;
$table = $wpdb->prefix . ‘aisp_logs’;
$wpdb->insert($table, [
‘user_id’ => $user_id ?: null,
‘route’ => $route,
‘req_hash’ => $req_hash,
‘tokens_in’ => $tin,
‘tokens_out’ => $tout,
‘duration_ms’ => $dur_ms,
‘status_code’ => $code,
‘error’ => $error,
], [‘%d’,’%s’,’%s’,’%d’,’%d’,’%d’,’%d’,’%s’]);
}
}
new AI_Secure_Proxy();

2) Secure frontend usage
– Admin page or block should never expose provider keys.
– Call the endpoint with wp.apiFetch or fetch, including nonce.
Example:

const res = await wp.apiFetch({
path: ‘/ai-secure-proxy/v1/infer’,
method: ‘POST’,
data: { prompt, model: ‘gpt-4o-mini’, cache: true }
});

3) Authentication options
– Logged-in only (capability-based): safest for back-office tools.
– Public usage: require HMAC-signed requests.
– Client sends X-Sign: HMAC_SHA256(body, PUBLIC_CLIENT_SECRET).
– Server validates before processing.
– Consider IP allowlists for server-to-server.

4) Caching strategy
– Key by normalized inputs and model.
– Use persistent object cache for speed and TTL-based eviction.
– Invalidate on model/version changes or when content sources update.

5) Rate limiting
– Example above uses per-user/hour counters.
– For high-traffic public endpoints, use Redis INCR with TTL or a token bucket.
– Return 429 with Retry-After.

6) Background jobs
– For long prompts or batch operations, offload to Action Scheduler:
– POST creates a job and returns job_id.
– Worker picks up job, calls AI, stores result in post meta or custom table.
– Optional webhook updates job status when provider supports async.

7) Webhooks
– Verify HMAC signature.
– Idempotency: store last seen event IDs to avoid duplicate processing.
– Enqueue work; do not block the webhook handler.

8) Observability
– Log request hash, tokens, duration, status code.
– Build admin UI with filters by user/date/status.
– Emit wp-json logs or use error_log for quick triage in staging.
– Add metrics: P95 latency, cache hit ratio, 4xx/5xx rates.

9) Security checklist
– Store API keys in environment or wp-config, never the DB.
– Enforce capability or signed requests.
– Validate and length-limit inputs.
– Strip PII where possible before sending to provider.
– Set conservative timeouts; implement retries with backoff.
– Disable indexing for admin tools; use HTTPS everywhere.

10) Performance notes
– Keep endpoint thin; avoid loading unnecessary WP subsystems.
– Enable OPcache and a persistent object cache.
– Batch multiple small prompts into one call when possible.
– Circuit breaker: if upstream is failing consistently, short-circuit for a cooldown period.

11) Deployment tips
– Separate staging keys and webhooks.
– Run a load test (k6/Locust) with realistic prompts.
– Monitor Redis hit ratio and PHP-FPM slow logs.
– Backup logs table and rotate old entries.

Extending the pattern
– Add SSE endpoint for streaming tokens to an admin tool.
– Implement content moderation via a pre-check model before saving results.
– Build a “dry run” mode for previews without committing changes.

This approach keeps AI logic server-side, with guardrails for cost, performance, and security—ready for production in WordPress environments.

This post shows how to implement a production-ready AI chat in WordPress that streams responses with low latency and strong security. The setup:

– WordPress plugin renders the chat UI and signs requests.
– Django backend handles model calls, streams with SSE, caches with Redis, and enforces auth/rate limits.
– No API keys in the browser or WordPress DB.

Architecture
– WordPress (frontend): Shortcode [ai_chat], enqueue JS, get WP nonce, fetch a signed JWT from WP to call the backend.
– Django API: /v1/chat/stream (SSE), validates JWT, checks rate limits, optional moderation, calls OpenAI (or other LLM), streams tokens.
– Redis: Caching prompt fingerprints and user quotas, storing partial transcripts if needed.
– Nginx/Proxy: Keep-Alive, proxy_buffering off for SSE, generous timeouts.

WordPress plugin (minimal, secure)
– Store no model keys in WordPress.
– Use a secret in wp-config.php for signing short-lived JWTs.
– Nonce to mitigate CSRF.

wp-config.php
define(‘AIGUY_SSO_JWT_SECRET’, ‘rotate-this-64-bytes’);
define(‘AIGUY_BACKEND_BASE’, ‘https://api.example.com’);

Plugin file: wp-content/plugins/aiguy-chat/aiguy-chat.php
<?php
/**
* Plugin Name: AI Guy Chat
* Description: Streaming AI chat via secure backend.
*/
if (!defined('ABSPATH')) exit;

add_action('init', function () { add_shortcode('ai_chat', 'aiguy_chat_shortcode'); });

function aiguy_chat_shortcode() {
if (!is_user_logged_in()) return '

Please log in to use chat.

‘;
$nonce = wp_create_nonce(‘aiguy_chat’);
$payload = [
‘sub’ => get_current_user_id(),
‘roles’ => wp_get_current_user()->roles,
‘iat’ => time(),
‘exp’ => time() + 300,
‘site’ => get_site_url(),
];
$jwt = aiguy_jwt_encode($payload, AIGUY_SSO_JWT_SECRET);
wp_enqueue_script(‘aiguy-chat-js’, plugin_dir_url(__FILE__).’chat.js’, [], ‘1.0’, true);
wp_localize_script(‘aiguy-chat-js’, ‘AIGUYCFG’, [
‘nonce’ => $nonce,
‘jwt’ => $jwt,
‘api’ => AIGUY_BACKEND_BASE.’/v1/chat/stream’,
]);
ob_start(); ?>

‘HS256′,’typ’=>’JWT’])), ‘+/’, ‘-_’), ‘=’);
$p = rtrim(strtr(base64_encode(json_encode($payload)), ‘+/’, ‘-_’), ‘=’);
$s = rtrim(strtr(base64_encode(hash_hmac(‘sha256’, “$h.$p”, $secret, true)), ‘+/’, ‘-_’), ‘=’);
return “$h.$p.$s”;
}

JS: wp-content/plugins/aiguy-chat/chat.js
(function(){
const log = document.getElementById(‘aiguy-log’);
const form = document.getElementById(‘aiguy-form’);
const input = document.getElementById(‘aiguy-input’);

function append(role, text) {
const el = document.createElement(‘div’);
el.className = role;
el.textContent = text;
log.appendChild(el);
log.scrollTop = log.scrollHeight;
}

form.addEventListener(‘submit’, async (e) => {
e.preventDefault();
const q = input.value.trim();
if (!q) return;
append(‘user’, q);
input.value = ”;

const url = AIGUYCFG.api;
const params = { message: q };
const ctrl = new AbortController();
const res = await fetch(url, {
method: ‘POST’,
headers: {
‘Content-Type’: ‘application/json’,
‘Authorization’: ‘Bearer ‘+AIGUYCFG.jwt,
‘X-WP-Nonce’: AIGUYCFG.nonce
},
body: JSON.stringify(params),
signal: ctrl.signal
});
if (!res.ok) { append(‘error’, ‘Request failed.’); return; }

const reader = res.body.getReader();
const dec = new TextDecoder();
let acc = ”;
append(‘assistant’, ”);
const last = log.lastChild;

while (true) {
const {value, done} = await reader.read();
if (done) break;
acc += dec.decode(value, {stream: true});
for (const line of acc.split(‘n’)) {
if (!line.startsWith(‘data:’)) continue;
const payload = line.slice(5).trim();
if (payload === ‘[DONE]’) break;
try {
const chunk = JSON.parse(payload);
last.textContent += chunk.delta || ”;
} catch(_) {}
}
}
});
})();

Django backend (SSE with OpenAI, Redis, and limits)
– Use Django or Django + Django Ninja/DRF. Example below uses plain Django view.
– Keep OpenAI key only on the server.
– Validate JWT from WordPress using the shared secret.

settings.py (env-driven)
AIGUY_JWT_SECRET = os.getenv(‘AIGUY_JWT_SECRET’)
OPENAI_API_KEY = os.getenv(‘OPENAI_API_KEY’)
REDIS_URL = os.getenv(‘REDIS_URL’, ‘redis://127.0.0.1:6379/0’)
ALLOWED_ORIGINS = [‘https://www.example.com’]

urls.py
from django.urls import path
from .views import chat_stream
urlpatterns = [ path(‘v1/chat/stream’, chat_stream), ]

views.py
import json, time, hashlib, hmac, base64, os
from django.http import StreamingHttpResponse, HttpResponseForbidden, JsonResponse
from django.views.decorators.csrf import csrf_exempt
import redis
import openai

r = redis.from_url(os.getenv(‘REDIS_URL’,’redis://localhost:6379/0′))
openai.api_key = os.getenv(‘OPENAI_API_KEY’)
JWT_SECRET = os.getenv(‘AIGUY_JWT_SECRET’,”)

def verify_jwt(token):
try:
h,p,s = token.split(‘.’)
sig = base64.urlsafe_b64encode(hmac.new(JWT_SECRET.encode(), f'{h}.{p}’.encode(), ‘sha256′).digest()).rstrip(b’=’).decode()
if sig != s: return None
payload = json.loads(base64.urlsafe_b64decode(p + ‘==’))
if payload.get(‘exp’,0) 60: return JsonResponse({‘error’:’rate_limited’}, status=429)

# Prompt fingerprint cache key
fp = hashlib.sha256((message.strip()).encode()).hexdigest()
cache_key = f’chat:fp:{fp}’

messages = [
{“role”:”system”,”content”:”You are a helpful assistant for the site.”},
{“role”:”user”,”content”:message}
]

resp = StreamingHttpResponse(
sse_iter(‘gpt-4o-mini’, messages, cache_key, payload[‘sub’]),
content_type=’text/event-stream’
)
resp[‘Cache-Control’] = ‘no-cache’
resp[‘X-Accel-Buffering’] = ‘no’ # for Nginx
return resp

Nginx proxy config (SSE safe)
location /v1/chat/stream {
proxy_pass http://django:8000/v1/chat/stream;
proxy_http_version 1.1;
proxy_set_header Connection ”;
proxy_set_header Host $host;
proxy_buffering off;
proxy_read_timeout 3600;
chunked_transfer_encoding on;
}

Security notes
– Keys only on backend. Never expose provider keys to WordPress or JS.
– Short-lived JWT (5 minutes). Rotate AIGUY_SSO_JWT_SECRET regularly.
– Validate CORS at the proxy. Pin ALLOWED_ORIGINS.
– Enforce rate limits with Redis. Add IP + user-based buckets.
– Log prompts and token counts with PII scrubbing.

Performance and cost
– Stream with SSE to improve UX latency.
– Cache frequent answers by prompt fingerprint (with TTL).
– Temperature low and max_tokens bounded server-side.
– Use connection pooling for Redis and HTTP.
– Add circuit breakers and provider timeouts.

Observability
– Log request_id, user_id, latency, tokens.
– Push metrics to Prometheus or Grafana via StatsD.
– Store error samples with stack traces and masked content.

Testing checklist
– Verify SSE survives 60+ seconds without proxy buffering.
– Confirm JWT expiration fails as expected.
– Confirm rate limit returns 429 and resets on TTL.
– Load test 100 concurrent chats; monitor CPU, Redis ops/sec, and open file handles.
– Validate caching hits on repeated inputs.

Deployment tips
– Put Django behind Nginx; keep workers async-friendly (uvicorn/gunicorn with gevent/uvicorn workers).
– Set Cloudflare to bypass cache on /v1/chat/stream and increase 524 timeout.
– Autoscale based on concurrent connections, not just CPU.

Extensions
– Add chat history with user scoping in Redis or Postgres.
– Add tools/functions (search, orders) via backend tool calling.
– Add moderation before model call and redact PII.

This pattern keeps WordPress simple and secure, offloads AI to a hardened backend, and delivers fast, streaming responses with production safeguards.

Goal
Add an AI assistant to WordPress that is fast, secure, and reliable under load. We’ll avoid long-running PHP, protect API keys, and support streaming. This is a production pattern we’ve deployed in client sites.

Architecture (high level)
– WP Plugin (PHP): UI + REST endpoints, stores minimal state, never holds LLM keys.
– API Gateway/Edge (Cloudflare/AWS): Auth, rate limiting, logging.
– Worker (Python/FastAPI): Orchestrates LLM calls (OpenAI/Azure/Anthropic), tools, retries; pushes results.
– Queue (Redis/SQS): Buffer workloads, smooth spikes.
– Storage: WordPress (postmeta/options) for user-visible data; object store/DB for logs.
– Optional Streaming: SSE via edge worker or reverse proxy, not from PHP-FPM.

Why this pattern
– Security: Keys never live in WP DB or theme code.
– Performance: No long PHP requests; async jobs run outside WordPress.
– Reliability: Queue + retries handle spikes/timeouts.
– Observability: Centralized logs/metrics for LLM latency and errors.

Data flow
1) Browser -> WP REST: user prompt + context (nonce + auth).
2) WP -> API Gateway: signed JWT; enqueue job.
3) Worker -> LLM/tools: executes, streams or finalizes.
4a) Streaming: Gateway proxies SSE to browser.
4b) Async: Worker posts result webhook back to WP; WP stores and notifies client via polling or webhooks.

Security hardening
– Cap checks: Only authenticated users, or signed public sessions with rate limits.
– Nonces for UI actions, WP REST permissions_callback.
– JWT from WP to gateway with short TTL; rotated HS256 secret in env, not DB.
– IP allowlist for worker->WP webhooks.
– Sanitize/escape all content rendered in WP.
– Log PII boundaries; redact before sending to LLM.

WordPress plugin skeleton (core pieces)
// Plugin header omitted for brevity
1) Settings (no LLM keys in WP):
– api_gateway_url
– jwt_issuer, jwt_kid
– public_rate_limit/window
– webhook_secret (HMAC shared with worker)

2) Admin settings + capability checks (manage_options).

3) REST routes:
– /ai/v1/chat (POST) -> enqueues
– /ai/v1/webhook (POST) -> receives results (IP + HMAC verified)
– /ai/v1/status (GET) -> poll job state (transients or custom table)

Minimal REST route example (enqueue)
“`
add_action(‘rest_api_init’, function() {
register_rest_route(‘ai/v1’, ‘/chat’, [
‘methods’ => ‘POST’,
‘callback’ => ‘ai_chat_enqueue’,
‘permission_callback’ => function($req){ return is_user_logged_in() || ai_allow_public(); }
]);
});

function ai_chat_enqueue(WP_REST_Request $req){
$prompt = wp_strip_all_tags($req->get_param(‘prompt’) ?? ”);
if (!$prompt) return new WP_Error(‘bad_request’, ‘Missing prompt’, [‘status’=>400]);

// Rate limit (user or IP)
if (!ai_rate_ok($req)) return new WP_Error(‘rate_limited’,’Try later’,[‘status’=>429]);

// Create job record (custom table or postmeta)
$job_id = ai_create_job([‘user_id’=>get_current_user_id(), ‘prompt’=>$prompt, ‘status’=>’queued’]);

// Sign JWT for gateway
$jwt = ai_sign_jwt([
‘iss’=>get_option(‘ai_jwt_issuer’),
‘sub’=> (string) get_current_user_id(),
‘jti’=> $job_id,
‘iat’=> time(),
‘exp’=> time()+60
]);

// Send to gateway
$resp = wp_remote_post(get_option(‘ai_gateway_url’).’/jobs’, [
‘timeout’=>5,
‘headers’=>[‘Authorization’=>”Bearer $jwt”, ‘Content-Type’=>’application/json’],
‘body’=> wp_json_encode([
‘job_id’=>$job_id,
‘prompt’=>$prompt,
‘context’=> ai_collect_context(),
‘webhook’=> home_url(‘/wp-json/ai/v1/webhook’)
])
]);

if (is_wp_error($resp) || wp_remote_retrieve_response_code($resp) >= 300) {
ai_update_job($job_id,[‘status’=>’error’,’error’=>’enqueue_failed’]);
return new WP_Error(‘upstream_error’,’Queue unavailable’,[‘status’=>502]);
}

return [‘job_id’=>$job_id, ‘status’=>’queued’];
}
“`

Webhook handler (worker -> WP)
“`
register_rest_route(‘ai/v1′,’/webhook’,[
‘methods’=>’POST’,
‘callback’=>’ai_webhook_handler’,
‘permission_callback’=>’__return_true’
]);

function ai_webhook_handler(WP_REST_Request $req){
// Verify HMAC and IP allowlist
if (!ai_verify_hmac($req)) return new WP_Error(‘forbidden’,’bad signature’,[‘status’=>403]);

$job_id = sanitize_text_field($req->get_param(‘job_id’));
$content = wp_kses_post($req->get_param(‘content’) ?? ”);
$usage = $req->get_param(‘usage’) ?? [];

ai_update_job($job_id, [‘status’=>’done’,’content’=>$content,’usage’=>$usage]);
// Optionally cache a hash of the prompt->response
set_transient(‘ai_job_’.$job_id, [‘status’=>’done’], 600);
return [‘ok’=>true];
}
“`

Frontend usage (enqueue + poll)
– Enqueue via wp.ajax or fetch to /wp-json/ai/v1/chat.
– Poll /status until done or use SSE channel if implemented at gateway.
– Render with esc_html or safe HTML subset.

Python worker (FastAPI outline)
– Validates JWT (issuer, exp, kid).
– Pushes to queue (Redis/SQS) with retry/backoff.
– Consumes jobs, calls LLM with tools/function-calls as needed.
– Optional SSE: Streams tokens through gateway to client.
– Posts final result to WP webhook with HMAC.

FastAPI snippets
“`
@app.post(“/jobs”)
def enqueue(job: Job, auth=Depends(verify_jwt)):
q.enqueue(job.dict())
return {“accepted”: True}

def process(job):
try:
resp = client.chat.completions.create(
model=os.getenv(“MODEL”),
messages=[{“role”:”user”,”content”: job[“prompt”]}],
timeout=30
)
content = resp.choices[0].message.content
usage = resp.usage.model_dump() if hasattr(resp,’usage’) else {}
post_webhook(job[“webhook”], job[“job_id”], content, usage)
except Exception as e:
post_webhook_error(job, str(e))
“`

Streaming via SSE (recommended)
– Don’t stream from PHP. Use edge worker that:
– Authenticates client cookie/session with a one-time token from WP.
– Opens SSE to Python worker that proxies LLM token stream.
– Applies per-user rate limits.
– WordPress only issues short-lived stream tokens; never touches LLM socket.

Rate limiting
– Per IP for anonymous, per user_id for logged-in.
– Server-side bucket at gateway (e.g., 30 requests/5m).
– UI throttle + backoff to avoid hammering.

Caching and cost control
– Cache deterministic prompts by normalized hash (strip whitespace, lowercased).
– Store minimal response + usage metrics for analytics.
– Deny-list high-cost paths; enforce max tokens and model allowlist at worker.

Observability
– Centralized logs with job_id correlation across WP, gateway, worker.
– Metrics: queue depth, LLM latency, error rate, token usage per user/site.

Database notes
– For scale, use a custom table ai_jobs (job_id PK, user_id, prompt_hash, status, content MEDIUMTEXT, usage JSON, created_at, updated_at).
– Index status, created_at for cleanup jobs.
– Cron: purge >30 days or anonymize prompt text.

Hardening checklist
– No LLM keys in WordPress DB or code.
– Short JWT TTL; rotate secrets; pin alg and kid.
– Webhook HMAC + IP allowlist.
– Escape output; sanitize HTML.
– Strict CORS only to site origin.
– Disable file edit in wp-config; limit plugin access.

When to choose what
– Small sites: WordPress + API Gateway + single worker (no queue) with short timeouts and SSE.
– Growing traffic: Add queue, retries, and circuit breakers.
– Enterprise: Multi-region gateway, KMS-managed secrets, private networking to WP origin.

Deliverables you can ship today
– WP plugin skeleton above + settings UI.
– FastAPI worker with JWT verify + enqueue + result postback.
– Cloudflare Worker for SSE proxy + rate limit.
– IaC to provision queue, logs, secrets.

Why you need a proxy
– Keeps vendor keys off WordPress.
– Centralizes auth, rate limits, caching, and retries.
– Enables streaming and uniform observability across sites.

High-level architecture
– WordPress site(s) → Proxy (Django ASGI) → Provider APIs (OpenAI/Anthropic/etc.)
– Redis for rate limits, idempotency, and response cache.
– Postgres optional for audit logs.
– Cloudflare → NGINX → Uvicorn (Django ASGI).

Security model
– Per-site API key pair: site_id + site_secret (stored in WP).
– Request signature: HMAC-SHA256 over body + timestamp.
– JWT issued by proxy for short-lived sessions (optional).
– Nonce in WordPress UI, capability checks for settings.
– IP allowlist and user-agent tagging for WordPress clients.
– Enforce TLS end-to-end.

Django proxy (ASGI) essentials
– Django 5.x, Python 3.11+, Redis, httpx (async), uvicorn.
– Endpoints:
– POST /v1/chat (stream or non-stream)
– POST /v1/embeddings
– GET /v1/models (capability discovery)
– Headers:
– X-Site-ID, X-Timestamp, X-Signature, X-Client-Request-ID

Example models.py (optional audit)
from django.db import models

class InferenceLog(models.Model):
request_id = models.CharField(max_length=64, db_index=True)
site_id = models.CharField(max_length=64, db_index=True)
route = models.CharField(max_length=32)
prompt_hash = models.CharField(max_length=64, db_index=True)
provider = models.CharField(max_length=32)
tokens_in = models.IntegerField(default=0)
tokens_out = models.IntegerField(default=0)
status = models.IntegerField()
elapsed_ms = models.IntegerField()
created_at = models.DateTimeField(auto_now_add=True)

Rate limiting (Redis, sliding window)
– Key: rl:{site_id}:{route}
– Allow N requests per window, e.g., 60/min, 600/hour.
– Return 429 with Retry-After.

Pseudo-implementation (views.py, chat streaming)
import asyncio, hmac, hashlib, time, json, os
import httpx
from django.http import StreamingHttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.utils.crypto import constant_time_compare
import redis

r = redis.Redis.from_url(os.environ[“REDIS_URL”], decode_responses=False)
PROVIDER_KEY = os.environ[“PROVIDER_KEY”]
HMAC_SECRET = os.environ[“HMAC_SECRET”].encode()

def verify_sig(raw, ts, sig):
if abs(int(time.time()) – int(ts)) > 60:
return False
mac = hmac.new(HMAC_SECRET, raw + ts.encode(), hashlib.sha256).hexdigest()
return constant_time_compare(mac, sig)

def rl_allow(site_id, route, limit=60, window=60):
key = f”rl:{site_id}:{route}”
now = int(time.time())
pipe = r.pipeline()
pipe.zremrangebyscore(key, 0, now-window)
pipe.zadd(key, {str(now): now})
pipe.zcard(key)
pipe.expire(key, window)
_, _, count, _ = pipe.execute()
return count 3:
yield b”event: errorndata: {“message”:”upstream_failed”}nn”
break
await asyncio.sleep(backoff)
backoff *= 2
yield b”event: donendata: {}nn”

if stream:
return StreamingHttpResponse(gen(), content_type=”text/event-stream”)
else:
# Non-stream path
async with httpx.AsyncClient(timeout=60) as client:
resp = await client.post(
“https://api.openai.com/v1/chat/completions”,
headers={“Authorization”: f”Bearer {PROVIDER_KEY}”},
json={“model”: payload.get(“model”,”gpt-4o-mini”), “messages”: prompt}
)
data = resp.json()
r.setex(ck, 60, json.dumps(data))
return JsonResponse(data, status=resp.status_code)

NGINX (SSE buffering)
– proxy_buffering off;
– proxy_read_timeout 300s;
– add_header Cache-Control no-cache;

Gunicorn/Uvicorn
– uvicorn app.asgi:application –host 0.0.0.0 –port 8000 –workers 2 –loop uvloop –http h11

WordPress plugin (minimal)
– Stores Proxy Base URL, Site ID, Site Secret.
– Adds a shortcode [ai_chat] that renders a simple chat box.
– Uses SSE via EventSource to stream responses.
– Nonces for AJAX init; sanitize all options; only admins can edit.

Plugin main file (ai-proxy-chat/ai-proxy-chat.php)
‘POST’,
‘permission_callback’=>function(){ return wp_verify_nonce($_POST[‘_wpnonce’] ?? ”, ‘aig_sig’); },
‘callback’=>[$this,’sign’]
]);
});
}
public function menu() {
add_options_page(‘AI Proxy Chat’,’AI Proxy Chat’,’manage_options’,’aig-proxy-chat’,[$this,’settings’]);
}
public function register() {
register_setting(self::OPT, self::OPT, [‘sanitize_callback’=>[$this,’sanitize’]]);
add_settings_section(‘main’,’Settings’, ‘__return_false’,’aig-proxy-chat’);
add_settings_field(‘base_url’,’Proxy Base URL’,[$this,’field’],’aig-proxy-chat’,’main’,[‘k’=>’base_url’]);
add_settings_field(‘site_id’,’Site ID’,[$this,’field’],’aig-proxy-chat’,’main’,[‘k’=>’site_id’]);
add_settings_field(‘site_secret’,’Site Secret’,[$this,’field’],’aig-proxy-chat’,’main’,[‘k’=>’site_secret’]);
}
public function sanitize($v){
return [
‘base_url’=>esc_url_raw($v[‘base_url’] ?? ”),
‘site_id’=>sanitize_text_field($v[‘site_id’] ?? ”),
‘site_secret’=>sanitize_text_field($v[‘site_secret’] ?? ”)
];
}
public function field($args){
$o = get_option(self::OPT,[]);
$k = $args[‘k’];
$type = $k===’site_secret’ ? ‘password’ : ‘text’;
printf(”, $type, self::OPT, esc_attr($k), esc_attr($o[$k] ?? ”));
}
public function settings(){
echo ‘

‘;
}
public function assets() {
wp_register_script(‘aig-chat’, plugins_url(‘chat.js’, __FILE__), [], ‘1.0’, true);
wp_localize_script(‘aig-chat’, ‘AIG_CHAT’, [
‘nonce’=>wp_create_nonce(‘aig_sig’),
‘sigEndpoint’=>rest_url(‘aig/v1/sig’)
]);
}
public function shortcode(){
wp_enqueue_script(‘aig-chat’);
ob_start(); ?>

get_param(‘body’) ?? ”;
$ts = time();
$sig = hash_hmac(‘sha256’, $body . $ts, $o[‘site_secret’] ?? ”);
return [‘ts’=>$ts,’sig’=>$sig,’site_id’=>$o[‘site_id’] ?? ”,’base_url’=>$o[‘base_url’] ?? ”];
}
}
new AIG_Proxy_Chat();

Client JS (ai-proxy-chat/chat.js)
(function(){
const log = (t)=>{ const el = document.getElementById(‘aig-log’); el.innerHTML += t + ‘
‘; el.scrollTop=el.scrollHeight; };
document.addEventListener(‘submit’, async (e)=>{
if(e.target.id !== ‘aig-form’) return;
e.preventDefault();
const input = document.getElementById(‘aig-input’);
const msg = input.value.trim(); if(!msg) return;
log(‘You: ‘+msg); input.value = ”;

const body = JSON.stringify({model:’gpt-4o-mini’, stream:true, messages:[{role:’user’,content:msg}]});
const sigRes = await fetch(AIG_CHAT.sigEndpoint, {method:’POST’, credentials:’same-origin’, headers:{‘Content-Type’:’application/x-www-form-urlencoded’}, body:new URLSearchParams({_wpnonce:AIG_CHAT.nonce, body})});
const {ts,sig,site_id,base_url} = await sigRes.json();

const url = base_url.replace(//+$/,”) + ‘/v1/chat’;
const es = new EventSourcePolyfill ? new EventSourcePolyfill(url, {
headers: {‘X-Site-ID’:site_id,’X-Timestamp’:String(ts),’X-Signature’:sig,’Content-Type’:’application/json’},
payload: body
}) : null;

if (es){
let acc = ”;
es.onmessage = (ev)=>{ try {
const d = JSON.parse(ev.data);
const delta = d.choices?.[0]?.delta?.content || d.choices?.[0]?.message?.content || ”;
if(delta){ acc += delta; }
if(delta) log(delta);
} catch(_){} };
es.addEventListener(‘done’, ()=>{ log(‘

‘); es.close(); });
es.addEventListener(‘error’, ()=>{ log(‘Stream error‘); es.close(); });
} else {
// Fallback: POST then append
const res = await fetch(url, {method:’POST’, headers:{‘X-Site-ID’:site_id,’X-Timestamp’:String(ts),’X-Signature’:sig,’Content-Type’:’application/json’}, body});
const data = await res.json();
const text = data.choices?.[0]?.message?.content || ‘[no content]’;
log(text); log(‘

‘);
}
}, true);
})();

Hardening checklist
– WordPress: escape output, sanitize options, restrict settings to manage_options, use nonces everywhere.
– Proxy: validate JSON schema, enforce token limits, redact PII in logs, cap request size (e.g., 256KB), timeouts + retries, 429/503 behavior.
– NGINX: limit_req by IP as outer guard; set client_max_body_size 512k.
– Redis: use ACLs and TLS; set maxmemory with allkeys-lru for cache eviction.
– Keys: rotate provider keys; per-site secrets; revoke on abuse.
– Observability: request_id header, structured JSON logs, latency and token metrics.

Performance notes
– Streaming path: TTFB ~80–150 ms via proxy; throughput limited by provider stream.
– Non-stream with cache: ~5–15 ms from Redis hit.
– Uvicorn workers scale horizontally; keep WordPress PHP-FPM unchanged.

When to extend
– Add /v1/embeddings with response cache and vector store indexing.
– Add model routing policy and quota per site.
– Add document upload pipeline (signed URLs, antivirus, OCR) before LLM.

What we’re building
– A WordPress plugin that streams AI responses to the browser using Server-Sent Events (SSE)
– Front-end shortcode + minimal JS
– Backend REST endpoint with nonce validation, rate limits in Redis, and provider key isolation
– Works with any SSE-capable model provider (OpenAI, Anthropic, etc.)

High-level architecture
– Browser: Shortcode renders chat UI and script. JS posts user messages to /wp-json/ai/v1/chat and reads streaming tokens via EventSource.
– WordPress plugin: Validates nonce, enforces rate limit, proxies request to the AI provider, streams tokens back.
– Redis: Token bucket per user/IP for rate limiting.
– Secrets: API key injected via environment (wp-config.php), never exposed to front-end.

Prereqs
– WordPress 6.4+
– PHP 8.1+
– Redis available (phpredis or Predis). Fallback to transients if Redis isn’t available.
– Web server buffering disabled for the streaming route (notes below).

Plugin: file structure
– wp-content/plugins/ai-stream-chat/ai-stream-chat.php
– wp-content/plugins/ai-stream-chat/assets/chat.js

ai-stream-chat.php (core plugin)
esc_url_raw(rest_url(self::REST_NAMESPACE . ‘/’ . self::REST_ROUTE)),
‘nonce’ => wp_create_nonce(self::NONCE_ACTION),
]);
wp_enqueue_script($handle);
wp_enqueue_style(‘ai-sc-css’, false);
$css = ‘.ai-sc{max-width:720px;margin:1rem auto;padding:1rem;border:1px solid #ddd;border-radius:8px}.ai-sc-log{white-space:pre-wrap;font-family:system-ui, -apple-system, Segoe UI, Roboto, Arial;padding:.5rem;height:320px;overflow:auto;background:#fafafa;border:1px solid #eee;border-radius:6px;margin:.5rem 0}.ai-sc-row{display:flex;gap:.5rem}.ai-sc-row input{flex:1;padding:.6rem}.ai-sc-row button{padding:.6rem 1rem}’;
wp_add_inline_style(‘ai-sc-css’, $css);
}

public function shortcode($atts) {
ob_start(); ?>
<div class="ai-sc" data-endpoint="”>