# The audit log Python API

Query the audit log from your services with self.audit - search events, find silent feeds and read stored payloads.

Every service has `self.audit` - the same [audit log](https://zato.io/docs/admin/audit-log/index.html) the Dashboard shows, queried from Python. A service that confirms a message arrived, lists last night's failures or reports feed silence uses the same API for every supported [audit-log database](https://zato.io/docs/admin/audit-log/index.html#storage-and-configuration).

The API is read-only and has three methods:

| Method | What it returns |
| --- | --- |
| `self.audit.search(...)` | Which events match - by free text, source, object, outcome, event type or time window |
| `self.audit.last_seen(source)` | The newest event time of each object of one source |
| `self.audit.get_payload(event_id)` | The full message body of one event |

## Searching events {#searching-events}

`self.audit.search` returns a list of dicts, newest first. Every parameter is optional and they combine - each one you pass narrows the result down further:

```python
# -*- coding: utf-8 -*-

# Zato
from zato.common.audit_log.api import AuditOutcome
from zato.server.service import Service

class RecentFailures(Service):
    """ Returns the failures recorded since a given moment.
    """
    input = 'time_from'

    def handle(self) -> 'None':

        # Search for failures recorded in the requested period ..
        events = self.audit.search(
            outcome=AuditOutcome.Error,
            time_from=self.request.input.time_from,
        )

        # .. and return the matching events.
        self.response.payload = {'events': events}
```

The parameters:

| Parameter | Default | Description |
| --- | --- | --- |
| `query` | (none) | Free text to search for - see [what the free text covers](https://zato.io/docs/admin/audit-log/python.html#what-the-free-text-covers) |
| `source` | all | Which part of the log to read - an [AuditSource constant](https://zato.io/docs/admin/audit-log/python.html#sources), e.g. `AuditSource.MLLP_Channel` |
| `object_name` | all | An object's name, e.g. the name of a specific channel or connection |
| `outcome` | all | `AuditOutcome.OK`, `AuditOutcome.Error` or `AuditOutcome.Expired` |
| `event_type` | all | An `AuditEvent` constant, e.g. `AuditEvent.Message_Received` |
| `time_from` | (none) | Only events at or after this ISO timestamp |
| `time_to` | (none) | Only events before this ISO timestamp |
| `status` | (none) | `Status_Outstanding` returns [open exchanges](https://zato.io/docs/admin/audit-log/python.html#exchanges-still-waiting) - the source must be named too |
| `page` | `1` | Which page of results to return |
| `page_size` | `50` | How many events one page holds |

Each of `source`, `object_name`, `outcome` and `event_type` accepts one value or a list of values - a list matches events with any of them:

```python
# Search for failures and expirations across both MLLP directions.
events = self.audit.search(
    source=[AuditSource.MLLP_Channel, AuditSource.MLLP_Outgoing],
    outcome=[AuditOutcome.Error, AuditOutcome.Expired],
)
```

All the constants come from one module:

```python
# Zato
from zato.common.audit_log.api import AuditEvent, AuditOutcome, AuditSource, Status_Outstanding
```

Event times are ISO timestamps compared as strings, so a bare date works as a prefix - `time_from='2026-08-29'` matches everything from that day on.

Each returned event is a dict with these keys:

| Key | Description |
| --- | --- |
| `id` | The event's ID - what `get_payload` takes |
| `cid` | The correlation ID shared by all events of one message |
| `source` | The [source](https://zato.io/docs/admin/audit-log/python.html#sources) the event is filed under |
| `event_type` | What happened, e.g. `message-received` or `ack-sent` |
| `object_name` | The channel or connection the event belongs to |
| `msg_id` | The message's own ID, e.g. the MSH-10 control ID of an HL7 message |
| `correl_id` | The ID of the message this one answers or resubmits |
| `ext_client_id` | The ID the external caller sent, if any |
| `event_time_iso` | When the event happened, as an ISO timestamp |
| `endpoint` | The address the message arrived from or was sent to |
| `size` | The size of the data, in bytes |
| `outcome` | `ok`, `error` or `expired` |
| `classification` | For failures - `transient`, `permanent` or `operator-fixable` |
| `status` | The error text or other status details |
| `duration_ms` | How long the exchange took, in milliseconds |
| `data` | A preview of the payload - `get_payload` returns the full payload |

## What the free text covers {#what-the-free-text-covers}

The `query` parameter looks inside payloads, message IDs, correlation IDs, endpoints, external client IDs, statuses, classifications and event types. Wildcards match literally - a `%` in the query matches a literal percent sign.

When the search names a source, it also covers that source's own searchable attributes - an MLLP source's message type, MRN, facility and ACK status, a FHIR source's resource type and method, a scheduler source's job ID. An attribute match returns the whole trace it appears in, not just the single event that has it - a search by an MRN returns the message that mentioned the MRN together with the ACK that answered it:

```python
# Search for arrivals and acknowledgments of one patient's messages.
events = self.audit.search(
    source=AuditSource.MLLP_Channel,
    query='12345678',
)
```

This is the same search the Dashboard's search box runs, described on [audit log views](https://zato.io/docs/admin/audit-log/views.html).

## Exchanges still waiting {#exchanges-still-waiting}

`status=Status_Outstanding` returns the open exchanges of one source. For outgoing MLLP, these are messages awaiting acknowledgment. Each source defines what counts as open, so this filter requires the source parameter:

```python
# Search for outgoing HL7 messages awaiting an acknowledgment.
events = self.audit.search(
    source=AuditSource.MLLP_Outgoing,
    status=Status_Outstanding,
)
```

## When each channel was last seen {#when-each-channel-was-last-seen}

`self.audit.last_seen` takes one source and returns a dict mapping each of its object names to the ISO timestamp of its newest event. Comparing these times with a threshold identifies silent feeds:

```python
# -*- coding: utf-8 -*-

# stdlib
from datetime import datetime, timedelta

# Zato
from zato.common.audit_log.api import AuditSource
from zato.server.service import Service

class SilentFeeds(Service):
    """ Returns the HL7 channels whose latest event is older than one hour.
    """
    output = 'silent'

    def handle(self) -> 'None':

        # Get the newest event time of each channel ..
        last_seen = self.audit.last_seen(AuditSource.MLLP_Channel)

        # .. calculate the one-hour silence threshold ..
        cutoff = datetime.utcnow() - timedelta(hours=1)
        cutoff_iso = cutoff.isoformat()

        # .. collect channels whose latest event precedes the threshold ..
        silent = []

        for channel_name, last_event_time in last_seen.items():
            if last_event_time < cutoff_iso:
                silent.append(channel_name)

        # .. and return their names.
        self.response.payload = {'silent': silent}
```

## Reading a payload {#reading-a-payload}

The event's `data` key is a preview. `self.audit.get_payload` returns the whole body of one event, by the event's `id`:

```python
# Zato
from zato.common.audit_log.api import AuditBody

# Read the latest stored body of an event.
payload = self.audit.get_payload(event_id)

# Select a specific stored body.
request_body = self.audit.get_payload(event_id, AuditBody.Request)
response_body = self.audit.get_payload(event_id, AuditBody.Response)
error_body = self.audit.get_payload(event_id, AuditBody.Error)
```

The method returns `None` for a missing event, a payload-free source such as the [MCP audit log](https://zato.io/docs/ai/mcp/audit-log.html), or an event whose [payload retention period](https://zato.io/docs/admin/audit-log/index.html#keeping-payloads-for-less-time-than-events) has ended.

Payloads are the sensitive part of the record - in clinical traffic they are protected health information - so return them from a service only when the caller needs the message body itself, not just the metadata about it.

## Sources {#sources}

The `source` parameter of `search` and `last_seen` names which part of the log to read, always through `AuditSource` constants - `search` accepts one or a list of them, `last_seen` takes one. These are the most commonly queried ones - the full catalog of what each source records is on the [audit log overview](https://zato.io/docs/admin/audit-log/index.html):

| Constant | Events of |
| --- | --- |
| `AuditSource.REST_Channel` | REST channels |
| `AuditSource.REST_Outgoing` | Outgoing REST connections |
| `AuditSource.SOAP_Channel` | SOAP channels |
| `AuditSource.SOAP_Outgoing` | Outgoing SOAP connections |
| `AuditSource.PubSub` | Publish/subscribe topics |
| `AuditSource.Email_IMAP` | IMAP connections |
| `AuditSource.Email_SMTP` | SMTP connections |
| `AuditSource.File_Outgoing` | File transfer connections |
| `AuditSource.SQL_Outgoing` | Outgoing SQL connections |
| `AuditSource.MLLP_Channel` | HL7 MLLP channels |
| `AuditSource.MLLP_Outgoing` | Outgoing HL7 MLLP connections |
| `AuditSource.FHIR` | Outgoing FHIR connections |
| `AuditSource.MCP` | MCP gateways |
| `AuditSource.LLM` | LLM connections |
| `AuditSource.Scheduler` | Scheduler jobs |
| `AuditSource.Config` | Configuration changes and payload access |

## A complete example {#a-complete-example}

One service that answers whether a patient's messages arrived, ready to be [exposed to AI agents as an MCP tool](https://zato.io/docs/dev/healthcare/ai/agents.html):

```python
# -*- coding: utf-8 -*-

# Zato
from zato.common.audit_log.api import AuditEvent, AuditSource
from zato.server.service import Service

# #####################################################################
# #####################################################################

class PatientMessageStatus(Service):
    """ Returns whether HL7 messages for a given patient MRN arrived,
    with the control ID, event type and arrival time of each.
    """
    name = 'patient.message-status'

    input  = 'patient_mrn'
    output = 'has_arrived', 'messages'

    def handle(self) -> 'None':

        # Read the MRN the caller asked about ..
        patient_mrn = self.request.input.patient_mrn

        # .. search for received MLLP messages that mention it ..
        events = self.audit.search(
            source=AuditSource.MLLP_Channel,
            event_type=AuditEvent.Message_Received,
            query=patient_mrn,
        )

        # .. copy only the fields needed by the caller ..
        messages = []

        for event in events:
            messages.append({
                'control_id': event['msg_id'],
                'event_type': event['event_type'],
                'received_at': event['event_time_iso'],
            })

        # .. determine whether at least one message arrived ..
        has_arrived = len(messages) > 0

        # .. and return the minimized result.
        self.response.payload = {
            'has_arrived': has_arrived,
            'messages': messages,
        }
```

## Learn more {#learn-more}

- [Production checklist](https://zato.io/docs/admin/production-checklist.html) - What to confirm before an environment takes real traffic
- [Audit log](https://zato.io/docs/admin/audit-log/) - Every request, response and error, and what you can do with them afterwards
- [Health checks](https://zato.io/docs/admin/health-checks.html) - Have Zato ping your outgoing connections instead of waiting for traffic to fail
- [Logging](https://zato.io/docs/admin/logging.html) - What is written where, and how to ship it to Splunk, Loki, Elastic or Datadog
- [GitOps with enmasse](https://zato.io/docs/admin/enmasse.html) - Keep the whole environment in git as YAML and import it anywhere