Building AI Agents for DevOps Part 3: RBAC Who Can Do What

In Part 2, I brought the DevOps agent into Slack. Now any team member can chat with the agent from a channel. But that raises a question: should a junior engineer be able to delete a Kafka topic?...

Updated for Orrery 0.6.0 (October 2026). Roles are still inferred from @confirm and @destructive, exactly as described here. Three layers now sit around them: autonomy levels L2 to L4 (what this deployment may do at all), protected namespaces (opt-in: where non-admins may change things) and, since 0.6.0, impersonation, so Kubernetes and Elasticsearch also enforce their own permissions for the person who asked. Tools without a Python function, such as MCP tools, carry the same labels: see Part 17.

In Part 2, I brought the DevOps agent into Slack. Now any team member can chat with the agent from a channel. But that raises a question: should a junior engineer be able to delete a Kafka topic? Should an on-call viewer be able to restart pods?

The guardrails from Part 1 ask "are you sure?", but they don't ask "are you allowed?". This post adds that missing layer — a role-based access control system that reuses the guardrail decorators I already had.

The Problem

Before RBAC, every user had the same power. The confirmation prompt was the only gate:

User: "delete topic events"

Agent: ⚠️ This is a destructive operation. Are you sure?
User: yes
Agent: ✅ Topic deleted.

That's fine for a solo developer. In a shared Slack channel with 20 people, it's a liability. I needed a way to say: viewers can read, operators can mutate, admins can destroy.

The Key Insight: I Already Classified Every Tool

The guardrail decorators from Part 1 already tell me the risk level of every tool:

# Read-only — no decorator
def list_kafka_topics() -> dict: ...

# Mutating — @confirm
@confirm("creates a new topic on the cluster")
def create_kafka_topic(topic_name: str) -> dict: ...

# Irreversible — @destructive
@destructive("permanently deletes the topic and all its data")
def delete_kafka_topic(topic_name: str) -> dict: ...

So instead of creating a separate permission system, I derived roles from decorators:

No decorator  →  VIEWER can call it    (read-only)
@confirm      →  OPERATOR can call it  (mutating)
@destructive  →  ADMIN can call it     (irreversible)

Zero re-annotation. Every existing tool automatically got the right permission level.

Three Roles, One Hierarchy

The role system is intentionally simple — an integer enum:

class Role(IntEnum):
    VIEWER = 0    # read-only tools
    OPERATOR = 1  # + mutating tools (@confirm)
    ADMIN = 2     # + destructive tools (@destructive)

Higher roles inherit lower permissions. An ADMIN can do everything an OPERATOR can, which can do everything a VIEWER can. The check is a single comparison:

if user_role >= required_role:
    return None  # authorized, proceed

I considered more granular models (per-tool ACLs, resource-level permissions) but they would have added complexity without solving a real problem yet. Three roles cover 90% of DevOps team structures.

How It Works

The entire RBAC system is one callback — authorize() — that plugs into the same before_tool_callback pipeline as the guardrails:

root_agent = create_agent(
    name="kafka_health_agent",
    tools=[list_kafka_topics, create_kafka_topic, delete_kafka_topic],
    before_tool_callback=[authorize(), require_confirmation()],
    #                      ▲ RBAC first    ▲ then guardrails
)

Order matters. When a viewer tries to delete a topic:

An operator trying the same thing also gets blocked — @destructive requires ADMIN. But an operator can create a topic (@confirm → OPERATOR), after confirming:

Where Does the Role Come From?

The user's role is a simple string in ADK's session state:

session.state["user_role"] = "operator"  # set at session creation

The authorize() callback reads it on every tool call:

def get_user_role(tool_context: Context) -> Role:
    role_name = tool_context.state.get("user_role", "viewer")
    try:
        return Role[role_name.upper()]
    except (KeyError, AttributeError):
        return Role.VIEWER  # unknown role → least privilege

The integration layer — not the agent — sets the role. This keeps the agent transport-agnostic. The Slack bot resolves roles from Slack user IDs:

class SlackBotConfig(AgentConfig):
    slack_admin_users: str = ""      # comma-separated Slack user IDs
    slack_operator_users: str = ""

    def resolve_role(self, user_id: str) -> str:
        admins = {u.strip() for u in self.slack_admin_users.split(",") if u.strip()}
        operators = {u.strip() for u in self.slack_operator_users.split(",") if u.strip()}
        if user_id in admins:
            return "admin"
        if user_id in operators:
            return "operator"
        return "viewer"

When a new Slack thread starts, the role is injected into the session:

role = self._config.resolve_role(user_id)
session = await self.session_service.create_session(
    app_name="slack_devops",
    user_id=user_id,
    state={"user_role": role},  # ← role set once per thread
)

Configuration is two env vars in agents/slack-bot/.env:

SLACK_ADMIN_USERS=U01ABC123,U02DEF456
SLACK_OPERATOR_USERS=U04XYZ789
# everyone else → viewer

What It Looks Like in Slack

Here's a real interaction. An operator checks Kafka health (read-only, allowed), then creates a topic (mutating, allowed after confirmation):

If a viewer tried the same topic creation, they'd get an access denied message before the confirmation prompt even fires:

A Subtlety: Callbacks Don't Propagate

One thing caught me off guard: ADK's before_tool_callback does not propagate from parent to child agents. When the root devops_assistant delegates to kafka_health_agent, only the Kafka agent's own callbacks fire.

This means authorize() must be registered on every agent that has tools:

# In devops-assistant/agent.py — shared callback instances
_authorize = authorize()

# Every tool-bearing agent gets authorize()
kafka_health_checker = create_agent(
    name="kafka_health_checker",
    tools=[get_kafka_cluster_health, list_kafka_topics, ...],
    before_tool_callback=[_authorize],  # ← must be explicit
    ...
)

docker_agent = create_agent(
    name="docker_agent",
    tools=[list_containers, inspect_container, ...],
    before_tool_callback=[_authorize],  # ← same
    ...
)

This is actually a good design — each agent controls its own security boundary. But it's easy to forget when adding new agents, so I documented it as a rule: when adding a new agent with tools, always include authorize() in its before_tool_callback.

Overrides: When Inference Isn't Enough

Sometimes the decorator-based inference doesn't match what you want. A RolePolicy lets you override per tool:

# This read-only tool contains sensitive data — restrict to operators
policy = RolePolicy(overrides={"get_secrets": Role.OPERATOR})

create_agent(
    ...,
    before_tool_callback=[authorize(policy), require_confirmation()],
)

And for tools that don't use @destructive/@confirm at all, the @requires_role decorator sets an explicit minimum:

@requires_role(Role.ADMIN)
def manage_users() -> dict:
    """Admin-only tool that doesn't modify infrastructure."""
    ...

What This Doesn't Do (On Purpose)

I intentionally scoped RBAC to be minimal:

Each of these could be added later without changing the core design. The authorize() callback, the RolePolicy, and the session state contract are all extension points.

Testing

The RBAC module has 25 tests covering every combination of role × tool type:

def test_viewer_blocked_from_destructive(self, fake_tool, fake_ctx):
    callback = authorize()
    tool = fake_tool(name="delete_topic", func=_make_func(guard="destructive"))
    ctx = fake_ctx(state={"user_role": "viewer"})

    result = callback(tool=tool, args={}, tool_context=ctx)

    assert result["status"] == "access_denied"
    assert "admin" in result["message"]
    assert "viewer" in result["message"]

def test_admin_can_call_everything(self, fake_tool, fake_ctx):
    callback = authorize()
    for guard in [None, "confirm", "destructive"]:
        tool = fake_tool(name="any_tool", func=_make_func(guard=guard))
        ctx = fake_ctx(state={"user_role": "admin"})
        result = callback(tool=tool, args={}, tool_context=ctx)
        assert result is None  # all allowed

Edge cases covered: unknown role strings default to VIEWER, missing state defaults to VIEWER, role comparison is case-insensitive.

The Full Picture

Here's how RBAC fits into the callback pipeline alongside guardrails, audit logging, and activity tracking:

Four independent callbacks, each doing one thing, composed in a list. No inheritance, no middleware chain, no framework magic.

What's Next

With RBAC in place, the platform now has identity-aware access control. Next, I turn the monitoring inward: Part 4 instruments every agent with Prometheus metrics so the platform can observe itself.

The full source code is on GitHub. The RBAC design is documented in ADR-001.