Skip to content

Ops Agents

OpsAgent

OpsAgent

Bases: ABC

Base class for operational agents.

Ops agents are registered with AgenticApp via register_ops_agent() and participate in the application lifecycle. They start when the app starts and stop when the app shuts down.

Subclasses must implement start(), stop(), and check_health().

Example

class MyOpsAgent(OpsAgent): async def start(self) -> None: self._running = True

async def stop(self) -> None:
    self._running = False

async def check_health(self) -> OpsHealthStatus:
    return OpsHealthStatus(healthy=self._running)

app = AgenticApp() app.register_ops_agent(MyOpsAgent(name="my-agent"))

Source code in src/agenticapi/ops/base.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
class OpsAgent(ABC):
    """Base class for operational agents.

    Ops agents are registered with AgenticApp via register_ops_agent()
    and participate in the application lifecycle. They start when the
    app starts and stop when the app shuts down.

    Subclasses must implement start(), stop(), and check_health().

    Example:
        class MyOpsAgent(OpsAgent):
            async def start(self) -> None:
                self._running = True

            async def stop(self) -> None:
                self._running = False

            async def check_health(self) -> OpsHealthStatus:
                return OpsHealthStatus(healthy=self._running)

        app = AgenticApp()
        app.register_ops_agent(MyOpsAgent(name="my-agent"))
    """

    def __init__(
        self,
        *,
        name: str,
        autonomy: AutonomyLevel = AutonomyLevel.SUPERVISED,
        max_severity: Severity = Severity.MEDIUM,
    ) -> None:
        """Initialize the ops agent.

        Args:
            name: Unique name identifying this ops agent.
            autonomy: The autonomy level for this agent's actions.
            max_severity: Maximum severity level the agent can handle autonomously.
        """
        self._name = name
        self._autonomy = autonomy
        self._max_severity = max_severity
        self._running = False

    @property
    def name(self) -> str:
        """The unique name of this ops agent."""
        return self._name

    @property
    def autonomy(self) -> AutonomyLevel:
        """The autonomy level for this agent."""
        return self._autonomy

    @property
    def max_severity(self) -> Severity:
        """Maximum severity level handled autonomously."""
        return self._max_severity

    @property
    def running(self) -> bool:
        """Whether the agent is currently running."""
        return self._running

    @abstractmethod
    async def start(self) -> None:
        """Start the ops agent.

        Called during application startup. Implementations should
        initialize background tasks, connect to monitoring systems, etc.
        """
        ...

    @abstractmethod
    async def stop(self) -> None:
        """Stop the ops agent.

        Called during application shutdown. Implementations should
        clean up resources and cancel background tasks.
        """
        ...

    @abstractmethod
    async def check_health(self) -> OpsHealthStatus:
        """Check the health of this ops agent.

        Returns:
            The current health status.
        """
        ...

    def can_handle_autonomously(self, severity: Severity) -> bool:
        """Check if the agent can handle an issue at the given severity autonomously.

        Args:
            severity: The severity of the issue.

        Returns:
            True if the agent can act without human approval.
        """
        if self._autonomy == AutonomyLevel.MANUAL:
            return False
        if self._autonomy == AutonomyLevel.AUTO:
            return True
        # SUPERVISED: compare severity levels
        severity_order = [Severity.LOW, Severity.MEDIUM, Severity.HIGH, Severity.CRITICAL]
        return severity_order.index(severity) <= severity_order.index(self._max_severity)

name property 

name: str

The unique name of this ops agent.

autonomy property 

autonomy: AutonomyLevel

The autonomy level for this agent.

max_severity property 

max_severity: Severity

Maximum severity level handled autonomously.

running property 

running: bool

Whether the agent is currently running.

__init__ 

__init__(
    *,
    name: str,
    autonomy: AutonomyLevel = AutonomyLevel.SUPERVISED,
    max_severity: Severity = Severity.MEDIUM,
) -> None

Initialize the ops agent.

Parameters:

Name Type Description Default
name str

Unique name identifying this ops agent.

 required
autonomy AutonomyLevel

The autonomy level for this agent's actions.

 SUPERVISED
max_severity Severity

Maximum severity level the agent can handle autonomously.

 MEDIUM
Source code in src/agenticapi/ops/base.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    *,
    name: str,
    autonomy: AutonomyLevel = AutonomyLevel.SUPERVISED,
    max_severity: Severity = Severity.MEDIUM,
) -> None:
    """Initialize the ops agent.

    Args:
        name: Unique name identifying this ops agent.
        autonomy: The autonomy level for this agent's actions.
        max_severity: Maximum severity level the agent can handle autonomously.
    """
    self._name = name
    self._autonomy = autonomy
    self._max_severity = max_severity
    self._running = False

start abstractmethod async 

start() -> None

Start the ops agent.

Called during application startup. Implementations should initialize background tasks, connect to monitoring systems, etc.

Source code in src/agenticapi/ops/base.py 
83
84
85
86
87
88
89
90
@abstractmethod
async def start(self) -> None:
    """Start the ops agent.

    Called during application startup. Implementations should
    initialize background tasks, connect to monitoring systems, etc.
    """
    ...

stop abstractmethod async 

stop() -> None

Stop the ops agent.

Called during application shutdown. Implementations should clean up resources and cancel background tasks.

Source code in src/agenticapi/ops/base.py 
92
93
94
95
96
97
98
99
@abstractmethod
async def stop(self) -> None:
    """Stop the ops agent.

    Called during application shutdown. Implementations should
    clean up resources and cancel background tasks.
    """
    ...

check_health abstractmethod async 

check_health() -> OpsHealthStatus

Check the health of this ops agent.

Returns:

Type Description
OpsHealthStatus

The current health status.
Source code in src/agenticapi/ops/base.py 
101
102
103
104
105
106
107
108
@abstractmethod
async def check_health(self) -> OpsHealthStatus:
    """Check the health of this ops agent.

    Returns:
        The current health status.
    """
    ...

can_handle_autonomously 

can_handle_autonomously(severity: Severity) -> bool

Check if the agent can handle an issue at the given severity autonomously.

Parameters:

Name Type Description Default
severity Severity

The severity of the issue.

 required

Returns:

Type Description
bool

True if the agent can act without human approval.
Source code in src/agenticapi/ops/base.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def can_handle_autonomously(self, severity: Severity) -> bool:
    """Check if the agent can handle an issue at the given severity autonomously.

    Args:
        severity: The severity of the issue.

    Returns:
        True if the agent can act without human approval.
    """
    if self._autonomy == AutonomyLevel.MANUAL:
        return False
    if self._autonomy == AutonomyLevel.AUTO:
        return True
    # SUPERVISED: compare severity levels
    severity_order = [Severity.LOW, Severity.MEDIUM, Severity.HIGH, Severity.CRITICAL]
    return severity_order.index(severity) <= severity_order.index(self._max_severity)

OpsHealthStatus

OpsHealthStatus

Health status of an ops agent.

Attributes:

Name Type Description
healthy

Whether the agent is operating normally.
message

Optional human-readable status message.
details

Optional additional details.
Source code in src/agenticapi/ops/base.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class OpsHealthStatus:
    """Health status of an ops agent.

    Attributes:
        healthy: Whether the agent is operating normally.
        message: Optional human-readable status message.
        details: Optional additional details.
    """

    def __init__(
        self,
        *,
        healthy: bool = True,
        message: str = "",
        details: dict[str, Any] | None = None,
    ) -> None:
        self.healthy = healthy
        self.message = message
        self.details = details or {}