Proxy an AG-UI Event Stream

An agent that drives a user interface can stream text, open a form, ask the user to confirm a refund, or navigate the page. With the AG-UI proxy in your server, each of those events passes a capability check before your transport forwards it, and every decision leaves a signed receipt.

AG-UI, the Agent User Interaction Protocol, streams events such as text-message content and state updates; its event reference defines that wire format. Your host normalizes each upstream event into Chio’s AgUiEvent, a separate envelope for classifying UI effects, and keeps ownership of the connection.

1. Add the proxy to your Rust server

The proxy is a Rust library: chio-ag-ui-proxy, configured through AgUiProxyConfig. Add it to the server that holds your agent’s event stream:

cargo add chio-ag-ui-proxy --git https://github.com/backbay-labs/chio

You need an agent that emits AG-UI events and a client that subscribes over SSE or WebSocket. The wire format the client sees stays the same; the proxy adds a classification, a capability check, and a receipt per event.

2. Classify each event the way the server will

The UI stream reaches the user directly. A wrongly issued prompt can phish the user, a fabricated notification can trigger a support escalation, and unchecked navigation can send the user to an attacker’s page. Classification is how the proxy tells those effects apart, and it derives the classification itself:

You should see two consequences in the table. A custom event type has no server classification, and every classification error becomes a block, so a custom event never rides through. A lifecycle event carries its verb under action, lifecycle, or event in the payload; without one, derivation fails and the event is blocked before the capability is read. Add alert to restricted_classifications to gate notifications.

pub(super) fn derive_server_classification(
    event: &AgUiEvent,
) -> Result<EventClassification, String> {
    match &event.event_type {
        EventType::TextStream => Ok(EventClassification::Display),
        EventType::StateUpdate => Ok(EventClassification::Mutate),
        EventType::Navigation => Ok(EventClassification::Navigate),
        EventType::Lifecycle => derive_lifecycle_classification(&event.payload),
        EventType::FormAction => Ok(EventClassification::Submit),
        EventType::Notification | EventType::Error => Ok(EventClassification::Alert),
        EventType::Custom(name) => Err(format!(
            "custom AG-UI event type cannot be server-classified: {name}"
        )),
    }
}

fn derive_lifecycle_classification(
    payload: &serde_json::Value,
) -> Result<EventClassification, String> {
    let action = payload
        .get("action")
        .or_else(|| payload.get("lifecycle"))
        .or_else(|| payload.get("event"))
        .and_then(serde_json::Value::as_str)
        .map(str::to_ascii_lowercase);

    match action.as_deref() {
        Some("create" | "created" | "mount" | "mounted" | "open" | "opened") => {
            Ok(EventClassification::Create)
        }
        Some("destroy" | "destroyed" | "unmount" | "unmounted" | "close" | "closed") => {
            Ok(EventClassification::Destroy)
        }
        Some("update" | "updated" | "change" | "changed") => Ok(EventClassification::Mutate),
        Some(other) => Err(format!(
            "lifecycle AG-UI event action is not classifiable: {other}"
        )),
        None => Err("lifecycle AG-UI event missing classifiable action".to_string()),
    }
}

{
  "event_id": "evt_01HZ8A1",
  "timestamp": 1744993921,
  "agent_id": "agent-support",
  "session_id": "sess_01HZ...",
  "event_type": "text_stream",
  "target": {
    "component_type": "chat-window",
    "component_id": "main"
  },
  "classification": "display",
  "payload": {
    "text": "I'll check your order status now."
  }
}

{
  "event_id": "evt_01HZ8A2",
  "timestamp": 1744993923,
  "agent_id": "agent-support",
  "session_id": "sess_01HZ...",
  "event_type": "lifecycle",
  "target": {
    "component_type": "form",
    "component_id": "refund-request"
  },
  "classification": "create",
  "payload": {
    "action": "create",
    "fields": ["order_id", "reason", "amount"]
  }
}

{
  "event_id": "evt_01HZ8A3",
  "timestamp": 1744993928,
  "agent_id": "agent-support",
  "session_id": "sess_01HZ...",
  "event_type": "form_action",
  "target": {
    "component_type": "modal",
    "component_id": "confirm-refund"
  },
  "classification": "submit",
  "payload": {
    "prompt": "Issue $42.50 refund to order #1024?"
  }
}

3. Evaluate before each SSE frame

Your server owns the text/event-stream response and the open connection. Build the proxy from an AgUiProxyConfig, record the connection as a Transport of kind Sse, and evaluate each event before writing its frame:

use chio_ag_ui_proxy::{
    AgUiProxy, AgUiProxyConfig, EventClassification, ProxyDecision, Transport,
    TransportKind,
};

let config = AgUiProxyConfig {
    // Display rides through without a token; the five restricted
    // classifications below require a verified capability in the session.
    allow_display_without_capability: true,
    restricted_classifications: vec![
        EventClassification::Mutate,
        EventClassification::Navigate,
        EventClassification::Create,
        EventClassification::Destroy,
        EventClassification::Submit,
    ],
    max_events_per_second: 200,
    trusted_issuers: issuer_keys, // capability-issuer public keys
    ..AgUiProxyConfig::default()
};

let proxy = AgUiProxy::try_new(config, kernel_keypair)?;

// Your transport code owns the connection. Record it once, then evaluate
// every event before writing an SSE frame.
let mut transport =
    Transport::new(TransportKind::Sse, "conn-01".into(), "agent-support".into());

let (decision, receipt) = proxy.evaluate(&event, capability.as_ref(), &mut transport)?;
persist(&receipt)?; // your fallible receipt journal; retain before forwarding
match decision {
    ProxyDecision::Forward => write_sse_frame(&upstream_event), // original wire message
    ProxyDecision::Block { reason } => tracing::warn!(%reason, "ag-ui event blocked"),
}

Here event is your host’s normalized record for upstream_event. Your host supplies the authenticated session, trusted issuer keys, signing key, fallible journal, and frame writer. Resolve the capability from that authenticated context, because an event payload cannot assign its own authority, and forward the same upstream message whose normalized effect you checked.

The browser subscribes to your server’s SSE endpoint, unchanged:

const es = new EventSource("/ag-ui/stream?session=" + sessionId);

es.addEventListener("message", (ev) => {
  const event = JSON.parse(ev.data);
  // The host forwards the original event after evaluating its normalized form.
  render(event);
});

es.addEventListener("error", (err) => {
  console.warn("ag-ui stream closed", err);
});

You should see a blocked event stay off the stream. Treat an evaluation or journal error as a refusal to forward. An invalid event identity or a signing failure returns an error with no receipt, so retain those errors in your host’s operational log.

4. Evaluate both directions on a WebSocket

Outbound events are re-derived and receipted as with SSE, and inbound client-to-agent messages run through the same evaluate path, so a compromised browser can neither claim capabilities the session does not carry nor fabricate state that skips the capability check. Only the transport kind changes:

let mut transport =
    Transport::new(TransportKind::WebSocket, "conn-01".into(), "agent-support".into());

// Outbound (agent -> client) and inbound (client -> agent) events both flow
// through evaluate before you relay them.
let (decision, receipt) = proxy.evaluate(&event, capability.as_ref(), &mut transport)?;

const ws = new WebSocket("wss://example.app/ag-ui?session=" + sessionId);

ws.addEventListener("message", (ev) => {
  const event = JSON.parse(ev.data);
  render(event);
});

// The shape is whatever your server deserializes into
// chio_ag_ui_proxy::event::AgUiEvent.
function sendToAgent(partial: unknown) {
  ws.send(JSON.stringify(partial));
}

Make the proxy the only path between the client and the agent. A separate WebSocket endpoint on the agent, kept for convenience or debugging, lets a client skip the capability check, so close it or place the agent behind a network boundary that routes all traffic through your server.

5. Prove the gates, then read the receipt

evaluate decides from the event and the capability alone, so the crate’s own tests pin its gates. The four display-event tests are the four outcomes a display event can have:

$ cargo test --release -p chio-ag-ui-proxy --lib \
    -- --test-threads 1 proxy::tests::display_event

running 4 tests
test proxy::tests::display_event_allowed_when_configured ... ok
test proxy::tests::display_event_blocked_without_capability_by_default ... ok
test proxy::tests::display_event_rejects_untrusted_capability_by_default ... ok
test proxy::tests::display_event_requires_display_scope_when_capability_supplied ... ok

test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 31 filtered out; finished in 0.00s

You should see a display event forwarded only when the configuration allows it and no capability is attached. With no capability and no allowance, with an untrusted capability, or with a trusted capability that lacks display scope, it is blocked: attaching a capability never loosens the gate. The re-derivation is pinned too, so a caller cannot label a state_update as display to slip past the restricted set:

$ cargo test --release -p chio-ag-ui-proxy --lib \
    -- --test-threads 1 classification

running 1 test
test proxy::tests::state_update_cannot_downgrade_classification_to_display ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 34 filtered out; finished in 0.00s

Mirror these cases in your own suite against your AgUiProxyConfig. Each evaluation returns an AgUiReceipt with schema chio.ag-ui-receipt.v1; this one records the refund confirmation above, allowed over a WebSocket:

{
  "id": "agui-evt_01HZ8A3",
  "timestamp": 1744993928,
  "event_id": "evt_01HZ8A3",
  "agent_id": "agent-support",
  "session_id": "sess_01HZ...",
  "capability_id": "cap-ui-confirm-01HZ...",
  "event_type": "form_action",
  "target": {
    "component_type": "modal",
    "component_id": "confirm-refund"
  },
  "classification": "submit",
  "transport": "web_socket",
  "allowed": true,
  "payload_hash": "<sha256 of the canonical payload>",
  "kernel_key": "<64 hex>",
  "signature": "<128 hex>"
}

You should see the payload as a hash, never the payload itself, and an Ed25519 signature over the whole body. classification, event_type, and target are the axes an auditor queries, for example how many submit events reached a modal in the last hour.

pub struct AgUiReceipt {
    /// Unique receipt ID.
    pub id: String,
    /// Unix timestamp (seconds) when the receipt was created.
    pub timestamp: u64,
    /// The event ID that was evaluated.
    pub event_id: String,
    /// Agent that produced the event.
    pub agent_id: String,
    /// Session, if bound.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Capability ID used for authorization.
    pub capability_id: String,
    /// Type of event.
    pub event_type: EventType,
    /// Target component.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target: Option<TargetComponent>,
    /// Action classification.
    pub classification: EventClassification,
    /// Transport used.
    pub transport: TransportKind,
    /// Whether the event was allowed or denied.
    pub allowed: bool,
    /// Denial reason, if applicable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub denial_reason: Option<String>,
    /// SHA-256 hash of the event payload.
    pub payload_hash: String,
    /// Kernel public key for verification.
    pub kernel_key: PublicKey,
    /// Signature over the receipt body.
    pub signature: Signature,
}

fn decide(&self, event: &AgUiEvent, capability: Option<&CapabilityToken>) -> ProxyDecision {
    let requires_capability = self
        .config
        .restricted_classifications
        .contains(&event.classification);

    if let Some(capability) = capability {
        return self.decide_capability_bound_event(event, capability);
    }

    if self.config.allow_display_without_capability && !requires_capability {
        return ProxyDecision::Forward;
    }

    let reason = if requires_capability {
        format!("capability required for {:?} events", event.classification)
    } else {
        "no capability token provided".to_string()
    };
    ProxyDecision::Block { reason }
}

pub(super) fn capability_error_message(error: &CapabilityError) -> &'static str {
    match error {
        CapabilityError::UntrustedIssuer => "issuer is not trusted",
        CapabilityError::InvalidSignature => "signature did not verify",
        CapabilityError::NotYetValid => "token is not yet valid",
        CapabilityError::Expired => "token has expired",
        CapabilityError::CryptoFloorRejected(_) => "capability crypto floor rejected",
        CapabilityError::AttenuationViolation(_) => "capability rejected by chain binding",
        CapabilityError::BudgetSplitRejected(_) => "capability rejected by sibling-sum budget",
        CapabilityError::Internal(_) => "internal verification error",
    }
}