Data sources and sinks

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

S3
HTTPS webhook (inbound)

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": "<lastSyncedAt>" }. 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:

"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:

"data_sinks": {
  "auto_push_on_job_complete": true,
  "retry_failed_push": 2
}

S3
HTTPS webhook (outbound)

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 <secret>
HmacSha256 — header X-MT-Signature: sha256=<hmac(body, secret)>
Basic — Authorization: Basic <base64(secret)>

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 <token>), the user:pass for basic, or the signing key for hmac_sha256 (requests carry X-Timestamp, X-Signature, X-Signature-Version: v1).

{
  "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.

{
  "type": "SLACK_WEBHOOK",
  "target": "https://hooks.slack.com/services/T000/B000/XXXX",
  "eventSubscriptions": ["cycle_completed", "on_action"],
  "enabled": true
}

Get Started

Core Concepts

Guides

Delivery destinations

MCP

Agents

Roles

Inbound: pulling inventory

Auto-pull on schedule

Outbound: pushing results

Secret handling

Testing a source / sink

Sync history

API: create a data source

API: create a channel

​Roles

​Inbound: pulling inventory

​Auto-pull on schedule

​Outbound: pushing results

​Secret handling

​Testing a source / sink

​Sync history

​API: create a data source

​API: create a channel

​Related

Roles

Inbound: pulling inventory

Auto-pull on schedule

Outbound: pushing results

Secret handling

Testing a source / sink

Sync history

API: create a data source

API: create a channel

Related