10xGraph with Next.js: Build a Streaming Agent Frontend

[ Next.js (App Router) ]
   ├── React UI (client component)
   └── Route Handler / Server Action (Node runtime)
              │
              ▼ HTTP / SSE
[ 10xGraph API ] ← deployed separately
              │
              ▼
[ Postgres + Redis (PgCheckpointer) ]

npm install @10xscale/agentflow-client

// app/api/agent/stream/route.ts
import {NextRequest} from "next/server";
import {AgentFlowClient, Message} from "@10xscale/agentflow-client";
import {auth} from "@/lib/auth";

export const runtime = "nodejs";

export async function POST(req: NextRequest) {
  const user = await auth();
  if (!user) return new Response("Unauthorized", {status: 401});

  const {text} = await req.json();

  const client = new AgentFlowClient({
    baseUrl: process.env.AGENTFLOW_URL!,
    headers: {Authorization: `Bearer ${process.env.AGENTFLOW_API_KEY}`},
  });

  const stream = client.stream(
    [Message.text_message(text)],
    {config: {thread_id: `user-${user.id}`}, recursion_limit: 25},
  );

  // Convert AsyncIterable to ReadableStream and re-emit as SSE
  const encoder = new TextEncoder();
  const body = new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of stream) {
          controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`));
        }
        controller.enqueue(encoder.encode(`event: done\ndata: {}\n\n`));
      } catch (e) {
        controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({message: String(e)})}\n\n`));
      } finally {
        controller.close();
      }
    },
  });

  return new Response(body, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "X-Accel-Buffering": "no",
    },
  });
}

"use client";

import {useState, useCallback} from "react";

// Mirrors StreamChunk from @10xscale/agentflow-client. `event` is one of
// StreamEventType: "message" | "updates" | "state" | "error".
type Chunk = {
  event: "message" | "updates" | "state" | "error";
  message?: {role: string; content: Array<Record<string, unknown>>} | null;
  state?: Record<string, unknown> | null;
  data?: Record<string, unknown>;
  thread_id?: string;
  run_id?: string;
};

// `message.content` is an array of content blocks; concatenate the text ones.
// (The Message class has a `text()` method, but chunks that crossed the proxy
// as JSON are plain objects.)
const chunkText = (chunk: Chunk) =>
  (chunk.message?.content ?? [])
    .filter((b) => b.type === "text")
    .map((b) => String(b.text ?? ""))
    .join("");

export function useAgentStream() {
  const [output, setOutput] = useState("");
  const [streaming, setStreaming] = useState(false);
  const [status, setStatus] = useState("");

  const send = useCallback(async (text: string) => {
    setOutput("");
    setStreaming(true);
    try {
      const r = await fetch("/api/agent/stream", {
        method: "POST",
        headers: {"Content-Type": "application/json"},
        body: JSON.stringify({text}),
      });
      if (!r.ok || !r.body) throw new Error("stream failed");

      const reader = r.body.getReader();
      const decoder = new TextDecoder();
      let buf = "";

      for (;;) {
        const {value, done} = await reader.read();
        if (done) break;
        buf += decoder.decode(value, {stream: true});

        // Split SSE events on "\n\n"
        const events = buf.split("\n\n");
        buf = events.pop() ?? "";
        for (const ev of events) {
          const dataLine = ev.split("\n").find((l) => l.startsWith("data: "));
          if (!dataLine) continue;
          const json = dataLine.slice(6);
          if (json === "{}") continue;
          const chunk = JSON.parse(json) as Chunk;
          if (chunk.event === "message") {
            setOutput((s) => s + chunkText(chunk));
          } else if (chunk.event === "updates") {
            // Run-lifecycle marker, e.g. {status: "completed"}.
            setStatus(String(chunk.data?.status ?? ""));
          } else if (chunk.event === "error") {
            // The server never raises mid-stream; it emits this chunk instead.
            throw new Error(String(chunk.data?.reason ?? "stream error"));
          }
        }
      }
    } finally {
      setStreaming(false);
    }
  }, []);

  return {output, streaming, status, send};
}

"use client";
import {useAgentStream} from "./useAgentStream";

export default function Chat() {
  const {output, streaming, status, send} = useAgentStream();
  return (
    <div>
      <button onClick={() => send("Weather in Tokyo?")} disabled={streaming}>
        Ask
      </button>
      {status ? <small>{status}</small> : null}
      <pre>{output}</pre>
    </div>
  );
}

// app/actions/agent.ts
"use server";
import {AgentFlowClient, Message} from "@10xscale/agentflow-client";
import {auth} from "@/lib/auth";

const client = new AgentFlowClient({
  baseUrl: process.env.AGENTFLOW_URL!,
  headers: {Authorization: `Bearer ${process.env.AGENTFLOW_API_KEY}`},
});

export async function ask(text: string) {
  const user = await auth();
  if (!user) throw new Error("Unauthorized");

  const result = await client.invoke(
    [Message.text_message(text)],
    {config: {thread_id: `user-${user.id}`}},
  );
  return result.messages.at(-1)?.text() ?? "";
}