Who on-calls an AI? This is the unresolved crisis of agent ownership

At 2:07 a.m., an AI agent starts failing. The application is still running, but requests are timing out because a downstream API is returning errors. The agent has already retried twice, using more tokens than usual, and its response time has roughly doubled. The first question should not be:

  “Which model should we use?”

It should be:

  “Who owns this system right now?”

That question is easy to overlook when building AI agents. Teams often pour effort into prompts, retrieval, tools, and model selection, but production ownership determines whether an agent becomes a dependable service or another source of operational risk. “Teams often pour effort into prompts, retrieval, tools, and model selection, but production ownership determines whether an agent becomes a dependable service or another source of operational risk.” A production agent needs an owner, boundaries, telemetry, failure policies, escalation paths, and a clear distinction between what the system may automate and what requires human intervention. A production agent should not have a direct path from reasoning to action. Each tool invocation should pass through explicit controls, remain observable, and have a defined path to human intervention. Here’s a practical way to build those controls.

Step 1: Treat the agent as a production service

An agent should have the same operational identity as any other production service. Start by defining its ownership metadata.

Python
from pydantic import BaseModel

class AgentService(BaseModel):
    name: str
    owner_team: str
    service_tier: str
    escalation_channel: str

agent_service = AgentService(
    name="incident-command-agent",
    owner_team="platform-engineering",
    service_tier="critical",
    escalation_channel="#platform-oncall",
)

This may look simple, but it changes the conversation. The agent is no longer an experiment running somewhere in a Python process. It is a service with an accountable team and an escalation path. “The agent is no longer an experiment running somewhere in a Python process. It is a service with an accountable team and an escalation path.” You can later attach the same metadata to logs, traces, dashboards, alerts, and deployment records.

Step 2: Define what the agent is allowed to do

Ownership also means being clear about what an agent is allowed to do. An agent that can only read telemetry carries far less risk than one that can restart production services, change infrastructure, or deploy code. That’s why I prefer an explicit tool allowlist, which limits the agent to the actions you’ve deliberately granted it.

Python
ALLOWED_TOOLS = {
    "get_service_health",
    "get_recent_deployments",
    "search_runbooks",
    "summarize_incident",
}

class ToolAuthorizationError(Exception):
    """Raised when an agent attempts to use an unapproved tool."""

def authorize_tool(tool_name: str) -> None:
    if tool_name not in ALLOWED_TOOLS:
        raise ToolAuthorizationError(
            f"Tool '{tool_name}' is not authorized."
        )

The guiding principle is deny by default. Adding a new capability should be an intentional engineering decision, not something that happens because a tool became available to the agent. For destructive operations, authorisation should be even stricter. A production restart or deployment should require authenticated human approval, not the agent’s own judgment.

Step 3: Put budgets around agent execution

An agent that can reason indefinitely is difficult to operate. Every production workflow should have limits around execution time, model calls, tool calls, and ideally token consumption.

Python
import os

class ExecutionBudget:
    def __init__(
        self,
        max_steps: int | None = None,
        max_tools: int | None = None,
    ):
        self.steps = 0
        self.tool_calls = 0

        self.max_steps = (
            max_steps
            if max_steps is not None
            else int(os.getenv("MAX_AGENT_STEPS", "5"))
        )

        self.max_tools = (
            max_tools
            if max_tools is not None
            else int(os.getenv("MAX_TOOL_CALLS", "8"))
        )

        if self.max_steps  None:
        if self.steps >= self.max_steps:
            raise RuntimeError(
                "Agent step budget exceeded."
            )

        self.steps += 1

    def check_tool_call(self) -> None:
        if self.tool_calls >= self.max_tools:
            raise RuntimeError(
                "Agent tool-call budget exceeded."
            )

        self.tool_calls += 1

Execution limits should be configurable rather than buried in application code. By passing these limits through the constructor, engineers can assign different budgets to different agents without changing global settings. A diagnostic agent might need only three steps, while a more complex research workflow may require considerably more. When no limits are supplied, the class falls back to environment variables. What matters is that every workflow has a defined ceiling, and that the same budget is enforced throughout its execution.

Step 4: Make model calls observable

An owner cannot operate what they cannot see. At minimum, record:

• request latency;
• model used;
• success/failure;
• tool calls;
• token usage where available;
• correlation or request ID;
• escalation events. A simple wrapper provides a useful boundary.

Python
import logging
import time
import uuid

logger = logging.getLogger(__name__)

def invoke_model(model, messages):
    request_id = str(uuid.uuid4())
    started = time.perf_counter()

    # Extract the model name across different provider libraries.
    model_name = (
        getattr(model, "model_name", None)
        or getattr(model, "model", None)
        or "unknown_model"
    )

    try:
        response = model.invoke(messages)

        elapsed = time.perf_counter() - started

        logger.info(
            "agent_model_call_success",
            extra={
                "request_id": request_id,
                "model": model_name,
                "latency_seconds": round(elapsed, 3),
            },
        )

        return response

    except Exception:
        elapsed = time.perf_counter() - started

        logger.exception(
            "agent_model_call_failed",
            extra={
                "request_id": request_id,
                "model": model_name,
                "latency_seconds": round(elapsed, 3),
            },
        )

        raise

The request ID becomes especially valuable when an incident involves several services. An engineer should be able to move from an alert to the agent request and then to the underlying model and tool calls. In production, supplement this logging layer with distributed tracing and metrics for a more complete view of system behaviour and performance.

Step 5: Separate failure from escalation

A failed model call does not necessarily mean the entire service should fail. Likewise, a successful model call does not mean the agent’s recommendation should execute automatically. These are different decisions.

Python
def handle_agent_failure(error: Exception) -> dict:
    logger.error(
        "Agent execution failed; escalating to human.",
        exc_info=(
            type(error),
            error,
            error.__traceback__,
        ),
    )

    return {
        "status": "escalated",
        "message": (
            "The AI agent could not complete the operation. "
            "Human intervention is required."
        ),
        "error_type": type(error).__name__,
    }

This distinction matters during a real incident. The agent might go down, but the engineering team should still be able to respond using the underlying workflows and runbooks. That is what graceful degradation looks like in practice.

Step 6: Make high-risk actions human-gated

One of the most useful patterns for enterprise agents is separating recommendation from execution. For example:

Telemetry
    ↓
AI Agent
    ↓
Diagnosis
    ↓
Recommended Action
    ↓
Risk Check
    ↓
Human Approval
    ↓
Production Change

The agent might recommend:

  “Increase the database connection pool.”

But it shouldn’t push that change to production just because the model generated the recommendation. Here’s what a simple approval boundary looks like:

Python
def execute_action(
    action: str,
    approved_by: str | None,
) -> dict:
    if not approved_by:
        return {
            "status": "awaiting_approval",
            "action": action,
        }

    return {
        "status": "approved",
        "action": action,
        "approved_by": approved_by,
    }

A real system needs a more sophisticated implementation, but the architectural boundary is what matters. “The agent can own reasoning without owning authorisation.” The agent can own reasoning without owning authorisation.

Step 7: Give the agent a health check

A production agent needs a health endpoint just like any other service. With FastAPI:

`Python
import os

import httpx
from fastapi import FastAPI, Response, status

app = FastAPI()

SERVICE_NAME = "incident-command-agent"

@app.get("/health/live")
async def liveness():
    """Check whether the application process is responsive."""
    return {
        "status": "alive",
        "service": SERVICE_NAME,
    }

async def check_provider() -> bool:
    """Check a configured provider health endpoint."""
    health_url = os.getenv("MODEL_PROVIDER_HEALTH_URL")

    if not health_url:
        return False

    try:
        async with httpx.AsyncClient(timeout=2.0) as client:
            result = await client.get(health_url)
            return result.status_code == 200
    except httpx.RequestError:
        return False

@app.get("/health/ready")
async def readiness(response: Response):
    """Check whether required dependencies are available."""
    if not await check_provider():
        response.status_code = (
            status.HTTP_503_SERVICE_UNAVAILABLE
        )
        return {
            "status": "unhealthy",
            "reason": "Model provider unavailable",
        }

    return {
        "status": "ready",
        "service": SERVICE_NAME,
    }

Liveness and readiness serve different purposes. The liveness endpoint confirms that the application process is responsive, while readiness checks whether the service can handle requests based on the availability of its required dependencies. In this example, the readiness endpoint checks a configured model-provider health URL and returns HTTP 503 when the check fails. The check is deliberately lightweight and time-limited to avoid introducing unnecessary delays or costs. In production, teams should adapt it to their provider’s supported health-check mechanism and other critical dependencies, such as a vector database or configuration service. Keeping these checks separate lets orchestration platforms such as Kubernetes distinguish between an application that needs restarting and one that can’t serve traffic temporarily.

Step 8: Fail closed when safety controls break

One of the worst failure modes is letting an agent keep running after its safety mechanisms go down. For example:

Python
def run_controlled_agent(
    tool_name: str,
    model,
    messages,
    budget: ExecutionBudget,
):
    try:
        authorize_tool(tool_name)

        # Reuse the budget across agent steps and tool calls.
        budget.check_step()
        budget.check_tool_call()

        return invoke_model(
            model,
            messages,
        )

    except ToolAuthorizationError:
        logger.exception(
            "Agent authorisation check failed."
        )

        return {
            "status": "blocked",
            "message": (
                "The requested operation is not authorised."
            ),
        }

    except RuntimeError:
        logger.exception(