Pull Moderation Logs From The API

Read moderation logs from the logs API when you need history from before your collector started, or when you cannot run a collector on the DynamoGuard cluster. For live collection, send the API's standard output to your SIEM instead. The reference poller script is under Downloads.

Scope

Describes release 3.26.

Choose An Endpoint

Use the applications endpoint with a dedicated service user that holds the organization role org:dynamoguard:admin. The endpoint returns 403 to users without that role.

The AI system endpoint reads one AI system. Access checks on it are tightened in an upcoming 3.26 patch. Prefer the applications endpoint until you run a release that carries the change.

Endpoint	Returns	Count endpoint response
GET /v1/applications/APPLICATION_ID/logs	Every AI system attached to a project (projects are called applications in the API). Filter with aiSystemIds.	GET .../logs/count returns a bare number
GET /v1/moderation/model/AI_SYSTEM_ID/logs	One AI system	GET .../logs/count returns {"count": N}

Both endpoints read the moderation log tables in the DynamoAI PostgreSQL database.

Attach AI Systems To The Project

The applications endpoint returns logs for the AI systems attached to the project, including logs written before an AI system was attached. Attach AI systems after you create the project:

curl -sS -X POST \
  -H "Authorization: Bearer TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"addAiSystems":["AI_SYSTEM_ID"]}' \
  https://API_HOST/v1/applications/APPLICATION_ID/assign-ai-systems

The response is 207 Multi-Status, with one result per AI system. POST /v1/applications ignores an aiSystems field. Use this call to attach AI systems.

Create A Service User And Token

1. Create a dedicated DynamoAI user for log export and grant it the organization role org:dynamoguard:admin.
2. Sign in as that user and generate a token, as described in Access Tokens.
3. Store the token in your secret store. Send it as Authorization: Bearer TOKEN.

Each user has one token, and generating a new token replaces the old one. Keep the export on its own user. If it shares a person's account, that person generating a token stops the export.

Query Parameters

Parameter	Required	Description
startTime	Yes	Window start, UTC seconds. Inclusive.
endTime	Yes	Window end, UTC seconds. Inclusive.
sortDir	No	asc or desc. Default desc.
pageNum	No	Page number, from 1. Default 1.
perPage	No	Logs per page. Default 50. 0 returns every log in the window.
actions	No	JSON array of final actions, for example ["BLOCK","WARN"]
aiSystemIds	No	Applications endpoint only. JSON array of AI system IDs.

Verify access and the time window:

curl -sS -H "Authorization: Bearer TOKEN" \
  "https://API_HOST/v1/applications/APPLICATION_ID/logs/count?startTime=START_UTC_SECONDS&endTime=END_UTC_SECONDS"

Response Fields

Each entry in logs has id, timestamp (ISO 8601, UTC), action (the most severe action of the analyses), analyses, metadata, and multiturnMetadata, plus ragContext when the request included one. Each analysis has text, textType, and appliedPolicies; each applied policy has policy, action, outputs, feedback, and promptId. The applications endpoint adds applicationAiSystemId, applicationAiSystemName, and applicationAiSystemRemovedAt.

Limits To Plan For

• Text is truncated to 2000 characters per field. Fetch the full text of one log with GET /v1/moderation/log/LOG_ID/full-prompts.
• Retention is three months by default, and fixed at install. Older logs move out of the tables the API reads when the database's partition maintenance runs. MODERATION_LOGS_RETENTION_PERIOD is read once, while the moderation log tables are created; setting it on a running deployment changes nothing. Read the value your deployment uses before you size a backfill window: SELECT parent_table, retention FROM part_config WHERE parent_table LIKE 'public.moderation_logs%';
• Order is by timestamp only, and both window ends are inclusive. Logs with equal timestamps can change order between pages, and a log can become visible shortly after later logs. Re-read an overlap window on every run and remove duplicates by id.
• The API reads the database. A request whose database save failed returns an error to the caller. It has a standard output line but no API record.
• A project with no attached AI systems returns logs from every AI system. On release 3.26, once every AI system in a project was removed before the start of the window, the applications endpoint drops the AI system filter. Export from a project that has at least one attached AI system, or read each AI system with the AI system endpoint. The reference poller refuses to run against a project with no attached AI systems.

Run The Reference Poller

pull_moderation_logs.py is a reference sample. DynamoAI does not support it as a product. It uses only the Python 3.9+ standard library and writes one log per line (NDJSON) to standard output, in the shape the API returns.

export DYNAMOAI_API_KEY=TOKEN
python3 pull_moderation_logs.py \
  --base-url https://API_HOST \
  --application-id APPLICATION_ID \
  --since 2026-06-01T00:00:00Z >> moderation-logs.ndjson

Run it on a schedule, for example every five minutes. Point your SIEM's file input at the output file.

Option	Default	Description
--application-id or --ai-system-id	Required	Endpoint to read
--since	Now minus --overlap	First run only: where to start. ISO 8601 or UTC seconds.
--state-file	moderation_logs_state.json	Last position and recently written IDs. Delete it to start over.
--overlap	300	Seconds re-read on every run, to catch logs that became visible late
--lag	30	Seconds before now at which each run stops
--window	3600	Largest time window per request, in seconds
--max-batch	1000	A window holding more logs than this is split in half

How it reads:

1. It counts the logs in each window and splits any window above --max-batch, then reads the whole window with perPage=0. It never pages by offset, and equal timestamps cannot shift logs between pages.
2. It skips IDs it wrote during the overlap and saves its state after each window.
3. Delivery is at least once: a crash between writing and saving can repeat a window. Remove duplicates by id in your SIEM, for example | dedup id in Splunk.

Source of pull_moderation_logs.py

#!/usr/bin/env python3
"""Reference sample: pull DynamoGuard moderation logs from the DynamoAI API as NDJSON.

Not a supported product. Behaviour and limits: "Pull Moderation Logs From The API" in the DynamoAI docs.
"""
import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone

RETRYABLE_STATUS = {429, 500, 502, 503, 504}


def parse_args(argv):
    parser = argparse.ArgumentParser(
        description="Pull DynamoGuard moderation logs and write them to stdout as NDJSON."
    )
    parser.add_argument("--base-url", required=True, help="API base URL, for example https://api.example.com")
    target = parser.add_mutually_exclusive_group(required=True)
    target.add_argument("--application-id", type=int, help="Read every AI system in this application (project)")
    target.add_argument("--ai-system-id", help="Read one AI system")
    parser.add_argument("--state-file", default="moderation_logs_state.json")
    parser.add_argument("--since", help="First run only: ISO 8601 time or UTC seconds to start from")
    parser.add_argument("--overlap", type=float, default=300.0, help="Seconds re-read on every run (default 300)")
    parser.add_argument("--lag", type=float, default=30.0, help="Stop this many seconds before now (default 30)")
    parser.add_argument("--window", type=float, default=3600.0, help="Largest window per request, seconds (default 3600)")
    parser.add_argument("--max-batch", type=int, default=1000, help="Split windows holding more logs (default 1000)")
    parser.add_argument("--timeout", type=float, default=60.0)
    parser.add_argument("--retries", type=int, default=5)
    return parser.parse_args(argv)


def to_seconds(value):
    try:
        return float(value)
    except ValueError:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed.timestamp()


class Client:
    def __init__(self, args, token):
        self.base = args.base_url.rstrip("/")
        if args.application_id is not None:
            self.logs_url = f"{self.base}/v1/applications/{args.application_id}/logs"
        else:
            self.logs_url = f"{self.base}/v1/moderation/model/{urllib.parse.quote(args.ai_system_id, safe='')}/logs"
        self.token = token
        self.timeout = args.timeout
        self.retries = args.retries
        self.max_batch = args.max_batch

    def get(self, url, params=None):
        query = f"?{urllib.parse.urlencode(params)}" if params else ""
        request = urllib.request.Request(
            f"{url}{query}",
            headers={"Authorization": f"Bearer {self.token}", "Accept": "application/json"},
        )
        delay = 1.0
        for attempt in range(self.retries + 1):
            try:
                with urllib.request.urlopen(request, timeout=self.timeout) as response:
                    return json.load(response)
            except urllib.error.HTTPError as err:
                if err.code not in RETRYABLE_STATUS or attempt == self.retries:
                    raise
                retry_after = err.headers.get("Retry-After", "")
                wait = float(retry_after) if retry_after.isdigit() else delay
            except urllib.error.URLError:
                if attempt == self.retries:
                    raise
                wait = delay
            time.sleep(wait)
            delay = min(delay * 2, 60.0)
        raise RuntimeError("unreachable")

    def attached_ai_systems(self, application_id):
        return self.get(f"{self.base}/v1/applications/{application_id}").get("aiSystems") or []

    @staticmethod
    def window_params(start, end):
        return {"startTime": f"{start:.3f}", "endTime": f"{end:.3f}", "sortDir": "asc"}

    def count(self, start, end):
        body = self.get(f"{self.logs_url}/count", self.window_params(start, end))
        # The applications endpoint returns a bare number; the AI system endpoint returns {"count": n}.
        return body["count"] if isinstance(body, dict) else int(body)

    def fetch(self, start, end):
        total = self.count(start, end)
        if total == 0:
            return []
        if total > self.max_batch and end - start > 1.0:
            middle = (start + end) / 2
            return self.fetch(start, middle) + self.fetch(middle, end)
        params = dict(self.window_params(start, end), perPage=0)
        return self.get(self.logs_url, params)["logs"]


def load_state(path):
    if not os.path.exists(path):
        return {"position": None, "seen": {}}
    with open(path, encoding="utf-8") as handle:
        return json.load(handle)


def save_state(path, state):
    temporary = f"{path}.tmp"
    with open(temporary, "w", encoding="utf-8") as handle:
        json.dump(state, handle)
    os.replace(temporary, path)


def main(argv):
    args = parse_args(argv)
    token = os.environ.get("DYNAMOAI_API_KEY")
    if not token:
        sys.exit("Set DYNAMOAI_API_KEY to the service user's API key.")
    client = Client(args, token)
    try:
        if args.application_id is not None and not client.attached_ai_systems(args.application_id):
            sys.exit(
                f"Project {args.application_id} has no attached AI systems. On release 3.26 its logs endpoint "
                "then returns logs from every AI system. Attach an AI system, or use --ai-system-id."
            )
    except urllib.error.HTTPError as err:
        sys.exit(f"HTTP {err.code} from {err.url}: {err.read().decode(errors='replace')[:300]}")
    state = load_state(args.state_file)
    seen = state["seen"]

    stop = time.time() - args.lag
    if state["position"] is not None:
        start = state["position"] - args.overlap
    elif args.since:
        start = to_seconds(args.since)
    else:
        start = stop - args.overlap

    written = 0
    while start < stop:
        end = min(start + args.window, stop)
        try:
            logs = client.fetch(start, end)
        except urllib.error.HTTPError as err:
            sys.exit(f"HTTP {err.code} from {err.url}: {err.read().decode(errors='replace')[:300]}")
        for record in sorted(logs, key=lambda item: (item["timestamp"], item["id"])):
            if record["id"] in seen:
                continue
            sys.stdout.write(json.dumps(record, separators=(",", ":")) + "\n")
            seen[record["id"]] = to_seconds(record["timestamp"])
            written += 1
        sys.stdout.flush()
        horizon = end - args.overlap
        state = {"position": end, "seen": {key: ts for key, ts in seen.items() if ts >= horizon}}
        seen = state["seen"]
        save_state(args.state_file, state)
        start = end

    print(f"wrote {written} new logs", file=sys.stderr)


if __name__ == "__main__":
    main(sys.argv[1:])

Downloads

File	Description
pull_moderation_logs.py	Reference poller. Python 3.9 or later, standard library only, no install step. See Run The Reference Poller.