> This page is for Agent Handler. > For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://docs.merge.dev/llms.txt. > For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.merge.dev/_mcp/server. # Stream logs to a SIEM > Set up your SIEM or observability platform to retrieve Agent Handler logs from the public log endpoints, with verified step-by-step instructions per platform. Agent Handler exposes two cursor-paginated, pull-based log endpoints so your SIEM or observability platform can retrieve logs on a schedule, without you building a bespoke integration: * **Tool Call logs** — `GET /api/v1/logs/tool-calls/` (see [Tool Call Logs](/merge-agent-handler/observe/tool-call-logs)) * **API Request logs** — `GET /api/v1/logs/api-calls/` (see [API Request Logs](/merge-agent-handler/observe/api-request-logs)) Full request/response schemas are in the [API reference](/merge-agent-handler/agent-handler). > **Note** > > The log endpoints are an **Enterprise** feature. If they're not enabled for your organization, the endpoints return `403`. Contact your account team to enable them. ## How the endpoints work Both endpoints behave identically: | Aspect | Detail | | -------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Auth** | `Authorization: Bearer ` — a production API key from [Settings → API Keys](https://ah.merge.dev/settings/api-keys). | | **Order** | Oldest-first, so you can tail the log forward without gaps. | | **Pagination** | Cursor-based. Each response returns `next_cursor`; pass it back as `?cursor=<…>` for the next page. Stop when `has_more` is `false` (then `next_cursor` is `null`). Persist `next_cursor` between polls to resume exactly where you left off. | | **Filters** | `page_size` (1–1000, default 100), `created_after` / `created_before` (ISO 8601), plus `tool_name` / `status` / `connector_id` (tool calls) and `method` / `response_status` / `connector_id` / `toolcall_id` (API requests). | | **Retention** | Only the **last 30 days** are available. If a collector is down longer than 30 days, the oldest records age out and can't be recovered — alert on collector staleness. | | **Redaction** | Credential values in headers (`Authorization`, API keys, cookies) and secret-named body fields are redacted before they leave the platform. | A poll loop looks like this: ```bash # First poll: start from the beginning of the retention window. curl -s "https://ah-api.merge.dev/api/v1/logs/tool-calls/?page_size=1000&created_after=2026-05-13T00:00:00Z" \ -H "Authorization: Bearer $PRODUCTION_KEY" # → {"results": [ ... ], "next_cursor": "eyJ0…", "has_more": true} # Next polls: pass the cursor from the previous response. curl -s "https://ah-api.merge.dev/api/v1/logs/tool-calls/?page_size=1000&cursor=eyJ0…" \ -H "Authorization: Bearer $PRODUCTION_KEY" # Repeat until "has_more": false, persist the last next_cursor, resume later from it. ``` ## Native pull vs. forwarder Platforms fall into two camps: * **Native pull** — the platform has a built-in REST/API poller that follows cursor pagination for you. Just configure it. **Microsoft Sentinel, Elastic, Cribl Stream, and Sumo Logic** are in this camp. * **Forwarder** — the platform is push-only (it expects you to *send* logs to its intake), so you run a small scheduled job that walks the cursor loop above and posts batches to the platform. **Datadog and Splunk** are most reliably integrated this way. Either way, the source side is identical: page through the endpoint with `cursor`, persist `next_cursor`, stop on `has_more: false`. ## Platform setup #### Datadog **Forwarder.** Datadog log collection is push-based — there's no native input that follows cursor pagination against an arbitrary REST API. Run a scheduled job (Lambda, container, cron, or a Datadog Agent custom check) that pulls from the endpoint and POSTs to the Datadog Logs intake API. 1. Create a Datadog **API key** (Organization Settings → API Keys). This is the `DD-API-KEY` header value — an API key, *not* an application key. 2. Pick the intake host for **your Datadog site** (mismatched site is the most common failure): * US1 — `https://http-intake.logs.datadoghq.com` * US3 — `https://http-intake.logs.us3.datadoghq.com` * US5 — `https://http-intake.logs.us5.datadoghq.com` * EU1 — `https://http-intake.logs.datadoghq.eu` * AP1 — `https://http-intake.logs.ap1.datadoghq.com` 3. Store the Agent Handler `PRODUCTION_KEY` and the Datadog API key in a secret manager, and persist the cursor in durable storage (DynamoDB, SSM, Redis, a DB row). 4. Implement the pull→push loop and schedule it (e.g. every 1–5 minutes). Persist `next_cursor` **after** a page ships successfully so a crash re-ships at most one page. ```python import requests cursor = state.load() # None on first run ah = {"Authorization": f"Bearer {AH_PRODUCTION_KEY}"} dd_url = "https://http-intake.logs.datadoghq.com/api/v2/logs" # match your site dd = {"Content-Type": "application/json", "DD-API-KEY": DD_API_KEY} while True: params = {"page_size": 1000} if cursor: params["cursor"] = cursor else: params["created_after"] = "2026-05-13T00:00:00Z" # within the 30-day window body = requests.get( "https://ah-api.merge.dev/api/v1/logs/tool-calls/", headers=ah, params=params, timeout=30, ).json() batch = [ {"ddsource": "agent-handler", "service": "ah-tool-calls", "ddtags": "env:prod", "message": rec} for rec in body["results"] ] if batch: # Chunk conservatively (≤ ~500 logs / ≤ ~4 MB per POST). requests.post(dd_url, headers=dd, json=batch, timeout=30).raise_for_status() cursor = body["next_cursor"] state.save(cursor) if not body["has_more"]: break # next scheduled run resumes from the saved cursor ``` **Gotchas** * The intake auth header is `DD-API-KEY` with an API key — not `Authorization: Bearer`. The Bearer token is only for the Agent Handler side. * A single log is capped at \~1 MB (≤ 25 KB recommended). Chunk batches conservatively. * Datadog intake doesn't dedup. Persist the cursor only after a successful POST, and index a stable record ID so you can dedup downstream if a retry re-sends a page. * Set each log's `timestamp` (epoch ms) from the record's created time, or logs land at receive-time and look mis-ordered. * Don't use the Observability Pipelines HTTP/S Client source — it polls a fixed URL and can't carry a cursor. **Docs:** [Send logs (HTTP intake) & sites](https://docs.datadoghq.com/api/latest/logs/) · [Log collection](https://docs.datadoghq.com/logs/log_collection/) · [Agent custom check `send_log`](https://docs.datadoghq.com/logs/log_collection/agent_checks/) #### Splunk **Forwarder to HEC.** The most reliable path is a scheduled script that walks the cursor and posts to the **HTTP Event Collector (HEC)**. (Alternatively, the REST API Modular Input add-on can poll the endpoint but needs a custom response handler to follow the cursor.) 1. Enable HEC: **Settings → Data Inputs → HTTP Event Collector → Global Settings → Enabled** (default port `8088`). 2. **New Token** — name it, set a default index (e.g. `agent_handler`) and sourcetype (e.g. `merge:ah:tool_calls`). Copy the token. 3. Run a scheduled script (cron, systemd timer, or a Splunk scripted input) that reads a persisted watermark, pages through the endpoint, and POSTs each record to HEC. ```bash curl -k https://:8088/services/collector/event \ -H "Authorization: Splunk " \ -d '{"sourcetype":"merge:ah:tool_calls","index":"agent_handler","event": { ...one results[] record... }}' ``` (The HEC header is `Authorization: Splunk `; `Bearer` is only for the Agent Handler side.) Prefer to stay inside a Splunk add-on? The **REST API Modular Input** ([Splunkbase app 1546](https://splunkbase.splunk.com/app/1546)) can poll with a `GET`, an `Authorization: Bearer …` header, and a polling interval — but following the response-body `next_cursor` across pages requires a **custom response handler** (`responsehandlers.py`); it isn't automatic. **Gotchas** * Checkpoint on a `created_after` timestamp watermark, not only the opaque cursor — an opaque cursor may be stale across runs, while a timestamp always lets you resume. Use the cursor to walk pages *within* a run. * Use `>=` on the watermark to avoid gaps, then dedup boundary records downstream by a stable record ID. * Keep `tool-calls/` and `api-calls/` as separate inputs/sourcetypes with separate checkpoints. * Splunk Edge Processor is a transform/route layer, not a poller — don't use it to collect. **Docs:** [Format events for HEC](https://help.splunk.com/en/splunk-enterprise/get-started/get-data-in/9.4/get-data-with-http-event-collector/format-events-for-http-event-collector) · [Manage HEC with cURL](https://help.splunk.com/en/splunk-cloud-platform/get-started/get-data-in/10.3.2512/get-data-with-http-event-collector/use-curl-to-manage-http-event-collector-tokens-events-and-services) · [REST API Modular Input](https://splunkbase.splunk.com/app/1546) #### Microsoft Sentinel **Native pull** via the **Codeless Connector Framework (CCF) `RestApiPoller`** — a direct match for this endpoint. You author JSON for four components (custom table, DCR, connector UI definition, and the `RestApiPoller` connection rule), package them into an ARM template, and deploy. The key blocks of the `RestApiPoller` connection rule: ```json "auth": { "type": "APIKey", "ApiKey": "[[parameters('apikey')]", "ApiKeyName": "Authorization", "ApiKeyIdentifier": "Bearer" }, "request": { "apiEndpoint": "https://ah-api.merge.dev/api/v1/logs/tool-calls/", "httpMethod": "GET", "queryWindowInMin": 5, "startTimeAttributeName": "created_after", "endTimeAttributeName": "created_before", "queryParameters": { "page_size": 1000 } }, "paging": { "pagingType": "NextPageToken", "nextPageTokenJsonPath": "$.next_cursor", "nextPageParaName": "cursor", "hasNextFlagJsonPath": "$.has_more" }, "response": { "format": "json", "eventsJsonPaths": [ "$.results" ] } ``` Why this works: * **Auth** — the `APIKey` type builds `Authorization: Bearer `. You **must** set `ApiKeyIdentifier: "Bearer"`; the default is `token`, which would produce `Authorization: token …` and 401. * **Time window** — `startTimeAttributeName`/`endTimeAttributeName` make Sentinel append `created_after`/`created_before` each cycle and advance the window. This is your reliable cross-run checkpoint. * **Paging** — `NextPageToken` reads `next_cursor` from the response body (`nextPageTokenJsonPath`) and sends it as the `cursor` query param (`nextPageParaName`); paging stops when `has_more` is `false`/`next_cursor` is null. * **Events** — `eventsJsonPaths: ["$.results"]` splits the array into records, mapped into a `_CL` custom table via the DCR's `transformKql`. **Gotchas** * Drive cross-run progress with the **time window** (above), not a bare persistent cursor — there are documented limitations and conflicting reports about persisting opaque cursors across poll cycles. Use `NextPageToken` only to exhaust pages *within* each window. * Set the first-poll start (`InitialCheckpointTimeUtc`) within the 30-day retention window; keep it modest (e.g. 24–48h back) to avoid a huge first backfill. * CCF doesn't dedup; adjacent windows can overlap on boundary timestamps. Dedup in the DCR `transformKql` or at query time by a stable record ID. * Custom tables need the `_CL` suffix and the DCR stream name must start with `Custom-`. * If Agent Handler restricts inbound IPs, allowlist Sentinel's `Scuba` service tag. **Docs:** [RestApiPoller reference](https://learn.microsoft.com/en-us/azure/sentinel/data-connector-connection-rules-reference) · [Create a codeless connector](https://learn.microsoft.com/en-us/azure/sentinel/create-codeless-connector) #### Elastic **Native pull** via the **`httpjson` input** (the **Custom API** integration in Elastic Agent / Fleet, or `filebeat.inputs` for standalone Filebeat). It's purpose-built for a Bearer-auth, cursor-paginated JSON endpoint with state persisted across restarts. In Fleet: **Integrations → Custom API → Add**, set the URL, `GET`, and an interval, store the key as a secret, and add the advanced config below. For standalone Filebeat: ```yaml filebeat.inputs: - type: httpjson interval: 5m request.url: https://ah-api.merge.dev/api/v1/logs/tool-calls/ request.method: GET request.transforms: - set: { target: header.Authorization, value: 'Bearer ${PRODUCTION_KEY}' } - set: { target: url.params.page_size, value: '1000' } - set: target: url.params.created_after value: '[[.cursor.last_created]]' default: '[[formatDate (now (parseDuration "-720h"))]]' # 30-day cold start response.split: target: body.results type: array ignore_empty_value: true response.pagination: # next_cursor (body) -> cursor (query param). Empty/null cursor ends the loop. - set: target: url.params.cursor value: '[[.last_response.body.next_cursor]]' fail_on_template_error: false cursor: last_created: value: '[[.last_event.created]]' # your record's timestamp field ignore_empty_value: true processors: - fingerprint: # stable _id for dedup fields: ["json.id"] # your record's unique id field target_field: "@metadata._id" method: sha256 ``` **Gotchas** * Pagination stops when the `cursor` param template evaluates to **empty** — so reading `next_cursor` terminates cleanly when it's `null`. **Don't put a `default:` on the pagination cursor**, or it loops forever. * `request.transforms` run once per interval; `response.pagination` runs per page. Keep `created_after` in transforms and `cursor` in pagination. * Map the raw `json.*` fields to ECS via an ingest pipeline if you want the Elastic Security app views to populate. * Replace `created`/`json.id` with your actual record timestamp and unique-id fields; confirm the cold-start template-function spelling for your stack version. **Docs:** [`httpjson` input reference](https://www.elastic.co/docs/reference/beats/filebeat/filebeat-input-httpjson) · [Custom API integration](https://www.elastic.co/docs/reference/integrations/httpjson) · [Fingerprint processor](https://www.elastic.co/docs/reference/beats/filebeat/fingerprint) #### Cribl Stream **Native pull** via the **REST / API Endpoint Collector**. Your endpoint maps to Cribl's **Response Body Attribute** pagination exactly. 1. **Data → Sources → Collectors → REST → Add Collector**, give it an ID (e.g. `ah_tool_calls`). 2. **Collect method** `GET`; **Collect URL** `` `https://ah-api.merge.dev/api/v1/logs/tool-calls/` ``. 3. **Authentication = None**, then add a **Collect header**: `Authorization` = `` `Bearer ${C.Secret('ah_prod_key','text').value}` `` (store the key as a Cribl secret). 4. **Pagination = Response Body Attribute**, **Response attribute = `next_cursor`**. Optionally set a **Last-page expression** `has_more === false` as a guard. 5. **Collect parameters:** `cursor` = `` `${next_cursor}` ``, `page_size` = `1000`, and `created_after`/`created_before` from the run's time range (ISO 8601). 6. **Schedule** the Collector with a cron interval, **Mode = Full Run**, **State tracking on**, **Concurrent run limit = 1**. On the first request `next_cursor` is undefined, so `cursor=` is empty and the API starts from the oldest record. Paging stops when the attribute is null. **Gotchas** * A one-off **Run** defaults to Preview (sample only); a **Schedule** defaults to Discovery. For real ingestion set **Full Run**. * State tracking dedups reliably only with **Concurrent run limit = 1**. Drive non-overlapping windows off `created_after`/`created_before`. * Cribl's `${earliest}`/`${latest}` are epoch by default — format to ISO 8601 for these params; verify epoch units in your build. * If you keep **Discover type = None**, make sure the Event Breaker splits the `results[]` array so each record becomes one event. **Docs:** [REST/API Collector](https://docs.cribl.io/stream/collectors-rest/) · [Using REST/API Collectors](https://docs.cribl.io/stream/usecase-rest/) · [Schedule & run collector jobs](https://docs.cribl.io/stream/collectors-schedule-run/) #### Sumo Logic **Native pull** via the **Universal Connector** (Cloud-to-Cloud), which polls an arbitrary REST API with Bearer auth and continuation-token pagination. (Fallback: an **HTTP Logs Source** plus a forwarder script.) 1. **Data Management → Collection →** next to a Hosted Collector, **Add Source → Universal Connector**. 2. **Authentication = Bearer**, paste the `PRODUCTION_KEY`. 3. **Request:** base URL `https://ah-api.merge.dev/api/v1/logs/tool-calls/`; add query params `page_size=1000` and the time-window params. 4. **Pagination = Continuation Token**, token location **body** via **JSON Path `$.next_cursor`**, sent back as the **`cursor` query parameter**. 5. **Response → JSON with JPath:** **Logs JPath `$.results[*]`** (one record per element), plus the timestamp JPath/format from the record's created field. 6. **Time window:** map the window variables to `created_after`/`created_before`; set **Initial Lookback ≤ 30 days**; set a polling interval. **Gotchas** * Keep Initial Lookback within the 30-day retention window. * Confirm the connector treats `next_cursor: null` / `has_more: false` as the stop condition (test a full drain). * Neither path dedups; dedup downstream by a stable record ID. * Forwarder fallback: POST **NDJSON** (one record per line) to the HTTP Source URL — use `--data-binary`, not `-d` (which strips newlines and collapses records into one message). **Docs:** [Universal Connector source](https://help.sumologic.com/docs/send-data/hosted-collectors/cloud-to-cloud-integration-framework/universal-connector-source/) · [HTTP Logs & Metrics Source](https://help.sumologic.com/docs/send-data/hosted-collectors/http-source/logs-metrics/) #### Other platforms Any platform with an HTTP log intake can ingest these logs with a small scheduled job. The pattern is always the same: 1. Read the persisted `cursor` (or, on first run, set `created_after` within the 30-day window). 2. `GET` the endpoint with `Authorization: Bearer ` and `page_size=1000`. 3. Transform `results[]` into your platform's log format and send the batch to its intake. 4. Persist `next_cursor` after the batch ships; repeat while `has_more` is `true`; on `false`, stop and resume from the saved cursor on the next run. Run one loop per endpoint (`tool-calls/` and `api-calls/`) with independent cursors. See the [API reference](/merge-agent-handler/agent-handler) for the full schemas.