Pydantic Graph and CopilotKit: A Minimal Full-Stack Workflow Demo

In a previous article, I walked through connecting Pydantic AI and CopilotKit with a plain Agent.

That version is very direct.

FastAPI exposes a Pydantic AI agent through AG-UI, the CopilotKit runtime registers that backend endpoint, and the frontend renders a CopilotKit assistant UI.

This article builds the same kind of full-stack demo, but swaps the backend implementation from a standalone Pydantic AI Agent to a Pydantic Graph workflow.

The stack is similar:

• FastAPI for the backend API
• Pydantic AI for the model-backed chat step
• Pydantic Graph for the workflow structure
• AG-UI as the protocol layer
• Next.js for the frontend
• CopilotKit for the assistant UI

The goal is still modest.

I want the smallest useful demo that shows how these pieces fit together, while preserving enough structure to grow into a more capable graph-based assistant later.

Same frontend, different backend shape

CopilotKit still speaks AG-UI. The graph needs a bridge.

The important difference is on the backend. A Pydantic AI Agent can be handed directly to the AG-UI adapter. A Pydantic Graph is a workflow object, so this demo keeps AG-UI at the FastAPI router boundary. The router parses the AG-UI request, restores message history, runs the graph with app-native input, converts the graph result back into AG-UI messages, and emits the response events.

That bridge is small, but it is the part of the integration worth studying. It is also the part that may become unnecessary if Pydantic Graph eventually gets native AG-UI support.

How the pieces fit together

From the browser’s point of view, this version looks almost the same as the plain agent demo.

The request flow is:

• The browser talks to the Next.js app.
• Next.js exposes a CopilotKit API route.
• The CopilotKit runtime forwards agent requests to the FastAPI backend.
• FastAPI exposes a Pydantic Graph endpoint that speaks AG-UI.
• The graph runs a chat step backed by a Pydantic AI Agent.
• The graph returns Pydantic AI message history in app-native output.
• The router converts that output into AG-UI messages and streams the final assistant response.

That is the useful boundary.

The frontend only knows there is an AG-UI-compatible HTTP agent named demoGraph. The backend can decide whether that endpoint is powered by a single agent, a graph, or a more complex workflow.

That split is the main production-readiness improvement in the current repository. The graph no longer imports AG-UI or returns AG-UI-shaped output. It receives DemoGraphInput, returns DemoGraphOutput, and leaves HTTP protocol translation to the router.

Why graph workflows matter

For a simple assistant, a graph workflow may feel unnecessary.

The value tends to appear once workflows become:

• Multi-stage
• Long-running
• Stateful
• Approval-heavy
• Operationally sensitive

That is one reason I have become increasingly interested in graph orchestration recently.

I wrote about the broader architectural and governance side of this separately in In the AI Regulatory Landscape, Agentic Graphs Can Make a Lot of Sense.

The short version is that deterministic workflows still matter. Modern LLMs are powerful, but many business processes still benefit from explicit control, visible transitions, validation, approvals, and auditable flow logic.

That doesn’t mean every assistant needs a graph. It just means graphs can become very useful once the workflow itself is part of the product, not just the model response.

Why Pydantic Graph feels interesting

One thing I particularly like about Pydantic Graph is how naturally it aligns with the broader philosophy of the Pydantic ecosystem.

Once workflows become more deterministic and state-oriented, strong typing, explicit validation, and structured state transitions start feeling increasingly valuable.

That isn’t a criticism of LangGraph or the LangChain ecosystem. I still think LangGraph is an important and capable framework, and I have written about it several times.

This article is simply exploring a different tool with a slightly different philosophy. For some workflows, the combination of deterministic orchestration and Pydantic’s validation-first approach feels like a very interesting direction.

Code snapshot

The GitHub repository for this project is:

• pydantic-graph-copilotkit-demo

The repository can keep evolving after this article is published, so the exact code for this article is preserved separately:

• Branch: article/001-pydantic-graph-copilotkit
• Tag: article-001

To check out the exact version used by this article:

git fetch --all --tags
git checkout article-001

Use the branch if you want to inspect or compare the article version by name:

git checkout article/001-pydantic-graph-copilotkit

Use main for the latest version of the demo.

Prerequisites

• Python 3.13+
• uv, unless you prefer another Python package manager
• Node.js and pnpm
• An OpenAI API key

Create your project

All terminal commands below are written for Linux or macOS shells. Windows PowerShell users may need to adjust the commands slightly.

Open a terminal in the directory where you want to create the project and run:

uv init pydantic-graph-copilotkit-demo

Change into the new project:

cd pydantic-graph-copilotkit-demo

Sync the environment:

uv sync

Activate the virtual environment:

source .venv/bin/activate

Git ignore

If you plan to commit this project to Git, create a .gitignore before your first commit. At minimum, include:

.env
.venv
__pycache__/

Do this before committing so secrets and generated files do not end up in source control.

Backend setup

Install the minimum Python packages needed for this demo:

uv add python-dotenv fastapi uvicorn "pydantic-ai-slim[ag-ui,openai]"

The graph support used here comes through the Pydantic AI package set. The ag-ui extra gives us the AG-UI adapter utilities, and the openai extra gives us the provider integration used by the demo agent.

I like keeping the graph, graph steps, state, and API router separated because graph workflows usually grow beyond a single file quickly.

From the project root, create the backend folders:

mkdir -p backend/agents/demo_graph/steps backend/routers

The structure should now include:

backend/
  agents/
    demo_graph/
      steps/
  routers/

Create package marker files:

touch backend/agents/__init__.py backend/agents/demo_graph/__init__.py backend/agents/demo_graph/steps/__init__.py

Create backend/.env:

OPENAI_API_KEY=<your_openai_api_key_here>

Do not commit this file to GitHub.

Graph models, state, and step

The demo graph stores Pydantic AI message history in state.

It also uses explicit input and output models so the graph isn’t tied to AG-UI request or response objects.

Create backend/agents/demo_graph/models.py:

from pydantic import BaseModel
from pydantic_ai.messages import ModelMessage


class DemoGraphInput(BaseModel):
    prompt: str


class DemoGraphOutput(BaseModel):
    messages: list[ModelMessage]

Create backend/agents/demo_graph/state.py:

from dataclasses import dataclass, field

from pydantic_ai.messages import ModelMessage


@dataclass
class AppState:
    messages: list[ModelMessage] = field(default_factory=list)

Create backend/agents/demo_graph/config.py:

MODEL = "openai:gpt-4.1-mini"
INSTRUCTIONS = "Be extroverted and concise."

Create backend/agents/demo_graph/steps/chat.py:

from pydantic_ai import Agent
from pydantic_graph.step import StepContext

from ..config import INSTRUCTIONS, MODEL
from ..models import DemoGraphInput, DemoGraphOutput
from ..state import AppState

agent = Agent(model=MODEL, instructions=INSTRUCTIONS)


async def chat(ctx: StepContext[AppState, None, DemoGraphInput]) -> DemoGraphOutput:
    result = await agent.run(
        ctx.inputs.prompt,
        message_history=ctx.state.messages,
    )

    messages = result.all_messages()
    ctx.state.messages = messages

    return DemoGraphOutput(messages=messages)

This module creates the Pydantic AI agent once and reuses it for each graph run. The chat step receives graph-native input, runs the underlying agent with the current message history, writes the updated message history back into graph state, and returns graph-native output.

The point is to establish the state pattern before adding more graph nodes.

Build the graph

Create backend/agents/demo_graph/graph.py:

from pydantic_graph.graph_builder import GraphBuilder

from .models import DemoGraphInput, DemoGraphOutput
from .state import AppState
from .steps.chat import chat


def build_graph():
    graph_builder = GraphBuilder(
        state_type=AppState,
        input_type=DemoGraphInput,
        output_type=DemoGraphOutput,
    )

    chat_node = graph_builder.step(chat, label="Chat")

    graph_builder.add(
        graph_builder.edge_from(graph_builder.start_node).to(chat_node),
        graph_builder.edge_from(chat_node).to(graph_builder.end_node),
    )

    return graph_builder.build()


graph = build_graph()

Keep backend/agents/demo_graph/__init__.py lightweight:

"""Pydantic Graph demo package."""

There is one graph step:

• chat runs the model-backed agent, updates graph state, and returns DemoGraphOutput.

The graph output deliberately stays app-native. The router will shape that result as {"messages": ...} only after converting the Pydantic AI messages into AG-UI messages for CopilotKit.

Native AG-UI support may eventually remove this bridge

At the time of writing, Pydantic Graph workflows do not provide the same direct AG-UI adapter path as a plain Pydantic AI Agent.

That is why this demo uses a lightweight bridge layer between the graph workflow and the AG-UI protocol.

I also opened a feature request for this in the Pydantic AI repository: Native UIAdapter / AG-UI support for Pydantic Graph Beta workflows #5492.

If native support lands, this article and the demo repository can be simplified. Until then, the custom bridge is useful because it makes the integration boundary explicit.

The graph router bridge

This is the main difference from the previous agent article.

For a plain Pydantic AI Agent, the FastAPI route can be tiny:

from fastapi import Request
from pydantic_ai.ui.ag_ui import AGUIAdapter
from starlette.responses import Response


@router.post("/demo-agent")
async def demo_agent_endpoint(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=demo_agent)

That works because the adapter knows how to dispatch requests to a Pydantic AI agent.

The graph object is different. It is a pydantic_graph graph, not a Pydantic AI AbstractAgent, and it does not expose the same event-stream interface that AGUIAdapter.dispatch_request() expects.

So the graph route uses AGUIAdapter for the parts that still fit:

• Parse the incoming AG-UI request body.
• Convert prior AG-UI messages into Pydantic AI message history.
• Convert final Pydantic AI messages back into AG-UI messages.

Then it manually emits the AG-UI run and text message events.

Create backend/routers/graph_router.py:

from collections.abc import AsyncIterator, Sequence
from uuid import uuid4

from ag_ui.core import (
    AssistantMessage,
    Message,
    RunAgentInput,
    RunErrorEvent,
    RunFinishedEvent,
    RunStartedEvent,
    TextMessageContentEvent,
    TextMessageEndEvent,
    TextMessageStartEvent,
    UserMessage,
)
from ag_ui.encoder import EventEncoder
from fastapi import APIRouter, Request
from pydantic_ai.messages import ModelMessage
from pydantic_ai.ui.ag_ui import AGUIAdapter
from starlette.responses import Response, StreamingResponse

from agents.demo_graph.graph import graph as demo_graph
from agents.demo_graph.models import DemoGraphInput
from agents.demo_graph.state import AppState

router = APIRouter()


@router.get("/")
async def root() -> dict[str, str]:
    return {"message": "Hello, I'm the graph API!"}


@router.get("/health")
async def health() -> dict[str, str]:
    return {
        "api": "graph",
        "status": "healthy",
    }

The helper functions do three small jobs:

• Find the latest user message.
• Treat messages before that latest user message as previous conversation history.
• Find the latest assistant response after the graph runs.

def _user_text(message: UserMessage) -> str:
    if isinstance(message.content, str):
        return message.content

    return "\n".join(part.text for part in message.content if part.type == "text")


def _latest_user_message(run_input: RunAgentInput) -> tuple[int, str]:
    for index in range(len(run_input.messages) - 1, -1, -1):
        message = run_input.messages[index]
        if isinstance(message, UserMessage):
            return index, _user_text(message)

    return len(run_input.messages), ""


def _graph_input(run_input: RunAgentInput) -> tuple[str, list[ModelMessage]]:
    latest_user_index, prompt = _latest_user_message(run_input)
    previous_messages = run_input.messages[:latest_user_index]
    return prompt, AGUIAdapter.load_messages(previous_messages)


def _latest_assistant_text(messages: Sequence[Message]) -> str:
    for message in reversed(messages):
        if isinstance(message, AssistantMessage) and message.content:
            return message.content

    return ""

The endpoint itself builds the run input, runs the graph, and streams AG-UI events:

@router.post("/demo-graph")
async def demo_graph_endpoint(request: Request) -> Response:
    run_input = AGUIAdapter.build_run_input(await request.body())
    encoder = EventEncoder(request.headers.get("accept") or "")

    async def events() -> AsyncIterator[str]:
        yield encoder.encode(
            RunStartedEvent(
                thread_id=run_input.thread_id,
                run_id=run_input.run_id,
                input=run_input,
            )
        )

        message_id = str(uuid4())

        try:
            prompt, previous_messages = _graph_input(run_input)
            result = await demo_graph.run(
                state=AppState(messages=previous_messages),
                inputs=DemoGraphInput(prompt=prompt),
            )
            messages = AGUIAdapter.dump_messages(result.messages)
            response_text = _latest_assistant_text(messages)

            yield encoder.encode(
                TextMessageStartEvent(message_id=message_id, role="assistant")
            )

            if response_text:
                yield encoder.encode(
                    TextMessageContentEvent(
                        message_id=message_id,
                        delta=response_text,
                    )
                )

            yield encoder.encode(TextMessageEndEvent(message_id=message_id))
            yield encoder.encode(
                RunFinishedEvent(
                    thread_id=run_input.thread_id,
                    run_id=run_input.run_id,
                    result={"messages": messages},
                )
            )
        except Exception as exc:
            yield encoder.encode(
                RunErrorEvent(message=str(exc), code=exc.__class__.__name__)
            )

    return StreamingResponse(events(), media_type=encoder.get_content_type())

This isn’t a lot of code, but it is the most important code in the repo.

The frontend still receives an AG-UI stream. The graph still gets clean Python inputs and state. The router is the translation layer between those two worlds.

Notice that the graph result isn’t AG-UI-shaped. The router converts result.messages with AGUIAdapter.dump_messages(...) and only then returns {"messages": messages} in the AG-UI RunFinishedEvent.

FastAPI app

Create backend/main.py:

from dotenv import load_dotenv

load_dotenv()

from fastapi import FastAPI

from routers.graph_router import router as graph_router

app = FastAPI()
app.include_router(graph_router, prefix="/graph")


@app.get("/")
async def root():
    return {"message": "Hello, I'm the main API!"}


@app.get("/health")
async def health():
    return {
        "api": "main",
        "status": "healthy",
    }

From the backend directory, run:

uv run uvicorn main:app --reload

Check these URLs in your browser:

• http://127.0.0.1:8000/
• http://127.0.0.1:8000/health
• http://127.0.0.1:8000/graph
• http://127.0.0.1:8000/graph/health

The AG-UI endpoint itself is POST http://127.0.0.1:8000/graph/demo-graph, so you will call it through CopilotKit rather than opening it directly in the browser.

Next.js UI

This section assumes you have pnpm installed.

Open a new terminal in the project root and run:

pnpm create next-app@latest

Use ui as the project name and accept the recommended defaults.

Change into the UI project:

cd ui

Add the CopilotKit and AG-UI dependencies:

pnpm add @copilotkit/react-ui @copilotkit/react-core @copilotkit/runtime @ag-ui/client openai

Run Next.js:

pnpm dev

Navigate to http://localhost:3000.

For a small local demo, hardcoding http://localhost:8000/graph/demo-graph in the API route works. I prefer a tiny configuration layer because the backend URL usually changes between local development, preview deployments, and production.

Create ui/.env.example:

COPILOTKIT_GRAPH_BASE_URL=http://localhost:8000/graph/
COPILOTKIT_DEMO_GRAPH_NAME=demo-graph

For local development, copy that to ui/.env.local:

cp .env.example .env.local

These variables do not need the NEXT_PUBLIC_ prefix because they are only read by the server-side API route.

Create ui/lib/copilot-agent-ids.ts:

export const DEMO_GRAPH_ID = "demoGraph";

Create ui/lib/copilot-agent-config.ts:

const DEFAULT_GRAPH_BASE_URL = "http://localhost:8000/graph/";
const DEFAULT_DEMO_GRAPH_NAME = "demo-graph";

function trimSlashes(value: string) {
  return value.replace(/^\/+|\/+$/g, "");
}

function withTrailingSlash(value: string) {
  return value.endsWith("/") ? value : `${value}/`;
}

export function getDemoGraphUrl() {
  const graphBaseUrl =
    process.env.COPILOTKIT_GRAPH_BASE_URL ?? DEFAULT_GRAPH_BASE_URL;
  const demoGraphName =
    process.env.COPILOTKIT_DEMO_GRAPH_NAME ?? DEFAULT_DEMO_GRAPH_NAME;

  return new URL(trimSlashes(demoGraphName), withTrailingSlash(graphBaseUrl))
    .toString();
}

This keeps three related values distinct:

• COPILOTKIT_GRAPH_BASE_URL is the FastAPI router base URL.
• COPILOTKIT_DEMO_GRAPH_NAME is the backend route path for this graph.
• DEMO_GRAPH_ID is the CopilotKit runtime id used by the frontend.

CopilotKit runtime route

Create ui/app/api/copilotkit/route.ts:

import {
  CopilotRuntime,
  ExperimentalEmptyAdapter,
  copilotRuntimeNextJSAppRouterEndpoint,
} from "@copilotkit/runtime";
import { HttpAgent } from "@ag-ui/client";
import { NextRequest } from "next/server";
import { getDemoGraphUrl } from "@/lib/copilot-agent-config";
import { DEMO_GRAPH_ID } from "@/lib/copilot-agent-ids";

const serviceAdapter = new ExperimentalEmptyAdapter();

const runtime = new CopilotRuntime({
  agents: {
    [DEMO_GRAPH_ID]: new HttpAgent({ url: getDemoGraphUrl() }),
  },
});

export const POST = async (req: NextRequest) => {
  const { handleRequest } = copilotRuntimeNextJSAppRouterEndpoint({
    runtime,
    serviceAdapter,
    endpoint: "/api/copilotkit",
  });

  return handleRequest(req);
};

This route registers the FastAPI graph endpoint as a CopilotKit agent.

The frontend does not need to know that the backend endpoint runs a graph. It only needs the CopilotKit runtime URL and the agent id.

Assistants route group

This structure is optional, but it keeps CopilotKit-specific pages and layout code separate from the rest of the app.

Create the route group folder:

mkdir -p "ui/app/(assistants)/demo-graph"

Next.js route groups do not affect the URL path, so a page at ui/app/(assistants)/demo-graph/page.tsx is served at /demo-graph, not /assistants/demo-graph.

Create ui/app/(assistants)/layout.tsx:

import {
  CopilotKitProvider,
} from "@copilotkit/react-core/v2";
import "@copilotkit/react-core/v2/styles.css";

export default function AssistantsLayout({
  children,
}: {
  children: React.ReactNode;
}) {
  return (
    <CopilotKitProvider runtimeUrl="/api/copilotkit">
      {children}
    </CopilotKitProvider>
  );
}

Create ui/app/(assistants)/demo-graph/page.tsx:

"use client";

import { CopilotSidebar } from "@copilotkit/react-core/v2";
import { DEMO_GRAPH_ID } from "@/lib/copilot-agent-ids";

export default function DemoGraphPage() {
  return (
    <div className="flex h-full min-h-0 flex-col gap-4 px-4 pb-6">
      <CopilotSidebar agentId={DEMO_GRAPH_ID} />
    </div>
  );
}

The agentId must match the key registered in CopilotRuntime.

That detail matters more as the app grows. You can add more graph-backed assistants later without mixing backend route names, frontend page paths, and CopilotKit runtime ids together.

Try it

With both servers running, navigate to:

http://localhost:3000/demo-graph

You should now have a CopilotKit sidebar connected to a Pydantic Graph workflow running behind FastAPI.

Where this gets useful

A graph gives the assistant room to become a workflow.

The graph is currently very small, but the structure is already ready for richer orchestration. You can add routing, tool steps, approval stages, retrieval, validation, or more advanced state transitions without changing the CopilotKit surface area.

The browser still talks to one assistant. The backend can gradually evolve into a much more capable workflow behind that interface.

Conclusion

This gives you a minimal assistant built with Pydantic Graph, Pydantic AI, FastAPI, AG-UI, and CopilotKit.

The frontend shape is almost identical to the earlier plain-agent version.

The backend shape is the interesting part:

• A Pydantic Graph stores and updates conversation state.
• A chat step delegates model work to a Pydantic AI Agent.
• The graph receives DemoGraphInput and returns DemoGraphOutput.
• The FastAPI router bridges AG-UI requests into graph-native inputs.
• The router converts graph-native message history back into AG-UI messages and emits AG-UI events so CopilotKit can render the assistant response.

Right now, the graph route is a small amount of custom protocol glue.

That isn’t a bad tradeoff for a demo. It makes the integration explicit, and it gives you a clear place to improve later if Pydantic Graph gets a more direct AG-UI adapter path.