Building a CopilotKit Runtime Equivalent with Phoenix GenServer and AG-UI Protocol

To create a CopilotKit-style runtime in Phoenix, you’ll integrate GenServer for state management, PubSub for real-time communication, and AG-UI events for standardized agent interactions. This approach lets you use CopilotKit’s React components and hooks on the frontend while connecting to your Phoenix-based AG-UI backend.

Core Phoenix Components Architecture

1. GenServer as the Runtime Coordinator

Replace CopilotKit’s Node.js runtime with a GenServer that manages agent conversations and AG-UI events:

defmodule MyAppWeb.AgentRuntime do
  use GenServer
  
  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end
  
  def init(_opts) do
    # Subscribe to agent events from your AI backend
    Phoenix.PubSub.subscribe(MyApp.PubSub, "agent_events")
    
    {:ok, %{
      conversations: %{},      # Active conversation threads
      clients: %{},           # SSE connections mapped to threads  
      agent_states: %{}       # Current agent execution states
    }}
  end
  
  # Handle AG-UI events from your agent backend
  def handle_info({:agent_event, thread_id, event}, state) do
    # Transform to AG-UI format
    ag_ui_event = transform_to_ag_ui(event)
    
    # Broadcast to subscribed clients
    Phoenix.PubSub.broadcast(
      MyApp.PubSub, 
      "ag_ui:#{thread_id}", 
      {:ag_ui_event, ag_ui_event}
    )
    
    # Update conversation state
    updated_conversations = update_conversation_state(
      state.conversations, 
      thread_id, 
      ag_ui_event
    )
    
    {:noreply, %{state | conversations: updated_conversations}}
  end
end

2. Controller for SSE Event Streaming

Create a Phoenix controller that handles Server-Sent Events, mimicking CopilotKit’s runtime endpoint:

defmodule MyAppWeb.AgentController do
  use MyAppWeb, :controller
  
  def stream_events(conn, %{"thread_id" => thread_id}) do
    conn
    |> put_resp_header("cache-control", "no-cache")
    |> put_resp_header("access-control-allow-origin", "*") 
    |> put_resp_content_type("text/event-stream")
    |> send_chunked(200)
    |> register_client_and_stream(thread_id)
  end
  
  defp register_client_and_stream(conn, thread_id) do
    # Subscribe to AG-UI events for this thread
    Phoenix.PubSub.subscribe(MyApp.PubSub, "ag_ui:#{thread_id}")
    
    # Register this connection with the runtime
    GenServer.cast(MyAppWeb.AgentRuntime, {:register_client, self(), thread_id})
    
    # Start streaming loop
    stream_loop(conn)
  end
  
  defp stream_loop(conn) do
    receive do
      {:ag_ui_event, event} ->
        # Send AG-UI formatted event to client
        case chunk(conn, format_sse_data(event)) do
          {:ok, conn} -> stream_loop(conn)
          {:error, _} -> conn  # Client disconnected
        end
        
      :close -> 
        conn
    after
      30_000 -> 
        # Send heartbeat
        case chunk(conn, format_sse_heartbeat()) do
          {:ok, conn} -> stream_loop(conn)
          {:error, _} -> conn
        end
    end
  end
  
  defp format_sse_data(event) do
    data = Jason.encode!(event)
    "data: #{data}\n\n"
  end
end

3. Router Configuration

Set up routes similar to CopilotKit’s /api/copilotkit endpoint:

# router.ex
defmodule MyAppWeb.Router do
  use MyAppWeb, :router
  
  pipeline :api do
    plug :accepts, ["json"]
  end
  
  scope "/api", MyAppWeb do
    pipe_through :api
    
    # Main agent runtime endpoint (like CopilotKit's runtime)
    get "/agent/:thread_id/stream", AgentController, :stream_events
    post "/agent/:thread_id/message", AgentController, :send_message
    get "/agent/:thread_id/state", AgentController, :get_state
  end
end

CopilotKit React Integration

Required React Dependencies

Install the core CopilotKit React packages:

npm install @copilotkit/react-core @copilotkit/react-ui

Package Breakdown:

@copilotkit/react-core: Context providers, hooks, and core logic
@copilotkit/react-ui: Pre-built UI components like CopilotSidebar, CopilotChat

React App Setup with Phoenix Backend

CopilotKit Provider Configuration: Connect your React app to the Phoenix AG-UI runtime instead of the default Node.js runtime:

// app/layout.tsx (Next.js) or index.js (Create React App)
import { CopilotKit } from '@copilotkit/react-core';
import '@copilotkit/react-ui/styles.css';

export default function RootLayout({ children }) {
  return (
    <html lang="en">
      <body>
        <CopilotKit 
          runtimeUrl="http://localhost:4000/api/agent/stream"  // Phoenix endpoint
          agent="phoenixAgent"
        >
          {children}
        </CopilotKit>
      </body>
    </html>
  );
}

Chat Interface Components: Use CopilotKit’s React UI components that automatically work with AG-UI events:

// components/ChatInterface.jsx
import { CopilotSidebar, CopilotChat } from '@copilotkit/react-ui';

export default function ChatInterface() {
  return (
    <div className="app-layout">
      <main className="main-content">
        {/* Your existing app content */}
        <YourMainApp />
      </main>
      
      {/* CopilotKit sidebar - connects to Phoenix via AG-UI */}
      <CopilotSidebar
        defaultOpen={false}
        labels={{
          title: "Phoenix AI Assistant",
          initial: "Hello! I'm powered by Phoenix GenServer + AG-UI protocol 🚀"
        }}
        instructions="You are an AI assistant powered by Phoenix and Elixir GenServers."
      />
    </div>
  );
}

React Hooks for State Management

useCopilotReadable Hook: Share React application state with your Phoenix backend:

import { useCopilotReadable } from '@copilotkit/react-core';

function TaskManager() {
  const [tasks, setTasks] = useState([]);
  const [currentProject, setCurrentProject] = useState(null);
  
  // Make app state available to Phoenix GenServer
  useCopilotReadable({
    description: "Current tasks and project information",
    value: {
      tasks: tasks,
      currentProject: currentProject,
      taskCount: tasks.length
    }
  });
  
  return (
    <div>
      {/* Your task management UI */}
    </div>
  );
}

useCopilotAction Hook: Allow Phoenix GenServer to trigger React app actions:

import { useCopilotAction } from '@copilotkit/react-core';

function ProjectDashboard() {
  const [projects, setProjects] = useState([]);
  
  // Register actions that Phoenix can trigger
  useCopilotAction({
    name: "createProject",
    description: "Create a new project with the given details",
    parameters: [
      {
        name: "projectName",
        type: "string",
        description: "The name of the new project"
      },
      {
        name: "description", 
        type: "string",
        description: "Project description"
      }
    ],
    handler: ({ projectName, description }) => {
      const newProject = {
        id: Date.now(),
        name: projectName,
        description: description,
        createdAt: new Date()
      };
      setProjects(prev => [...prev, newProject]);
    }
  });
  
  return <ProjectList projects={projects} />;
}

Phoenix Controller Modifications for CopilotKit

Update your Phoenix AG-UI controller to handle CopilotKit-specific headers and events:

defmodule MyAppWeb.AgentController do
  use MyAppWeb, :controller
  
  def stream_events(conn, params) do
    thread_id = params["thread_id"] || "default"
    
    conn
    |> put_resp_header("cache-control", "no-cache")
    |> put_resp_header("access-control-allow-origin", "http://localhost:3000")  # React dev server
    |> put_resp_header("access-control-allow-headers", "content-type, authorization, x-copilotkit-*")
    |> put_resp_content_type("text/event-stream")
    |> send_chunked(200)
    |> register_copilot_client(thread_id)
    |> stream_ag_ui_events()
  end
  
  defp register_copilot_client(conn, thread_id) do
    # Extract CopilotKit specific info from headers
    copilot_agent = get_req_header(conn, "x-copilotkit-agent") |> List.first()
    
    # Subscribe to AG-UI events for this thread
    Phoenix.PubSub.subscribe(MyApp.PubSub, "ag_ui:#{thread_id}")
    
    # Register with AgentRuntime
    GenServer.cast(MyAppWeb.AgentRuntime, {
      :register_copilot_client, 
      self(), 
      thread_id, 
      %{agent: copilot_agent}
    })
    
    conn
  end
end

Custom React Components with AG-UI Events

Create custom React components that respond to specific AG-UI events from Phoenix:

import { useCopilotChat } from '@copilotkit/react-core';
import { useEffect, useState } from 'react';

function CustomAgentInterface() {
  const [agentState, setAgentState] = useState('idle');
  const [toolCalls, setToolCalls] = useState([]);
  
  const { 
    messages, 
    appendMessage, 
    isLoading,
    reload 
  } = useCopilotChat({
    id: "custom-phoenix-chat"
  });
  
  // Listen for specific AG-UI events from Phoenix
  useEffect(() => {
    const eventSource = new EventSource('/api/agent/stream');
    
    eventSource.onmessage = (event) => {
      const agUiEvent = JSON.parse(event.data);
      
      switch (agUiEvent.type) {
        case 'RunStarted':
          setAgentState('running');
          break;
          
        case 'ToolCallStart':
          setToolCalls(prev => [...prev, {
            id: agUiEvent.toolCallId,
            name: agUiEvent.toolCallName,
            status: 'starting'
          }]);
          break;
          
        case 'RunFinished':
          setAgentState('completed');
          break;
          
        case 'StateSnapshot':
          // Handle state updates from Phoenix GenServer
          console.log('Phoenix state update:', agUiEvent.snapshot);
          break;
      }
    };
    
    return () => eventSource.close();
  }, []);
  
  return (
    <div className="custom-agent-interface">
      <div className="agent-status">
        Status: {agentState}
      </div>
      
      <div className="tool-calls">
        {toolCalls.map(tool => (
          <div key={tool.id} className="tool-call">
            {tool.name}: {tool.status}
          </div>
        ))}
      </div>
      
      <div className="messages">
        {messages.map(msg => (
          <div key={msg.id} className="message">
            {msg.content}
          </div>
        ))}
      </div>
    </div>
  );
}

AG-UI Event Transformation Pipeline

Create utilities to handle the 16 AG-UI event types:

defmodule MyApp.AgUiTransformer do
  def transform_to_ag_ui(agent_event) do
    case agent_event do
      %{type: :run_started, run_id: run_id, thread_id: thread_id} ->
        %{
          type: "RunStarted", 
          runId: run_id,
          threadId: thread_id,
          timestamp: DateTime.utc_now()
        }
        
      %{type: :text_start, message_id: msg_id, role: role} ->
        %{
          type: "TextMessageStart",
          messageId: msg_id,
          role: role
        }
        
      %{type: :text_delta, message_id: msg_id, content: delta} ->
        %{
          type: "TextMessageContent", 
          messageId: msg_id,
          delta: delta
        }
        
      %{type: :tool_call_start, tool_call_id: id, name: name} ->
        %{
          type: "ToolCallStart",
          toolCallId: id,
          toolCallName: name
        }
        
      # Handle all 16 AG-UI event types...
      _ -> 
        %{type: "Raw", value: agent_event}
    end
  end
  
  def to_copilot_compatible(ag_ui_event) do
    # CopilotKit expects specific AG-UI event structure
    case ag_ui_event do
      %{type: "TextMessageContent", messageId: msg_id, delta: content} ->
        %{
          type: "TextMessageContent",
          messageId: msg_id, 
          delta: content,
          # CopilotKit specific fields
          timestamp: DateTime.utc_now() |> DateTime.to_iso8601()
        }
        
      %{type: "ToolCallStart", toolCallId: id, toolCallName: name} ->
        %{
          type: "ToolCallStart",
          toolCallId: id,
          toolCallName: name,
          parentMessageId: ag_ui_event[:parent_message_id]
        }
        
      _ -> ag_ui_event
    end
  end
end

Agent Backend Integration

Connect your AI agent (LangGraph, AG2, etc.) to emit events that the GenServer processes:

defmodule MyApp.AgentConnector do
  def handle_agent_response(agent_response, thread_id) do
    # Convert your agent's output to internal events
    events = parse_agent_events(agent_response)
    
    # Send each event to the runtime GenServer
    Enum.each(events, fn event ->
      Phoenix.PubSub.broadcast(
        MyApp.PubSub, 
        "agent_events", 
        {:agent_event, thread_id, event}
      )
    end)
  end
  
  defp parse_agent_events(response) do
    # Transform your agent's format to internal events
    # This depends on your specific agent framework
    [
      %{type: :run_started, run_id: UUID.uuid4(), thread_id: response.thread_id},
      %{type: :text_start, message_id: UUID.uuid4(), role: "assistant"},
      %{type: :text_delta, message_id: response.message_id, content: response.content}
    ]
  end
end

Application Supervision Tree

Add the runtime to your application supervision tree:

defmodule MyApp.Application do
  use Application
  
  def start(_type, _args) do
    children = [
      # Standard Phoenix components
      {Phoenix.PubSub, name: MyApp.PubSub},
      MyAppWeb.Endpoint,
      
      # Your AG-UI runtime
      MyAppWeb.AgentRuntime,
      
      # Optional: Dynamic supervisor for agent processes
      {DynamicSupervisor, name: MyApp.AgentSupervisor}
    ]
    
    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

Benefits of This Architecture

CopilotKit Advantages:

Production-ready UI components with built-in streaming, loading states, error handling
Type-safe React hooks for actions and state management
Customizable theming and component styling
Built-in state synchronization between React and backend

Phoenix + GenServer Benefits:

Superior concurrency handling thousands of simultaneous conversations
Fault tolerance with automatic process recovery
Real-time performance with microsecond response times
Distributed capabilities across multiple servers

Production Considerations

Monitoring and Observability:

def handle_info({:agent_event, thread_id, event}, state) do
  # Add telemetry for monitoring
  :telemetry.execute([:agent, :event, :processed], %{count: 1}, %{
    thread_id: thread_id,
    event_type: event.type
  })
  
  # Process event...
end

Memory Management:

Use ETS tables for large shared state
Implement conversation cleanup for inactive threads
Monitor GenServer mailbox size to prevent memory leaks

This Phoenix-based architecture provides the same capabilities as CopilotKit’s runtime while leveraging Elixir’s strengths in concurrency, fault tolerance, and real-time communication. The result gives you the developer experience of CopilotKit’s React ecosystem while leveraging the performance and reliability of Phoenix GenServers for your AI agent backend.

Jeffrey Hicks