AI Agent & Custom Streaming

In addition to the built-in Admin assistant, Dyrected exports a fully-typed AIAgent service directly from @dyrected/core. You can use AIAgent to power customer-facing chat widgets, real-time quote calculators, background workflows, or external bots on Telegram and WhatsApp.

By the end of this guide, you will understand how AIAgent coordinates your content schema and database, and you will have a complete, runnable chat endpoint ready for production.

The AIAgent mental model

AIAgent unifies your database adapter, schema collections, custom tools, and conversation history under one service boundary.

┌─────────────────────────────────────────────────────────────┐
│             Your Application Route or Webhook               │
│               (Next.js, Nuxt, Remix, Hono)                  │
└──────────────────────────────┬──────────────────────────────┘
                               │
                new AIAgent({ db, config, ... })
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│                      Dyrected AIAgent                       │
│  • Grounding: Builds system prompt from your collections    │
│  • Tools: Automatically binds query and mutation tools      │
│  • Custom Tools: Executes your config.ai.tools functions   │
│  • Memory: Persists turns to _dyrected_ai_messages in DB    │
│  • Streaming: Emits tokens & handles UI protocol streams    │
└─────────────────────────────────────────────────────────────┘

1. Initializing AIAgent

To use the agent in any server handler or script, instantiate AIAgent with your database adapter and configuration:

import { AIAgent } from "@dyrected/core";
import config from "./dyrected.config";

const agent = new AIAgent({
  db: config.db,
  config,
  projectId: "default",
  userId: "usr_alex",
  userName: "Alex Johnson",
  userRole: "customer",
});

2. Creating a streaming chat endpoint

The recommended way to expose AI streaming to your frontend is a server endpoint that returns a UI-compatible stream.

AIAgent provides two streaming methods:

createStreamResponse(threadId, userMessage, clientMessages?, signal?): Returns a web-standard Response formatted for UI streaming hooks like useChat.
streamReply(threadId, userMessage): An async generator that emits raw text chunks and automatically commits both the user prompt and assistant response to your database.

Create an API route in app/api/ai/chat/route.ts:

// app/api/ai/chat/route.ts
import { NextRequest, NextResponse } from "next/server";
import { AIAgent } from "@dyrected/core";
import config from "@/dyrected.config";

export const dynamic = "force-dynamic";

export async function POST(req: NextRequest) {
  try {
    const body = await req.json();
    const { messages = [], threadId } = body;

    const lastMessage = messages[messages.length - 1];
    const userMessage =
      typeof lastMessage?.content === "string"
        ? lastMessage.content
        : lastMessage?.parts?.[0]?.text || "";

    if (!userMessage.trim()) {
      return NextResponse.json({ error: "Message content is required" }, { status: 400 });
    }

    const agent = new AIAgent({
      db: config.db,
      config,
      projectId: "default",
      userId: threadId || "anonymous",
      userName: "Website Visitor",
      userRole: "customer",
    });

    // 1. Retrieve the existing thread or create one with your session ID
    const thread = await agent.getOrCreateThread(
      threadId,
      userMessage.slice(0, 40) || "Customer Inquiry"
    );

    // 2. Stream the response directly (automatically saves history to DB)
    return agent.createStreamResponse(
      thread.id,
      userMessage,
      messages,
      req.signal
    );
  } catch (error: any) {
    return NextResponse.json(
      { error: "Failed to generate AI response", details: error?.message },
      { status: 500 }
    );
  }
}

Create a server handler in server/api/ai/chat.post.ts:

// server/api/ai/chat.post.ts
import { AIAgent } from "@dyrected/core";
import config from "~~/dyrected.config";

export default defineEventHandler(async (event) => {
  const body = await readBody(event).catch(() => ({}));
  const { messages = [], threadId } = body;

  const lastMessage = messages[messages.length - 1];
  const userMessage =
    typeof lastMessage?.content === "string"
      ? lastMessage.content
      : lastMessage?.parts?.[0]?.text || "";

  if (!userMessage.trim()) {
    throw createError({ statusCode: 400, message: "Message content is required" });
  }

  const agent = new AIAgent({
    db: config.db,
    config,
    projectId: "default",
    userId: threadId || "anonymous",
    userName: "Website Visitor",
    userRole: "customer",
  });

  // 1. Retrieve the existing thread or create one with your session ID
  const thread = await agent.getOrCreateThread(
    threadId,
    userMessage.slice(0, 40) || "Customer Inquiry"
  );

  // 2. Stream the response directly using Web Standards Response
  const response = await agent.createStreamResponse(
    thread.id,
    userMessage,
    messages
  );

  return sendWebResponse(event, response);
});

3. Streaming raw tokens with custom streams

If you are building a custom client widget or want direct control over raw streaming tokens, use agent.streamReply().

streamReply automatically commits the user message, streams tokens chunk-by-chunk, and persists the finished assistant reply when the generator completes.

// app/api/ai/stream/route.ts
import { NextRequest, NextResponse } from "next/server";
import { AIAgent } from "@dyrected/core";
import config from "@/dyrected.config";

export async function POST(req: NextRequest) {
  const { message, threadId = `session-${Date.now()}` } = await req.json();

  if (!message) {
    return NextResponse.json({ error: "Message is required" }, { status: 400 });
  }

  const agent = new AIAgent({
    db: config.db,
    config,
    projectId: "default",
    userId: threadId,
  });

  // Ensure thread exists in Postgres / SQLite
  await agent.getOrCreateThread(threadId, message.slice(0, 30));

  const encoder = new TextEncoder();
  const generator = agent.streamReply(threadId, message);

  const readable = new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of generator) {
          controller.enqueue(encoder.encode(chunk));
        }
        controller.close();
      } catch (err) {
        controller.error(err);
      }
    },
  });

  return new Response(readable, {
    headers: {
      "Content-Type": "text/plain; charset=utf-8",
      "X-Thread-Id": threadId,
    },
  });
}

// server/api/ai/stream.post.ts
import { AIAgent } from "@dyrected/core";
import config from "~~/dyrected.config";

export default defineEventHandler(async (event) => {
  const { message, threadId = `session-${Date.now()}` } = await readBody(event);

  if (!message) {
    throw createError({ statusCode: 400, message: "Message is required" });
  }

  const agent = new AIAgent({
    db: config.db,
    config,
    projectId: "default",
    userId: threadId,
  });

  await agent.getOrCreateThread(threadId, message.slice(0, 30));

  setResponseHeader(event, "Content-Type", "text/plain; charset=utf-8");
  setResponseHeader(event, "X-Thread-Id", threadId);

  const encoder = new TextEncoder();
  const generator = agent.streamReply(threadId, message);

  const readable = new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of generator) {
          controller.enqueue(encoder.encode(chunk));
        }
        controller.close();
      } catch (err) {
        controller.error(err);
      }
    },
  });

  return sendWebResponse(event, new Response(readable));
});

4. Ingesting Telegram or WhatsApp webhooks

Because AIAgent.getOrCreateThread() supports explicit external session IDs, connecting messaging bots to persistent customer threads is straightforward.

Here is a complete Telegram webhook handler:

// app/api/telegram/webhook/route.ts
import { NextRequest, NextResponse } from "next/server";
import { AIAgent } from "@dyrected/core";
import config from "@/dyrected.config";

export async function POST(req: NextRequest) {
  const update = await req.json().catch(() => null);
  const message = update?.message;

  if (!message?.text) {
    return NextResponse.json({ ok: true });
  }

  const chatId = String(message.chat.id);
  const senderName = message.from.first_name || "Customer";
  const threadId = `tg-${chatId}`;

  const agent = new AIAgent({
    db: config.db,
    config,
    projectId: "default",
    userId: threadId,
    userName: senderName,
    userRole: "customer",
  });

  // 1. Maintain a persistent conversation thread keyed to the Telegram chat ID
  await agent.getOrCreateThread(threadId, `Telegram: ${senderName}`);

  // 2. Stream and collect the reply
  const result = agent.streamReply(threadId, message.text);
  let finalReply = "";

  for await (const chunk of result) {
    finalReply += chunk;
  }

  // 3. Send response back to Telegram
  const botToken = process.env.TELEGRAM_BOT_TOKEN;
  if (botToken) {
    await fetch(`https://api.telegram.org/bot${botToken}/sendMessage`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        chat_id: chatId,
        text: finalReply || "Thanks for your message! Our team will get back to you shortly.",
      }),
    });
  }

  return NextResponse.json({ ok: true });
}

5. Managing conversation threads programmatically

You can inspect, list, and clear conversation history in your database with the built-in thread management methods:

// 1. Get or create a thread with a custom or auto-generated ID
const thread = await agent.getOrCreateThread("custom-thread-id", "Product Inquiries");

// 2. Fetch all recorded messages for this thread (chronological order)
const messages = await agent.getMessages(thread.id);

// 3. Generate a concise 3-5 word title using AI
const smartTitle = await agent.generateTitle("I need 500 business cards printed on 350gsm cardstock");
await agent.updateThreadTitle(thread.id, smartTitle);

// 4. List the most recent threads for the project
const threads = await agent.listThreads(20);

// 5. Delete a thread and all associated messages
await agent.deleteThread(thread.id);