Skip to content

Documentation navigation

Events & Audit

Audit events, sinks, and webhook signature verification.

#Events and audit

When an Agent uses a tool to write to Salesforce or delete a file, you want a record of it — who did what, to which object, and whether it succeeded. And when a provider calls back into your system over a webhook, you need to prove the request is genuine before acting on it. maivn-tools ships two event surfaces for exactly these two directions:

  • Audit events describe operations connector tools perform (outbound).
  • Webhook helpers verify inbound deliveries from providers (inbound).

Both are lightweight and optional: you choose where audit events go by plugging in a sink, and verification is a single call. Audit detail is application-supplied and is not automatically scrubbed.

#Audit events

Python
from maivn_tools import (
    AuditEvent,
    AuditEventKind,
    AuditEventSeverity,
    InMemoryAuditSink,
)

sink = InMemoryAuditSink()
sink.emit(
    AuditEvent(
        kind=AuditEventKind.WRITE,
        provider="example",
        tool="create_post",
        target="post:123",
        severity=AuditEventSeverity.NOTICE,
        detail={"size": 4096},
    )
)
for event in sink.events:
    print(event.to_dict())
FieldNotes
kindOne of the normalized AuditEventKind values. Map provider events onto these so dashboards stay consistent.
providerMust match ProviderMetadata.name.
toolTool name as registered on the Agent.
connection_idUseful when a host manages many connections.
actorPrincipal that initiated the call (user id, agent id).
targetURL, ID, or path of the affected object.
detailFree-form, JSON-serializable map. Never include secrets.

#Custom sinks

AuditSink has a single emit method. Implementations should be exception- safe; failure to log audit events must not crash the connector.

Python
from maivn_tools import AuditEvent, AuditSink


class OtelAuditSink(AuditSink):
    def __init__(self, tracer):
        self._tracer = tracer

    def emit(self, event: AuditEvent) -> None:
        try:
            self._tracer.add_event(event.to_dict())
        except Exception:
            pass

#Webhook signature verification

WebhookVerifier covers the HMAC patterns most providers use.

Python
from maivn_tools import SignatureAlgorithm, WebhookVerifier

verifier = WebhookVerifier(
    secret="shared-secret",
    signature_header="X-Example-Signature",
    algorithm=SignatureAlgorithm.HMAC_SHA256,
    encoding="hex",
)
verifier.verify(request_headers, request_body)

A SignatureMismatchError is raised when the signature is missing, malformed, or wrong, or when the optional timestamp falls outside the configured tolerance window.

This example expects the bare hex digest, not GitHub's sha256= prefix or Stripe's timestamped signature format. The optional timestamp header is checked separately from the HMAC and cannot establish authenticated freshness. Implement the provider's signed-message and deduplication rules.

For one-off verification without building a verifier, call verify_hmac_signature(secret, payload, signature, algorithm=..., encoding=...) directly.

#Supported algorithms

SignatureAlgorithm.HMAC_SHA1, SignatureAlgorithm.HMAC_SHA256, and SignatureAlgorithm.HMAC_SHA512.

#Supported encodings

"hex" (default) and "base64". Provider documentation typically states which one to expect.

For higher-level normalization into a NormalizedWebhookEvent, see the WebhookListener adapter.