> ## Documentation Index > Fetch the complete documentation index at: https://docs.mapping.travel/llms.txt > Use this file to discover all available pages before exploring further. # Data sources and sinks > Pull inventory from S3 or a webhook. Push results back the same way. Agents can pull inventory from customer-configured **data sources** and push mapping results to customer-configured **data sinks**. v1 supports two transport types — **S3** and **HTTPS webhook** — for both inbound and outbound. SFTP and GCS are deferred. ## Roles | Role | Direction | What gets moved | | ------------------ | --------- | ------------------------------------------------------------------------------------------- | | `INVENTORY_SOURCE` | Inbound | CSV / JSON / XLSX files pulled on a schedule, uploaded via the existing inventory pipeline. | | `RESULTS_SINK` | Outbound | Mapping job results pushed on job complete (opt-in). | ## Inbound: pulling inventory **Required config:** * `bucket` — bucket name (e.g. `acme-inventory`). * `prefix` — key prefix to watch (e.g. `daily-uploads/`). * `region` — AWS region (e.g. `us-east-1`). * `secretRef` — vault id containing `accessKeyId` + `secretAccessKey` (IAM keys with `s3:GetObject` + `s3:ListBucket`). The agent calls `pull_inventory_from_source(sourceId)` either on its `scheduleCron` (default the agent's cron) or on demand. New files since `lastSyncedAt` are downloaded, uploaded via the standard inventory pipeline, and recorded in `agent_data_source_sync.manifest_json`. The agent issues `POST {url}` on its schedule with `{ "since": "" }`. Your server returns a JSON array of file references (signed URLs) the agent then downloads. **Required config:** * `url` — fully-qualified HTTPS URL. * `authScheme` — `Bearer`, `HmacSha256`, or `Basic`. * `secretRef` — vault id with the credential value. * `format` — `CSV`, `JSON`, or `XLSX`. ### Auto-pull on schedule Controlled by the rule set: ```json theme={null} "data_sources": { "auto_pull_on_schedule": true, "pause_source_after_failures": 3 } ``` After three consecutive failures, the source is auto-paused, a `DATA_SOURCE_PULL_FAILED` pending item is filed, and configured channels are notified. ## Outbound: pushing results Outbound auto-push is **default off** — data egress requires explicit opt-in. Enable it in the rule set: ```json theme={null} "data_sinks": { "auto_push_on_job_complete": true, "retry_failed_push": 2 } ``` Same shape as the inbound config (`bucket` + `prefix` + `region` + `secretRef`), plus IAM keys with `s3:PutObject` permission. The agent writes `{prefix}/job-{jobId}.{format}` on successful job completion. The agent issues `POST {url}` with the full result payload. Three auth modes: * `Bearer` — `Authorization: Bearer ` * `HmacSha256` — header `X-MT-Signature: sha256=` * `Basic` — `Authorization: Basic ` Retries follow `retry_failed_push` with exponential backoff. Push actions are **not undoable** — outbound webhooks are external side effects. Use the test endpoint before flipping auto-push on. ## Secret handling All credentials live in the existing secret vault used by `BillingService`. They are: * Never logged. * Never returned in full from the API (`mt_byo_****1234` redaction). * Encrypted at rest. * Validated at save time via a low-cost ping (`OPTIONS` for webhooks, `ListBucket` for S3). ## Testing a source / sink Every source has a **Test** button (and a `POST /api/v1/agents/{id}/data-sources/{sid}/test` endpoint). It runs a read-only smoke check — no data moved, no sync record created. Returns a verbose pass/fail with the diagnostic message. Data sources tab ## Sync history The Inbound / Outbound subsections show the last N syncs per source. Per-row manifest, byte counts, errors. Drill in via `GET /api/v1/agents/{id}/data-sources/{sid}/syncs`. ## API: create a data source `POST /api/v1/agents/{agentId}/data-sources` Request body: | Field | Type | Required | Default | Description | | -------------- | ------- | -------- | ------- | ----------------------------------------------------------------------------------------------------------- | | `role` | enum | Yes | — | `INVENTORY_SOURCE` or `RESULTS_SINK`. | | `type` | enum | Yes | — | `S3` or `HTTPS_WEBHOOK`. | | `name` | string | Yes | — | Human-readable label for the source. | | `configJson` | object | No | `{}` | Transport-specific config. See the per-type keys below. | | `secret` | string | No | — | Credential payload stored in the vault (write-only; redacted on read). Shape depends on `type` — see below. | | `format` | enum | No | `CSV` | `CSV`, `JSON`, or `XLSX`. | | `scheduleCron` | string | No | — | Cron expression (UTC) for auto-pull. | | `enabled` | boolean | No | `true` | Whether the source is active. | `configJson` is stored as-is and read by the transport adapter at pull/push/test time. Keys by `type`: **`type: S3`** | Key | Required | Default | Description | | -------------- | -------- | -------------------------- | ---------------------------------------------------------------------- | | `bucket` | Yes | — | S3 bucket name. | | `prefix` | No | `""` | Key prefix to list/read (inbound). | | `region` | No | `us-east-1` | AWS region. | | `endpoint` | No | — | Custom endpoint for S3-compatible storage (enables path-style access). | | `key_template` | No | `exports/{jobId}.{format}` | `RESULTS_SINK` only — output key; supports `{jobId}` and `{format}`. | For S3, `secret` is a JSON string with AWS credentials: `{"accessKey":"...","secretKey":"..."}`. **`type: HTTPS_WEBHOOK`** | Key | Required | Default | Description | | ------ | -------- | ------- | --------------------------------------------------------------------------------- | | `url` | Yes | — | Endpoint to pull from / push to. | | `auth` | No | — | Object; set `auth.type` to `bearer`, `basic`, or `hmac_sha256`. Omit for no auth. | For `HTTPS_WEBHOOK`, `secret` holds the credential for the chosen `auth.type`: the token for `bearer` (sent as `Authorization: Bearer `), the `user:pass` for `basic`, or the signing key for `hmac_sha256` (requests carry `X-Timestamp`, `X-Signature`, `X-Signature-Version: v1`). ```json theme={null} theme={null} { "role": "INVENTORY_SOURCE", "type": "S3", "name": "Daily ACME drop", "configJson": { "bucket": "acme-inventory", "prefix": "daily-uploads/", "region": "us-east-1" }, "secret": "{\"accessKey\":\"AKIA...\",\"secretKey\":\"...\"}", "format": "CSV", "scheduleCron": "0 6 * * *", "enabled": true } ``` ## API: create a channel `POST /api/v1/agents/{agentId}/channels` Notification channels deliver agent events (pending items, failures, summaries). Request body: | Field | Type | Required | Default | Description | | -------------------- | --------- | -------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `type` | enum | Yes | — | `EMAIL`, `SLACK_WEBHOOK`, or `GENERIC_WEBHOOK`. | | `target` | string | Yes | — | Destination: an email address (`EMAIL`), a Slack incoming-webhook URL (`SLACK_WEBHOOK`), or any webhook URL (`GENERIC_WEBHOOK`). | | `eventSubscriptions` | string\[] | No | `[]` | Event kinds this channel receives. **An empty array subscribes to all events.** Otherwise, values are matched by exact string against the dispatched event kind. | | `enabled` | boolean | No | `true` | Whether the channel is active. | `eventSubscriptions` is a free-form list of event-kind strings, not a fixed enum. The event kinds the agent runtime currently emits are `cost_threshold`, `on_pause`, `on_action`, and `cycle_completed` (plus `test`, used by the channel test endpoint). Leave the array empty to receive everything. ```json theme={null} theme={null} { "type": "SLACK_WEBHOOK", "target": "https://hooks.slack.com/services/T000/B000/XXXX", "eventSubscriptions": ["cycle_completed", "on_action"], "enabled": true } ``` ## Related * [Agents overview](/agents/overview) — broader context. * [Recipes: Pull inventory from S3](/agents/recipes/pull-inventory-from-s3) * [Recipes: Push results to webhook](/agents/recipes/push-results-to-webhook)