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

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)     │
└───────────────────────────────────────────────────────────────────────────┘

Enter fullscreen mode Exit fullscreen mode

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          │
└───────────────────────────────────────────────────────────────────────────┘

Enter fullscreen mode Exit fullscreen mode

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

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.

1. Cuga Policy Framing: Runs before 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: Runs inside the backend tool execution path. The tool requests token minting via an IdentityGateway, 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

Enter fullscreen mode Exit fullscreen mode

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 OAuth client_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 where sub is the user and act.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 explicit authorization_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 a local 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 has not 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 with init --targets bob). The showcase runs it into a temporary folder to prove it works, and uses its own harness/scenarios.py (10 conformance scenarios) for validation.

• The LLM engine is selectable: local Ollama (default, granite4.2:latest), llama.cpp (llama-server) or a cloud OpenAI-compatible API. Pick it in the UI or in .env (see LLM 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 and cloud were tested with mocked HTTP only (no llama-server and no cloud key were available), so treat them as unverified against a real server.

Technical Features of the use-case

1. Agent lifecycle: create, approve (a different person), register an OAuth client, deploy, suspend (kill switch), reactivate.

2. Four agents covering different identity patterns:

   • Learning Coach: delegated (on-behalf-of) user actions, with human approval and a privileged token for expensive enrolments.

   • 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. Defence in depth: Cuga policies (before the LLM), then per-tool identity checks (after the LLM). The LLM never sees a token.

4. Interactive documentation: six identity flows rendered as sequence diagrams, a token inspector, an audit trail keyed by Agent ID.

5. Pluggable LLM engines: Ollama, llama.cpp and cloud behind one interface, with detection, a settings page with Test connection, runtime switching without a restart, a documented fallback order and error messages that say how to fix the problem.

6. Validation: 118 unit tests, 10 harness conformance scenarios, a CLI demo and a UI smoke test.

7. 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.

# agent_identity/models.py
"""Domain models shared by the identity layer."""

from __future__ import annotations

from dataclasses import dataclass, fieldfrom datetime import datetime, timezonefrom enum import Enumfrom typing import Anyclass 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@dataclassclass 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

Enter fullscreen mode Exit fullscreen mode

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

# agent_identity/gateway.py
"""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 contextvarsfrom dataclasses import dataclassfrom typing import Any

from .audit import AuditLogfrom .errors import DelegationError, IdentityErrorfrom .lifecycle import AgentIdentityServicefrom .logging_setup import get_loggerfrom .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
    # Set when a supervisor delegates to a sub-agent: the supervisor's own
    # on-behalf-of token becomes the *subject* of the sub-agent's exchange, which
    # nests the actors in the ``act`` claim (the transaction/caller chain).
    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

Enter fullscreen mode Exit fullscreen mode

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

# agent_identity/tools.py
"""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 jsonfrom langchain_core.tools import toolfrom agent_identity.models import IdentityErrorfrom 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:
            # Privileged token authorization with specific authorization_details
            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]

Enter fullscreen mode Exit fullscreen mode

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

# agents/learning_coach.py
"""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, IntentGuardSpecfrom 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.",
    ),)

Enter fullscreen mode Exit fullscreen mode

• 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
Pause new messages          Enforce drain timeout           Keep chat history & tokens

Enter fullscreen mode Exit fullscreen mode

# agent_identity/llm_engines.py
"""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 itertoolsimport jsonimport osimport shutilimport statimport timefrom abc import ABC, abstractmethodfrom dataclasses import dataclass, fieldfrom pathlib import Pathfrom typing import Anyfrom urllib.parse import urlparse

import httpxfrom pydantic import SecretStr

from .config import Settingsfrom .errors import EngineConnectionError, EngineMisconfigured, EngineUnavailable, LLMEngineErrorfrom .logging_setup import get_loggerfrom .ui_helpers import probe_ollama...

import asynciofrom 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."""
        # Step 1: Gate incoming execution turns
        self._gate_open.clear()

        # Step 2: Drain active turns
        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)

        # Step 3: Swap configuration and rebuild agent execution graphs
        try:
            self.current_config = new_config
            new_model = self._build_model_instance(new_config)

            for agent in agent_instances:
                # Update underlying model reference and signal graph rebuild
                agent._model = new_model
                agent._graph = None
                agent._compiled_graph = None

        except Exception as exc:
            # Revert to safe state if swap fails
            self._gate_open.set()
            raise RuntimeError(f"Engine switch failed: {exc}")

        # Step 4: Re-open gate for new traffic
        self._gate_open.set()

    def _build_model_instance(self, config):
        # Initializes Ollama, llama.cpp, Cloud OpenAI API, or Fake test model
        pass

Enter fullscreen mode Exit fullscreen mode

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            │
└────────────────────────────────────────────────────────────────────────────┘

Enter fullscreen mode Exit fullscreen mode

1. Deterministic Defense-in-Depth: Policy guards run before the model evaluates prompts, while Identity Gateways enforce scope checks inside 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.

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 consistent Agent 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/