Introduction
When building modern web applications, the Headless WordPress + Nuxt.js architecture combination is increasingly favored by developers. WordPress provides powerful content management on the backend, Nuxt.js handles high-performance frontend rendering, and GraphQL serves as the data bridge between them. However, when we need to display AI-generated content on the frontend — especially in streaming scenarios — traditional REST API approaches fall short. This article provides a detailed guide on implementing AI content streaming through GraphQL in this architecture.
Why GraphQL + SSE?
Traditional REST APIs face several core challenges when handling AI streaming output:
Request-Response Model Limitations: REST follows a one-request-one-response pattern and cannot continuously push data. Streaming typically requires WebSocket, which adds architectural complexity and maintenance overhead.
Data Over-fetching: REST endpoints return fixed data structures. The frontend may only need a few fields but must receive large amounts of redundant information. In a Headless architecture, pages often require multiple REST endpoints to assemble complete data, increasing network overhead and page load times.
Limited Extensibility: When upgrading AI services or switching models, REST interfaces often require synchronized modifications, creating tight coupling between frontend and backend.
GraphQL's advantages lie in on-demand querying — the frontend precisely specifies needed fields, reducing data transfer; all data is fetched through a single GraphQL endpoint, eliminating multiple requests; and custom Mutations and Subscriptions enable flexible AI capability extension. Combined with SSE (Server-Sent Events), we can achieve efficient real-time data pushing within the GraphQL framework.
Architecture Design
The overall architecture consists of three clearly decoupled layers:
Nuxt.js Frontend (SSE Consumer)
↕ SSE (text/event-stream)
WordPress + WPGraphQL Plugin
↕ GraphQL Mutation + REST SSE Endpoint
AI Service (DeepSeek / OpenAI / Claude)On the WordPress side, we register a custom Mutation through the WPGraphQL plugin. This Mutation receives the user's prompt, calls the AI service, and returns a stream ID. The frontend then consumes the streaming data through a dedicated SSE endpoint. This design organically combines GraphQL's query capabilities with SSE's real-time pushing, with each component playing to its strengths.
Backend Implementation: WordPress + WPGraphQL
Step 1: Register a GraphQL Mutation
First, register a custom Mutation in your theme or plugin's functions.php to receive prompts and generate stream IDs:
add_action('graphql_register_types', function () {
register_graphql_mutation('generateAiContent', [
'inputFields' => [
'prompt' => ['type' => 'String'],
'model' => ['type' => 'String'],
],
'outputFields' => [
'streamId' => ['type' => 'String'],
],
'mutateAndGetPayload' => function ($input) {
$prompt = sanitize_text_field($input['prompt']);
$model = $input['model'] ?? 'deepseek-chat';
$stream_id = wp_generate_uuid4();
set_transient("ai_stream_{$stream_id}", [
'prompt' => $prompt,
'model' => $model,
], 300);
return ['streamId' => $stream_id];
},
]);
});We use the WordPress Transients API to temporarily store request information with a 5-minute expiration to prevent memory leaks.
Step 2: Create the SSE Endpoint
Register a custom REST API endpoint to stream AI-generated content. Key points: set correct response headers, use cURL's CURLOPT_WRITEFUNCTION for chunked data transfer, and disable Nginx buffering:
add_action('rest_api_init', function () {
register_rest_route('longxiao/v1', '/ai/stream/(?P<id>[a-zA-Z0-9-]+)', [
'methods' => 'GET',
'callback' => function ($request) {
$stream_id = $request->get_param('id');
$data = get_transient("ai_stream_{$stream_id}");
if (!$data) {
return new WP_Error('not_found', 'Stream not found', ['status' => 404]);
}
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
header('X-Accel-Buffering: no');
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => 'https://api.deepseek.com/v1/chat/completions',
CURLOPT_POST => true,
CURLOPT_HTTPHEADER => [
'Authorization: Bearer ' . DEEPSEEK_API_KEY,
'Content-Type: application/json',
],
CURLOPT_POSTFIELDS => json_encode([
'model' => $data['model'],
'messages' => [['role' => 'user', 'content' => $data['prompt']]],
'stream' => true,
]),
CURLOPT_WRITEFUNCTION => function ($ch, $chunk) {
$lines = explode("n", $chunk);
foreach ($lines as $line) {
$line = trim($line);
if (str_starts_with($line, 'data: ')) {
$json = substr($line, 6);
if ($json === '[DONE]') {
echo "data: [DONE]nn";
} else {
$decoded = json_decode($json, true);
$content = $decoded['choices'][0]['delta']['content'] ?? '';
if ($content) {
echo "data: " . json_encode(['content' => $content]) . "nn";
}
}
ob_flush();
flush();
}
}
return strlen($chunk);
},
]);
curl_exec($ch);
curl_close($ch);
delete_transient("ai_stream_{$stream_id}");
exit;
},
]);
});Important: the CURLOPT_WRITEFUNCTION callback must use ob_flush() and flush() to ensure real-time data output to the client. Always delete the transient after processing to free resources.
Frontend Implementation: Nuxt.js Composable
On the Nuxt.js side, we create a Composable to handle GraphQL requests and SSE stream consumption. The core approach has two steps: first, get the streamId through a GraphQL Mutation; second, consume the SSE stream via the fetch API:
// composables/useAiStream.ts
export function useAiStream() {
const content = ref('')
const isStreaming = ref(false)
const error = ref<string | null>(null)
async function streamGenerate(prompt: string, model = 'deepseek-chat') {
content.value = ''
isStreaming.value = true
error.value = null
try {
const mutation = `
mutation GenerateAiContent($prompt: String!, $model: String!) {
generateAiContent(input: { prompt: $prompt, model: $model }) {
streamId
}
}
`
const { data } = await useGql({ query: mutation, variables: { prompt, model } })
const streamId = data.value?.generateAiContent?.streamId
if (!streamId) throw new Error('Failed to get stream ID')
const streamUrl = `https://api.your-site.com/wp-json/longxiao/v1/ai/stream/${streamId}`
const controller = new AbortController()
const timeout = setTimeout(() => controller.abort(), 120000)
const response = await fetch(streamUrl, { signal: controller.signal })
const reader = response.body?.getReader()
const decoder = new TextDecoder()
if (!reader) throw new Error('Failed to read response stream')
while (true) {
const { done, value } = await reader.read()
if (done) break
const chunk = decoder.decode(value, { stream: true })
const lines = chunk.split('n')
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = line.slice(6)
if (data === '[DONE]') break
try {
const parsed = JSON.parse(data)
content.value += parsed.content || ''
} catch {}
}
}
}
clearTimeout(timeout)
} catch (e: any) {
if (e.name !== 'AbortError') {
error.value = e.message || 'Streaming error'
}
} finally {
isStreaming.value = false
}
}
return { content, isStreaming, error, streamGenerate }
}Key design decisions: using AbortController for 2-minute timeout protection; reactive binding via ref for automatic Vue component UI updates; distinguishing AbortError from other errors to avoid showing error messages on timeout.
Vue Component Integration
<template>
<div class="ai-chat">
<textarea v-model="prompt" placeholder="Enter your question..." />
<button @click="handleGenerate" :disabled="isStreaming">
{{ isStreaming ? 'Generating...' : 'Send' }}
</button>
<div class="ai-output" v-html="renderedContent" />
<p v-if="error" class="error">{{ error }}</p>
</div>
</template>
<script setup>
const prompt = ref('')
const { content, isStreaming, error, streamGenerate } = useAiStream()
const renderedContent = computed(() => content.value.replace(/n/g, '<br>'))
function handleGenerate() {
if (!prompt.value.trim() || isStreaming.value) return
streamGenerate(prompt.value)
}
</script>Performance Optimization and Production Practices
Nginx Buffer Configuration
Nginx reverse proxies buffer backend responses by default, preventing SSE streams from pushing in real time. You must disable buffering for the SSE endpoint:
location /wp-json/longxiao/v1/ai/stream/ {
proxy_buffering off;
proxy_cache off;
proxy_set_header X-Accel-Buffering no;
proxy_read_timeout 300s;
chunked_transfer_encoding on;
}Connection Pooling and Concurrency Control
In high-concurrency scenarios, use Redis to manage the AI request queue to prevent overwhelming the AI service with simultaneous requests, which can cause rate limiting or timeouts. A simple token bucket algorithm on the WordPress side can control request rates.
Error Handling and Reconnection
SSE connections can drop due to network fluctuations. Implement automatic reconnection logic in the Composable. When an abnormal disconnection is detected, send a resume request with the length of already-received content to prevent users from seeing interrupted output.
Conclusion
Through the WPGraphQL + SSE combination, we have successfully implemented AI content streaming in a WordPress Headless architecture. The advantages of this approach include: architectural simplicity — no additional WebSocket server required, reducing operational costs; progressive rendering — users see content appear word by word without waiting for a complete response, providing an excellent interactive experience; flexible extensibility — easily switch between DeepSeek, OpenAI, Claude, and other AI models; seamless integration with existing tech stacks — fully leveraging WordPress's plugin ecosystem and Nuxt.js's full-stack capabilities. In real-world projects, you can further extend this architecture with Markdown rendering, code highlighting, conversation history management, and more to build a complete AI chat application.
![Longxiao Theme-E-Commerce & Membership Integrated Product Plugin: Turn Content Monetization from [Can Sell] to [Sells Well]](https://demo.salongweb.com/longxiao/thumb/44.webp?x-oss-process=image/resize,m_fill,w_600,h_338/quality,q_60/format,webp)
