Building AI Agents for DevOps Part 7: ADK Plugins and Async Tools

In Part 6, I hardened the platform's security — input validation, confirmation bypass fixes, secret redaction, and authentication enforcement. The agents are safe. But the wiring that connects...

In Part 6, I hardened the platform's security — input validation, confirmation bypass fixes, secret redaction, and authentication enforcement. The agents are safe. But the wiring that connects cross-cutting concerns (RBAC, metrics, audit, resilience) to each agent is duplicated everywhere.

This post eliminates that duplication with ADK Plugins and converts all tools to async. Two changes, one goal: follow the framework's recommended patterns instead of fighting them.

The Problem: Callback Wiring Everywhere

Every agent in the platform needed the same six callbacks wired in the same order:

from ai_agents_core import (
    create_agent, authorize, require_confirmation,
    audit_logger, activity_tracker, graceful_tool_error,
    graceful_model_error, CircuitBreaker, MetricsCollector,
)

circuit_breaker = CircuitBreaker()
metrics = MetricsCollector(circuit_breaker=circuit_breaker)

root_agent = create_agent(
    name="kafka_health_agent",
    model=model,
    instruction="...",
    tools=[...],
    before_agent_callback=ensure_default_role(),
    before_tool_callback=[
        authorize(),
        require_confirmation(),
        circuit_breaker.before_tool_callback,
        metrics.before_tool_callback,
    ],
    after_tool_callback=[
        audit_logger(),
        activity_tracker(),
        metrics.after_tool_callback,
    ],
    on_tool_error_callback=[
        graceful_tool_error(),
        metrics.on_tool_error_callback,
    ],
    on_model_error_callback=graceful_model_error(),
)

This block appeared in every single agent.py. Six agents means six copies. Add a new concern (say, rate limiting) and you touch every agent. Forget one callback in one agent and you silently lose audit logging or circuit breaking for that agent's tools.

The Google ADK has a better answer: Plugins.

ADK Plugins: Register Once, Apply Everywhere

An ADK Plugin is a BasePlugin subclass registered on the Runner. Unlike agent-level callbacks, plugins apply globally to every agent, tool, and LLM call managed by that Runner — including sub-agents. Register once, never think about it again.

The Plugin API

Each plugin can implement any combination of these async hooks:

Plugins run in registration order. This matters — RBAC must block before metrics start timing.

Wrapping Existing Callbacks

I didn't rewrite the existing callback logic. Each plugin wraps the factory functions that already existed:

from google.adk.plugins import BasePlugin

class GuardrailsPlugin(BasePlugin):
    """RBAC authorization + confirmation prompts."""

    def __init__(self, role_policy=None, mode="confirm"):
        self._authorize = authorize(role_policy=role_policy)
        self._confirm = require_confirmation() if mode == "confirm" else dry_run()
        self._ensure_role = ensure_default_role()

    async def before_agent_callback(self, *, callback_context, **kwargs):
        return self._ensure_role(callback_context)

    async def before_tool_callback(self, *, tool, args, tool_context, **kwargs):
        result = self._authorize(tool, args, tool_context)
        if result is not None:
            return result
        return self._confirm(tool, args, tool_context)


class MetricsPlugin(BasePlugin):
    """Prometheus metrics for every tool call."""

    def __init__(self, circuit_breaker=None):
        self._collector = MetricsCollector(circuit_breaker=circuit_breaker)

    async def before_tool_callback(self, *, tool, args, tool_context, **kwargs):
        return self._collector.before_tool_callback(tool, args, tool_context)

    async def after_tool_callback(self, *, tool, args, tool_context, result, **kwargs):
        return self._collector.after_tool_callback(tool, args, tool_context, result)

    async def on_tool_error_callback(self, *, tool, args, tool_context, error, **kwargs):
        return self._collector.on_tool_error_callback(tool, args, tool_context, error)

Same pattern for ResiliencePlugin, AuditPlugin, ActivityPlugin, and ErrorHandlerPlugin. Each wraps one existing concern.

The Factory: default_plugins()

One function returns all six plugins in the correct order:

def default_plugins(
    role_policy=None,
    guardrail_mode="confirm",
    failure_threshold=5,
) -> list[BasePlugin]:
    resilience = ResiliencePlugin(failure_threshold=failure_threshold)
    metrics = MetricsPlugin(circuit_breaker=resilience.circuit_breaker)
    return [
        GuardrailsPlugin(role_policy=role_policy, mode=guardrail_mode),
        resilience,
        metrics,
        AuditPlugin(),
        ActivityPlugin(),
        ErrorHandlerPlugin(),
    ]

Order matters. Here's why:

Request → GuardrailsPlugin  →  RBAC blocks unauthorized users
       → ResiliencePlugin   →  circuit breaker rejects if open
       → MetricsPlugin      →  starts timer
       → [tool executes]
       → MetricsPlugin      →  records duration
       → AuditPlugin        →  logs tool call
       → ActivityPlugin     →  records to session state
       → ErrorHandlerPlugin →  catches exceptions, returns friendly errors

Before and After: Agent Code

Here's what a typical agent looks like now:

# agent.py — AFTER (entire file)
from ai_agents_core import create_agent, load_agent_env
from .tools import (
    get_cluster_health, list_topics, describe_topic,
    get_consumer_groups, get_consumer_lag,
    create_kafka_topic, delete_kafka_topic, produce_test_message,
)

config, model = load_agent_env()

kafka_health_agent = create_agent(
    name="kafka_health_agent",
    model=model,
    instruction="You are a Kafka health monitoring agent...",
    tools=[
        get_cluster_health, list_topics, describe_topic,
        get_consumer_groups, get_consumer_lag,
        create_kafka_topic, delete_kafka_topic, produce_test_message,
    ],
    description="Handles Kafka cluster health monitoring and management.",
)

No callbacks. No imports from six different modules. No CircuitBreaker or MetricsCollector instantiation. The agent just declares its tools and instructions.

The plugins are registered on the Runner, once:

# runner setup
from ai_agents_core import default_plugins
from google.adk.runners import Runner

runner = Runner(
    agent=root_agent,
    app_name="devops_assistant",
    session_service=session_service,
    plugins=default_plugins(),
)

Async Tools: Stop Blocking the Event Loop

Every tool in the platform was a synchronous def that called blocking I/O — Kafka admin operations, Kubernetes API calls, Docker subprocess commands, HTTP requests. ADK runs on asyncio, so each sync tool blocked the entire event loop during execution.

The fix: convert all tools to async def and offload blocking calls appropriately.

Three Patterns for Three Types of I/O

Kafka: Thread Pool Executor

The confluent-kafka library is not async-compatible. A small helper offloads calls to a thread pool:

from functools import partial

async def _run_sync(func, *args, **kwargs):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, partial(func, *args, **kwargs))

async def get_cluster_health() -> dict:
    admin = _get_admin_client()
    metadata = await _run_sync(admin.list_topics, timeout=10)
    # ... process metadata

Kubernetes: asyncio.to_thread()

The simpler built-in for Python 3.9+:

async def list_pods(namespace: str = "default") -> dict:
    if err := validate_string(namespace, "namespace", pattern=K8S_NAME_PATTERN):
        return err
    v1 = _get_core_v1_api()
    pods = await asyncio.to_thread(v1.list_namespaced_pod, namespace)
    return {"status": "success", "pods": [_format_pod(p) for p in pods.items]}

Docker: Native Async Subprocess

Docker tools were using subprocess.run(). This is the cleanest conversion — asyncio has native subprocess support:

async def _run_docker(*args: str, timeout: int = 30) -> tuple[bool, str]:
    proc = await asyncio.create_subprocess_exec(
        "docker", *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, stderr = await asyncio.wait_for(
            proc.communicate(), timeout=timeout
        )
    except asyncio.TimeoutError:
        proc.kill()
        return False, f"Command timed out after {timeout}s"

    if proc.returncode != 0:
        return False, stderr.decode().strip()
    return True, stdout.decode().strip()

The return type changed from subprocess.CompletedProcess to tuple[bool, str], which simplified all the calling code — no more checking .returncode and decoding .stdout.

Testing: From Sync to Async

Every test that called a tool function needed updating. The pattern is mechanical:

# Before
@patch("kafka_health_agent.tools._get_admin_client")
def test_get_cluster_health(self, mock_client):
    result = get_cluster_health()
    assert result["status"] == "success"

# After
@patch("kafka_health_agent.tools._get_admin_client")
@pytest.mark.asyncio
async def test_get_cluster_health(self, mock_client):
    result = await get_cluster_health()
    assert result["status"] == "success"

Three changes per test: add @pytest.mark.asyncio, change def to async def, add await before the tool call.

Docker Tests: Mock Strategy Change

The Docker tests required a deeper change. They previously mocked subprocess.run, but the tools now use asyncio.create_subprocess_exec. Instead of mocking the subprocess layer, I mock _run_docker directly:

# Before — mocking subprocess
@patch("devops_assistant.docker_tools.subprocess.run")
def test_list_containers(self, mock_run):
    mock_run.return_value = CompletedProcess(
        args=[], returncode=0, stdout="container1\ncontainer2"
    )
    result = list_containers()

# After — mocking the async helper
@patch("devops_assistant.docker_tools._run_docker", new_callable=AsyncMock)
@pytest.mark.asyncio
async def test_list_containers(self, mock_docker):
    mock_docker.return_value = (True, "container1\ncontainer2")
    result = await list_containers()

Cleaner mocks, cleaner assertions. AsyncMock handles the await transparently.

Test Count

9 new tests for the plugin system. Total: 395 → 404.

$ make test
404 passed in 4.12s

The Slack Bot: Hybrid Approach

The Slack bot is the one place where plugins don't fully replace callbacks. Slack needs interactive Approve/Deny buttons for confirmation — a UI concern that doesn't belong in a global plugin.

The solution: use default_plugins(guardrail_mode="none") for everything except confirmation, and keep the Slack-specific button callback on the agent:

# Slack bot setup
plugins = default_plugins(guardrail_mode="none")  # skip confirmation plugin

runner = Runner(
    agent=root_agent,
    app_name="slack_bot",
    session_service=session_service,
    plugins=plugins,
)

# Slack-specific confirmation stays as agent callback
root_agent = create_agent(
    ...,
    before_tool_callback=slack_confirmation_callback,
)

RBAC, metrics, audit, resilience, activity tracking, and error handling all come from plugins. Only the confirmation UI stays agent-level.

What Changed (Summary)

New file: core/ai_agents_core/plugins.py (6 plugins + factory) · Files changed: 16 · Lines removed from agents: ~180

Source code on GitHub. Plugins module: core/orrery_core/plugins/. Changelog: CHANGELOG.md.