Predictive state updates

This example demonstrates predictive state updates in the CopilotKit Feature Viewer.

What is this?#

Microsoft Agent Framework agents can stream state updates through AG-UI as tool arguments are generated by the LLM. CopilotKit surfaces these updates in the UI, enabling optimistic, real-time rendering. We call these predictive state updates.

When should I use this?#

Use predictive state updates when you want to:

Keep users engaged during long-running operations
Show step-by-step progress
Build trust by exposing what the agent is doing now, not only at the end
Enable agent steering (users can intervene if needed)

Source of truth

When the tool completes, the agent emits a final state snapshot. Any predictive updates should be reflected in that final state or they will be overwritten.

Implementation#

Define the state#

We will define an observed_steps array that is updated while the agent performs long-running tasks.

agent/Program.cs (excerpt)

using System.Text.Json.Serialization;
public class AgentStateSnapshot
{
    [JsonPropertyName("observed_steps")]
    public List<string> ObservedSteps { get; set; } = new();
}

agent/src/agent.py (excerpt)

STATE_SCHEMA: dict[str, object] = {
    "observed_steps": {
        "type": "array",
        "items": {"type": "string"},
        "description": "Array of completed steps"
    }
}

Map tool calls to state#

Configure AG-UI state management to map the step_progress tool arguments to observed_steps. The .NET adapter emits a state snapshot when it receives the completed tool call. The Python adapter can also stream partial tool arguments as state deltas.

agent/Program.cs (excerpt)

using System.ComponentModel;
using System.Text.Json;
using AGUI.Abstractions;
using AGUI.Server;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;
using Microsoft.Extensions.AI;
using OpenAI;
using OpenAI.Chat;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddAGUIServer();
var app = builder.Build();

string openAiApiKey = builder.Configuration["OPENAI_API_KEY"]
    ?? throw new InvalidOperationException("Set OPENAI_API_KEY");

// Define a tool the LLM may call to report its progress
[Description("Report current step progress.")]
static string StepProgress([Description("Steps completed so far")] string[] steps)
    => "Progress received.";

AITool stepProgress = AIFunctionFactory.Create(StepProgress, name: "step_progress");
var agent = new OpenAIClient(openAiApiKey)
    .GetChatClient("gpt-5.4-mini")
    .AsAIAgent(
        name: "AGUIAssistant",
        instructions: "You are a helpful assistant that may call the 'step_progress' tool to report intermediate steps.",
        tools: [stepProgress]);

// Map the completed tool arguments to a shared-state snapshot
AGUIStreamOptions streamOptions = new AGUIStreamOptions()
    .MapCall("step_progress", call =>
    {
        if (call.Arguments?.TryGetValue("steps", out object? steps) is not true)
        {
            return [];
        }

        JsonElement snapshot = JsonSerializer.SerializeToElement(
            new { observed_steps = steps });
        return [new StateSnapshotEvent { Snapshot = snapshot }];
    });

app.MapAGUIServer("/", agent).WithMetadata(streamOptions);
await app.RunAsync();

agent/src/agent.py (excerpt)

from __future__ import annotations
from typing import Annotated
from agent_framework import Agent, SupportsChatGetResponse, tool
from agent_framework.ag_ui import AgentFrameworkAgent
from pydantic import Field

# 1) Define state schema for AG-UI
STATE_SCHEMA: dict[str, object] = {
    "observed_steps": {
        "type": "array",
        "items": {"type": "string"},
        "description": "Array of completed steps"
    }
}

# 2) Predictive state mapping: observed_steps <- step_progress.steps
PREDICT_STATE_CONFIG: dict[str, dict[str, str]] = {
    "observed_steps": {
        "tool": "step_progress",
        "tool_argument": "steps",
    }
}

# 3) Tool that the LLM will call with step updates
@tool
def step_progress(
    steps: Annotated[list[str], Field(description="Steps completed so far")]
) -> str:
    return "Progress received."

def create_agent(chat_client: SupportsChatGetResponse) -> AgentFrameworkAgent:
    base = Agent(
        name="sample_agent",
        instructions="You are a task performer. Report progress using step_progress.",
        client=chat_client,
        tools=[step_progress],
    )
    return AgentFrameworkAgent(
        agent=base,
        name="CopilotKitMicrosoftAgentFrameworkAgent",
        description="Agent with predictive state updates for observed steps.",
        state_schema=STATE_SCHEMA,
        predict_state_config=PREDICT_STATE_CONFIG,
        require_confirmation=False,
    )

On .NET, MapCall maps the completed FunctionCallContent to a state snapshot. Progressive .NET tool-argument updates require a provider-specific extractor registered with MapStreamingToolCallArguments.

Observe state on the client#

Add a state renderer to observe the observed_steps updates as they arrive.

ui/app/page.tsx

"use client";

import { useAgent } from "@copilotkit/react-core/v2";

type AgentState = {
  observed_steps: string[];
};

export default function Page() {
  // Access the latest shared state
  const { agent } = useAgent({ agentId: "sample_agent" });

  // Read the agent's shared state
  const progress =
    agent.state?.observed_steps?.length ? (
      <div>
        <h3>Current Progress:</h3>
        <ul>
          {agent.state.observed_steps.map((step, i) => (
            <li key={i}>{step}</li>
          ))}
        </ul>
      </div>
    ) : null;

  return <div>{progress}</div>;
}

Give it a try!#

Ask the agent to perform a multi-step task (e.g., “write a short outline and report progress each step”). You’ll see observed_steps update in real time as the tool arguments stream in.