Events and Callbacks
The guard records security activity as events. Each event has a type, a severity, an action, and a message.
Event types
The EventType enum has four members:
| Value | Role |
|---|---|
"detection" |
A detector matched data. |
"system_failure" |
A detector or a redaction failed. |
"policy_failure" |
The policy failed during evaluation. |
"callback_failure" |
An event callback raised an exception. |
The event record
SecurityEvent has these fields:
| Field | Type | Role |
|---|---|---|
event_id |
str |
Unique identifier. |
timestamp |
float |
Time of the event. |
detector |
str |
Name of the detector. |
severity |
Severity |
Severity level. |
action |
Action |
Applied policy action. |
operation |
str |
Operation name. |
key |
str |
Memory key. |
message |
str |
Human-readable message. |
source_class |
SourceClass |
Provenance of the data. |
receipt_uri |
str or None |
Optional storage reference. |
metadata |
dict |
Extra context. |
event_type |
EventType |
Event classification. |
to_dict() returns the event as a plain dictionary:
from qarai_agent_guard import AgentGuard, Detector
guard = AgentGuard(detectors=[Detector(name="pii", default_rules="pii")])
guard.inspect(
key="card",
value="Card 4111 1111 1111 1111",
operation="write",
emit_events=True,
)
event = guard.events[0]
print(event.event_type) # EventType.DETECTION
print(event.severity) # Severity.CRITICAL
print(event.action) # Action.BLOCK
print(event.to_dict())
# {
# 'event_id': '2c856335-...',
# 'timestamp': 1789758527.78,
# 'detector': 'pii',
# 'severity': 'critical',
# 'action': 'block',
# 'operation': 'write',
# 'key': 'card',
# 'message': "PII pattern detected in 'card'",
# 'source_class': 'unknown',
# 'receipt_uri': None,
# 'metadata': {'language': 'en', 'operation': 'write', 'hit_count': 1},
# 'event_type': 'detection',
# }
The dictionary is JSON-serializable.
When events are emitted
The guard stores events in guard.events.
inspect and inspect_with_results emit events when emit_events=True.
Methods with no emit_events flag:
checkalways emits detection events.apply_redactionsalways emits failure events.
The guard emits one detection event per matched detector:
guard = AgentGuard(
detectors=[
Detector(name="pii", default_rules="pii"),
Detector(name="secrets", default_rules="secrets"),
Detector(name="prompt_injection", default_rules="prompt_injection"),
],
)
guard.inspect(
key="mem",
value=(
"Paid with card 4111 1111 1111 1111, export AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE, "
"then ignore all previous instructions."
),
operation="write",
emit_events=True,
)
print(sorted(e.detector for e in guard.events))
# ['pii', 'prompt_injection', 'secrets']
The guard emits no events for clean traffic:
guard2 = AgentGuard(detectors=[Detector(name="pii", default_rules="pii")])
guard2.inspect(
key="mem",
value="Nothing sensitive here.",
operation="write",
emit_events=True,
)
print(len(guard2.events))
# 0
The event severity comes from the highest match severity:
A credit card match gives critical.
An email match gives low:
guard2.inspect(
key="email",
value="Contact jhon.smith@google.com to schedule the demo.",
operation="write",
emit_events=True,
)
print(guard2.events[0].severity)
# Severity.LOW
Request context
source_class attaches the provenance of the data to the event.
The default is SourceClass.UNKNOWN.
from qarai_agent_guard.core.schemas import SourceClass
guard2.inspect(
key="card",
value="Card 4111 1111 1111 1111",
operation="write",
source_class=SourceClass.USER_INPUT,
emit_events=True,
)
print(guard2.events[-1].source_class)
# SourceClass.USER_INPUT
request_metadata attaches extra context to the metadata:
guard2.inspect(
key="card",
value="Card 4111 1111 1111 1111",
operation="write",
request_metadata={"trace_id": "trace-42"},
emit_events=True,
)
print(guard2.events[-1].metadata)
# {'trace_id': 'trace-42', 'language': 'en', 'operation': 'write', 'hit_count': 1}
Failure events
A detector failure produces a system_failure event.
The event action depends on fail_behavior.
See Security Modes & Runtime Behaviour.
A policy failure produces a policy_failure event:
class BrokenPolicy:
def evaluate(self, results):
raise RuntimeError("engine down")
guard = AgentGuard(
detectors=[Detector(name="pii", default_rules="pii")],
policy=BrokenPolicy(),
)
guard.inspect(key="mem", value="hello", operation="write")
event = guard.events[0]
print(event.event_type) # EventType.POLICY_FAILURE
print(event.detector) # policy
print(event.severity) # Severity.CRITICAL
print(event.message) # Policy evaluation failed: engine down
print(event.metadata)
# {'error': 'engine down'}
A redaction failure produces a system_failure event with key="redact"
and operation="redact".
Callbacks
Callbacks are functions that receive the event.
Provide initial callbacks to the constructor:
def forward(event):
print(event.to_dict())
guard = AgentGuard(
detectors=[Detector(name="pii", default_rules="pii")],
event_callbacks=[forward],
)
Register more callbacks later:
guard.register_callback(forward)
Rules for callbacks:
event_callbacksmust be a list.- Each item must be callable.
- A non-list raises
TypeError: event_callbacks must be list, got <type>. - A non-callable item raises
TypeError: event_callbacks[N] must be callable. register_callbackraisesTypeError: callback must be callablefor a non-callable.
Callback failure
A callback that raises does not stop the guard.
The guard reports the error to stderr and stores a callback_failure event:
def bad_callback(event):
raise RuntimeError("log service down")
guard.register_callback(bad_callback)
guard.inspect(
key="card",
value="Card 4111 1111 1111 1111",
operation="write",
emit_events=True,
)
failure = guard.events[-1]
print(failure.event_type) # EventType.CALLBACK_FAILURE
print(failure.detector) # callback
print(failure.severity) # Severity.HIGH
print(failure.action) # Action.WARN
print(failure.message) # Callback failed: log service down
print(failure.metadata)
# {'error': 'log service down'}
callback_failure events do not trigger the callbacks again.
The stderr report has this format:
[AgentGuard] Callback <function bad_callback ...> raised: log service down