Giving AI Agents a Governed Identity: A Comprehensive Guide to Architecture, Implementation, and Enforcement with CUGA

wpnews.pro

Implementing Agent Identity with CUGA Agent.

#

Introduction & Background

#

The Shift Towards Autonomous AI Agents

As enterprise AI transitions from stateless chat completion interfaces to autonomous agents capable of reading context, making decisions, executing multi-step workflows, and calling external APIs, traditional application security models begin to break down.

Historically, software integration relied on rigid permission boundaries: either a service used a single, static API key for system-to-system operations, or an application borrowed a user’s session token to perform actions. In an agentic architecture, however, LLM-powered agents plan their own actions, dynamically select tools, and interact across complex execution chains.

┌───────────────────────────────────────────────────────────────────────────┐
│                        UNGOVERNED AGENT ARCHITECTURE                      │
│                                                                           │
│  User ──────► [ Agent Runtime ] ──────► Shared Service Account / API Key  │
│                   │                                 │                     │
│                   ▼                                 ▼                     │
│            Dynamic Tool Call                Unrestricted Access           │
│         (No granular attribution)           (Cannot isolate failures)     │
└───────────────────────────────────────────────────────────────────────────┘

Without a dedicated, governed identity mechanism, agents present several key security risks:

Identity Spoofing & Blurs: When an agent borrows a shared service account or user password, auditors cannot differentiate between an intentional human action and an agent-generated call.
Over-Privileged Execution (Lack of Least Privilege): Agents often run with broad system access, allowing a single prompt injection or hallucinated tool call to inspect, modify, or delete off-limit data.
Uncontrolled Delegation: When an agent acts on behalf of a user, there is often no cryptographic guarantee that the agent’s permission context remains less than or equal to the authorizing user's entitlements.
Lack of Granular Containment (Kill Switch Gap): Disabling a compromised agent requires revoking broad service accounts or shared credentials, which risks taking down unaffected services or agent runtimes.

#

What is Governed Agent Identity?

An AI agent acts on its own: it reads data, calls tools and sometimes changes things. Agent Identity gives each agent its own verifiable identity in the same way a company gives each employee a badge. The agent is registered, approved by a different person, and then proves who it is on every call.

Its core purpose is to answer four questions about every action an agent takes:

Agent Identity establishes AI agents as first-class identity entities—similar to issuing an employee badge within an enterprise identity framework. Each agent is registered in a secure registry, reviewed under strict Separation of Duties (where the creator cannot approve their own agent), bound to its own OAuth client, and granted short-lived, cryptographically signed tokens.

┌───────────────────────────────────────────────────────────────────────────┐
│                         GOVERNED AGENT IDENTITY                           │
│                                                                           │
│  Developer ──► Register ──► Approver (Separation of Duties) ──► Active    │
│                                                                  │        │
│  User Token + Agent Credentials ──► RFC 8693 Token Exchange ─────┘        │
│                                           │                               │
│                                           ▼                               │
│  Audit Trail ◄── [ Identity Gateway ] ──► Short-Lived Scoped Token        │
│   (Agent ID)          (Verification)              │                       │
│                                                   ▼                       │
│                                          Protected Business Data          │
└───────────────────────────────────────────────────────────────────────────┘

Governed identity answers four critical operational questions for every execution step:

Security Pillar	Operational Requirement	Governed Agent Identity Realization
Accountability	Who performed this action?	Every call carries an unforgeable Agent ID. All runtime and lifecycle actions log to a single correlation trail.
Least Privilege	What is the maximum allowed scope?	Scope ceilings (allowed_scopes) are defined at onboarding. Issued tokens can never exceed this ceiling, even if a tool is exposed.
Safe Delegation	On whose authority is the agent acting?	Delegated tokens utilize RFC 8693 token exchange (sub = User, act.sub = Agent). Scope is bounded by the intersection of user rights and agent rights.
Immediate Control	Can we isolate a rogue agent instantly?	A centralized kill switch shifts status to SUSPENDED, blocking token issuance and forcing instant rejection at identity gateways.

#

System Architecture & Lifecycle Governance

#

Component Architecture & Layered Defense

A robust governed agent architecture splits responsibility into two distinct runtime layers: LLM Policy Framing and Resource-Level Identity Enforcement.

Cuga Policy Framing : Runsbefore the model or tool invocation. Intent guards filter dangerous prompts (e.g., bulk deletion), while tool approval rules freeze execution until human validation is received. 2. Resource-Level Identity Enforcement : Runsinside the backend tool execution path. The tool requests token minting via anIdentityGateway , which checks credentials, validates identity backend status, and asserts token signatures and scope ceilings.

#

Complete Agent Lifecycle State Machine

An agent's operational permissions are governed by an explicit state machine. Creation, approval, suspension, and reactivation are managed via formal administrative operations with embedded audit trails.

DEFAULT: The initial state after registration by a developer. The agent has an assigned Agent ID and scope ceiling, but cannot mint tokens or register an OAuth client.
ACTIVE: Set by an independent approver. Dynamic OAuth client registration is unlocked, and short-lived tokens can be issued for authorization.
SUSPENDED (Kill Switch): Suspends token issuance immediately. When coupled with strict status validation on resource tools, active tokens are rejected.

#

Access Delegation Mechanics & Scoping Modes

                                  GRANT TYPES & ACCESS MODES
                                              │
         ┌────────────────────────────────────┼────────────────────────────────────┐
         ▼                                    ▼                                    ▼
  Autonomous Mode                      Delegated Mode                      Privileged Mode
(client_credentials)                  (RFC 8693 Exchange)               (authorization_details)
         │                                    │                                    │
  • Agent acts alone                   • Agent acts for user                • Sensitive operations
  • Background analytics               • HR / Course enrollment             • Service restarts
  • Claims: agent_id                   • Claims: sub=user, act=agent        • Short TTL + Step-up human check

Governed identities support four primary access modes depending on operational context:

Autonomous Access (client_credentials) : Used for automated, non-interactive processes. The agent authenticates using its OAuthclient_id and secret, receiving a token scoped to its defined ceiling.
Delegated Access / On-Behalf-Of (RFC 8693 Token Exchange): Used when an agent executes tasks on behalf of an authenticated end user. The identity gateway exchanges the user's JWT for a delegated token wheresub is the user andact.sub is the agent.
Privileged Action (Human-in-the-Loop Scoping): Designed for sensitive operations (e.g., service restarts or financial transactions). Requires interactive human approval alongside short-lived tokens carrying explicitauthorization_details .
Agent Chaining (Nested Delegation): When a supervisor agent delegates sub-tasks to a downstream worker agent, tokens nest the actor claim (act: worker -> supervisor ), maintaining complete auditability across multi-agent steps.

#

Implementation with CUGA-Agent Integration

The following reference implementation demonstrates how to build a governed agent using Cuga, an Identity Gateway, an isolated Policy Store, and custom Guarded Tools.

#

The demonstration use-case

The following is what is implemented (containing simulation scenario)

Cuga Agent is used for real: agent construction, tool execution, intent-guard and tool-approval policies, human-in-the-loop resume, supervisor delegation. #

Agent Identity has no public SDK. The showcase therefore ships alocal mock of the IBM Verify identity service (`MockVerifyBackend` : RS256-signed JWTs, agent registry, dynamic client registration, client-credentials and RFC 8693 token exchange).`HttpVerifyBackend` is a reference implementation of the same interface for a real tenant. It hasnot been tested against a live tenant ; treat it as a starting point. #

Cuga Harness Kit is not a test runner. It scaffolds skills for coding assistants (11 skills, installed withinit --targets bob ). The showcase runs it into a temporary folder to prove it works, and uses its ownharness/scenarios.py (10 conformance scenarios) for validation.

The LLM engine is selectable : localOllama (default,granite4.2:latest ),llama.cpp (llama-server ) or acloud OpenAI-compatible API. Pick it in the UI or in.env (seeLLM engines ). Tests and the offline demo use a deterministic fake model, so no LLM is needed to run them.
Live status of the engines: Ollama detection, test and switching were exercised against a real local daemon.llama.cpp andcloud were tested with mocked HTTP only (nollama-server and no cloud key were available), so treat them as unverified against a real server.

#

Technical Features of the use-case

Agent lifecycle : create, approve (adifferent person), register an OAuth client, deploy, suspend (kill switch), reactivate. 2. Four agents covering different identity patterns: #

Revenue Analyst : autonomous agent with a read-only scope ceiling. #

Ops SRE Assistant : privileged actions (service restart) bound to a short-lived token. #

Workplace Supervisor : multi-agent orchestration with a nested actor chain in the token. 3. 4. Defence in depth : Cuga policies (before the LLM), then per-tool identity checks (after the LLM). The LLM never sees a token. 5. Interactive documentation : six identity flows rendered as sequence diagrams, a token inspector, an audit trail keyed by Agent ID. 6. Pluggable LLM engines : Ollama, llama.cpp and cloud behind one interface, with detection, a settings page withTest connection , runtime switching without a restart, a documented fallback order and error messages that say how to fix the problem. 7. Validation : 118 unit tests, 10 harness conformance scenarios, a CLI demo and a UI smoke test. 8. Operations : structured logging, typed configuration, scripts for start/stop/cleanup, a multi-stage Containerfile.

Identity Models and Gateway Setup : first, we establish core identity models, custom error types, context management, and token validation mechanisms.

"""Domain models shared by the identity layer."""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any

class AgentStatus(str, Enum):
    """Lifecycle states of an agent in the IBM Agent Registry."""

    DEFAULT = "DEFAULT"  # created by the developer, not yet approved
    ACTIVE = "ACTIVE"  # approved: may register clients and obtain tokens
    SUSPENDED = "SUSPENDED"  # blocks NEW token issuance (issued tokens stay valid!)
    UNDER_REVIEW = "UNDER_REVIEW"

class Entitlement(str, Enum):
    """Management-plane entitlements (names follow the IBM docs)."""

    READ_AGENTS = "readAgents"
    WRITE_AGENTS = "writeAgents"
    MANAGE_AGENT_STATUS = "manageAgentStatus"

@dataclass(frozen=True)
class Principal:
    """A human management user (developer or approver)."""

    user_id: str
    display_name: str
    entitlements: frozenset[Entitlement]

    def has(self, ent: Entitlement) -> bool:
        return ent in self.entitlements

@dataclass
class AgentRecord:
    """An entry of the Agent Registry."""

    agent_id: str
    name: str
    description: str
    owner: str  # user_id of the creating developer
    status: AgentStatus = AgentStatus.DEFAULT
    allowed_scopes: frozenset[str] = frozenset()
    approved_by: str | None = None
    client_id: str | None = None
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat(timespec="seconds"))
    metadata: dict[str, Any] = field(default_factory=dict)

@dataclass(frozen=True)
class ClientCredentials:
    """OAuth client bound to an agent. The secret is returned only once."""

    client_id: str
    client_secret: str
    agent_id: str

Next, we implement context handling to store end-user session state securely without exposing authorization tokens directly to the LLM model space.

"""Identity gateway used by tools.

The LLM never sees a token. Tools call :meth:`IdentityGateway.authorize`, which
reads the *call context* (who is the end user, which tokens are available) from
a ``ContextVar`` set by the runtime, mints a just-in-time token for the right
grant type, validates it like a resource server would, and returns the
validated :class:`~agent_identity.tokens.TokenContext`.
"""

from __future__ import annotations

import contextvars
from dataclasses import dataclass
from typing import Any

from .audit import AuditLog
from .errors import DelegationError, IdentityError
from .lifecycle import AgentIdentityService
from .logging_setup import get_logger
from .tokens import TokenContext, TokenValidator

log = get_logger("gateway")

@dataclass(frozen=True)
class CallContext:
    """Per-request identity context (propagated to tools via a ContextVar)."""

    user_id: str | None = None
    user_token: str | None = None
    supervisor_token: str | None = None

_CALL: contextvars.ContextVar[CallContext | None] = contextvars.ContextVar("agent_identity_call", default=None)

def set_call_context(ctx: CallContext | None) -> contextvars.Token:
    return _CALL.set(ctx)

def reset_call_context(token: contextvars.Token) -> None:
    _CALL.reset(token)

def current_call_context() -> CallContext | None:
    return _CALL.get()

class IdentityGateway:
    """Mints and validates tokens for tool executions."""

    def __init__(self, service: AgentIdentityService, issuer: str, public_key: Any, strict_status_check: bool = True) -> None:
        self._service = service
        self._issuer = issuer
        self._key = public_key
        self._strict = strict_status_check
        self._validators: dict[str, TokenValidator] = {}

    @property
    def audit(self) -> AuditLog:
        return self._service.audit

    def validator(self, audience: str) -> TokenValidator:
        """Return (and cache) the resource-server validator for ``audience``."""
        if audience not in self._validators:
            lookup = (lambda aid: self._service.backend.get_agent(aid).status) if self._strict else None
            self._validators[audience] = TokenValidator(
                self._key, self._issuer, audience, self._service.audit, status_lookup=lookup
            )
        return self._validators[audience]

    def authorize(
        self,
        *,
        agent_name: str,
        mode: str,  # "obo" (on behalf of the user) | "autonomous"
        audience: str,
        scopes: list[str],
        resource: str,
        authorization_details: list[dict[str, Any]] | None = None,
    ) -> TokenContext:
        """Mint + validate a token; raises an :class:`IdentityError` subclass on denial."""
        agent = self._service.backend.find_agent(agent_name)
        try:
            if mode == "autonomous":
                token = self._service.agent_token(agent.agent_id, scopes, audience)
                return self.validator(audience).validate(token, tuple(scopes), resource=resource)

            call = current_call_context()
            subject = (call.supervisor_token or call.user_token) if call else None
            if not subject:
                raise DelegationError("No end-user context: on-behalf-of access needs a logged-in user")
            token = self._service.delegated_token(agent.agent_id, subject, scopes, audience, authorization_details)
            return self.validator(audience).validate(
                token,
                tuple(scopes),
                require_delegation=True,
                require_privileged=bool(authorization_details),
                resource=resource,
            )
        except IdentityError as exc:
            log.warning("authorize failed agent=%s resource=%s: %s", agent_name, resource, exc)
            raise

Defining Guarded Tools: tools assert authorization boundaries via theIdentityGateway prior to interacting with backend resources.


"""Identity-guarded tools for the example agents.

Every tool obtains a token through the :class:`IdentityGateway` *before* touching
data. Denials are returned as ``ACCESS DENIED: ...`` text so the model can relay
them faithfully, and are also written to the audit log.
"""

import json
from langchain_core.tools import tool
from agent_identity.models import IdentityError
from agent_identity.gateway import IdentityGateway

COURSES_DB = {
    "SEC301": {"title": "Advanced Cloud Security", "cost": 1200},
    "PY101": {"title": "Python Basics", "cost": 300}
}

def create_learning_coach_tools(gateway: IdentityGateway, agent_name: str):

    @tool
    def list_courses() -> str:
        """List available training courses, costs, and IDs."""
        try:
            gateway.authorize(
                agent_name=agent_name,
                mode="obo",
                audience="hr-service",
                scopes=["courses:read"]
            )
        except IdentityError as exc:
            return f"ACCESS DENIED: {str(exc)}"

        return json.dumps(COURSES_DB)

    @tool
    def enroll_expensive_course(course_id: str) -> str:
        """Enroll the user in a course costing over 500 USD. Requires approval."""
        course = COURSES_DB.get(course_id.upper())
        if not course:
            return f"Error: Course {course_id} not found."

        details = [{
            "type": "course_enrollment",
            "course_id": course_id.upper(),
            "max_cost": course["cost"]
        }]

        try:
            gateway.authorize(
                agent_name=agent_name,
                mode="obo",
                audience="hr-service",
                scopes=["courses:enroll"],
                authorization_details=details
            )
        except IdentityError as exc:
            return f"ACCESS DENIED: {str(exc)}"

        return f"Successfully enrolled in {course['title']} ({course_id.upper()})."

    return [list_courses, enroll_expensive_course]

Constructing the Governed Cuga Agent Environment: next, we initialize the agent with isolated policy stores and human-in-the-loop gates.

"""Example agent 1 - Learning Coach.

Use case: an employee asks an agent to browse and enrol in training courses.
Identity pattern: **on-behalf-of (RFC 8693)** - tokens carry ``sub=<employee>`` and
``act.sub=<agent>``. Expensive courses need a human approval (Cuga tool-approval
policy) *and* a short-lived privileged token (``authorization_details``).
"""

from __future__ import annotations

from agent_identity.spec import AgentSpec, ApprovalSpec, IntentGuardSpec
from agent_identity.tools import build_course_tools

SPEC = AgentSpec(
    name="learning-coach",
    title="Learning Coach",
    description="Helps employees discover and enrol in training courses on their behalf.",
    use_case="HR / learning - delegated user actions",
    access_mode="obo",
    allowed_scopes=frozenset({"courses:read", "courses:enroll"}),
    tool_builder=build_course_tools,
    instructions=(
        "You are the Learning Coach. Use the provided tools to list courses and enrol the current user. "
        "Courses above 500 USD must be enrolled with enroll_expensive_course. "
        "If a tool answers 'ACCESS DENIED', report that denial verbatim and do not retry."
    ),
    guards=(
        IntentGuardSpec(
            name="No bulk record deletion",
            keywords=("delete all", "wipe", "erase all"),
            response="Blocked by policy: this agent cannot delete records.",
        ),
    ),
    approvals=(
        ApprovalSpec(
            name="Approve expensive enrolments",
            tools=("enroll_expensive_course",),
            message="This enrolment exceeds 500 USD. A human must approve it.",
        ),
    ),
    sample_prompts=(
        "List the available courses.",
        "Enroll me in PY101.",
        "Enroll me in SEC301.",
        "Delete all employee records.",
    ),
)

Runtime Switchable LLM Engine Infrastructure: to remain resilient, the runtime LLM engine layer supports non-disruptive switches between local (Ollama, llama.cpp) and cloud models without discarding context or invalidating agent tokens.

                          RUNTIME ENGINE SWITCHING
                                     │
    ┌────────────────────────────────┼────────────────────────────────┐
    │                                │                                │
    ▼                                ▼                                ▼
1. Gate New Turns           2. Drain Running Turns          3. Rebuild Graphs
Close `_ready` gate         Wait for active turn completion Reset `_graph` references
 new messages          Enforce drain timeout           Keep chat history & tokens

"""Pluggable LLM engines: Ollama, llama.cpp and OpenAI-compatible cloud services.

This module is UI-framework free. It provides:

* :class:`LLMEngine` - the abstraction (detect / validate / test_connection / build_model);
* three concrete engines plus an offline ``fake`` engine used by tests and as last resort;
* :class:`EngineManager` - detection, persisted UI choice, and the **resolution order**:

  1. an explicit selection (the UI override saved locally, then ``AI_LLM_ENGINE``);
  2. engines whose endpoint / credentials are configured or detected, in the order of
     ``AI_LLM_FALLBACK_ENGINES``;
  3. a hard-coded fallback (local Ollama), then the offline fake model as a last resort.

Secrets are :class:`pydantic.SecretStr` and are never logged, rendered or written to disk.
Every failure is a :class:`~agent_identity.errors.LLMEngineError` that carries a remediation hint.
"""

from __future__ import annotations

import itertools
import json
import os
import shutil
import stat
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import httpx
from pydantic import SecretStr

from .config import Settings
from .errors import EngineConnectionError, EngineMisconfigured, EngineUnavailable, LLMEngineError
from .logging_setup import get_logger
from .ui_helpers import probe_ollama
...

import asyncio
from typing import Dict, Any

class EngineManager:
    """Manages active LLM engines, handling fallbacks and hot swaps."""

    def __init__(self, current_config):
        self.current_config = current_config
        self._gate_open = asyncio.Event()
        self._gate_open.set()
        self._active_turns = 0

    async def switch_engine(self, new_config: Dict[str, Any], agent_instances: list):
        """Swaps the active LLM engine using a gate-drain-swap-rebuild workflow."""
        self._gate_open.clear()

        drain_timeout = 30.0
        start_time = asyncio.get_event_loop().time()
        while self._active_turns > 0:
            if asyncio.get_event_loop().time() - start_time > drain_timeout:
                self._gate_open.set()
                raise RuntimeError("Engine switch timed out while draining active turns.")
            await asyncio.sleep(0.1)

        try:
            self.current_config = new_config
            new_model = self._build_model_instance(new_config)

            for agent in agent_instances:
                agent._model = new_model
                agent._graph = None
                agent._compiled_graph = None

        except Exception as exc:
            self._gate_open.set()
            raise RuntimeError(f"Engine switch failed: {exc}")

        self._gate_open.set()

    def _build_model_instance(self, config):
        pass

#

Conclusion & Realization via CUGA-Agent

#

Summary of Architecture & Security Benefits

By combining the Cuga Agent Framework with a governed identity architecture, enterprise deployments achieve end-to-end security, regulatory compliance, and operational observability:

┌────────────────────────────────────────────────────────────────────────────┐
│                      GOVERNED AGENT SECURITY MATRIX                        │
│                                                                            │
│  Layer 1: Intent Policy ──► Pre-LLM prompt filtering                       │
│  Layer 2: Tool Approval ──► Human-in-the-Loop validation                   │
│  Layer 3: OAuth Gateway ──► Scoped token issuance (Intersection principle) │
│  Layer 4: Target Tool   ──► Strict identity signature & status checks      │
│  Layer 5: Audit Log     ──► Single trace correlated by Agent ID            │
└────────────────────────────────────────────────────────────────────────────┘

Deterministic Defense-in-Depth : Policy guards runbefore the model evaluates prompts, while Identity Gateways enforce scope checksinside the execution tool. Even if an LLM is hijacked via prompt injection, it remains constrained by the token's scope ceiling. 2. True Separation of Duties : Autonomous agents cannot self-approve or inherit excessive access, eliminating untracked administrative expansion. 3. Seamless Multi-Agent Delegation : RFC 8693 token exchange preserves the chain of custody across complex multi-agent workflows (User -> Supervisor -> Worker ), providing clear attribution for downstream audits. 4. Resilient Architecture : Runtime LLM engine switching allows teams to hot-swap model backends (local or cloud) without disrupting active user sessions, token stores, or audit histories.

#

Overall implementation: Governed CUGA Agent Implementation

Below is a detailed structural representation of the complete governed agent architecture:

Implementing CUGA-Agent with Governed Identity transitions autonomous systems from opaque risk drivers into auditable enterprise assets:

No Shared Passwords/Keys : Every agent operates under its own client registration. #

Cryptographic Bounds : Scope ceilings ensure agents can never exceed their approved boundaries, even when acting for privileged users. #

Instant Containment : The strict-status kill switch invalidates active sessions in real time, preventing rogue executions. #

Traceable Attribution : Unified audit logging tags every runtime action and lifecycle change with a consistentAgent ID .

Thanks for reading 🦉

#

Links

Code repository for this post: https://github.com/aairom/cuga-agentidentity
Cuga + Agent Identity showcase: https://github.com/aairom/cuga-agentidentity/blob/main/Docs/AgentIdentity-Showcase.html
CUGA Project: https://github.com/cuga-project
CUGA Agent: https://github.com/cuga-project/cuga-agent
CUGA Harness Kit: https://github.com/cuga-project/cuga-agent
Agent Identity: https://www.ibm.com/products/agent-identity
What is AI Agent Identity: https://www.okta.com/identity-101/what-is-ai-agent-identity/

source & further reading

dev.to — original article T3 Code: Control All Your AI Coding Agents From Your Phone AI agents are your new customers. I built a checker to see which stores they can actually read Karpathy Skills: One File That Makes Your AI Coding Agent Write Better Code