API Quickstart

Authenticate to the Streamkap REST API with bearer tokens, then create sources, destinations, and pipelines, monitor lag, and trigger incremental snapshots.

The Streamkap REST API lets you programmatically manage sources, destinations, pipelines, and transforms. This guide covers authentication and common workflows to get you up and running quickly.

Authentication

The Streamkap API uses bearer tokens for authentication. The flow is:

Create API credentials (Client ID and Secret) in the Streamkap UI – see API Tokens for instructions.
Exchange credentials for an access token by calling the /auth/access-token endpoint.
Use the access token as a Bearer token in the Authorization header for all subsequent requests.

Get an Access Token

curl -X POST "https://api.streamkap.com/auth/access-token" \
  -H "Content-Type: application/json" \
  -d '{
    "client_id": "YOUR_CLIENT_ID",
    "secret": "YOUR_CLIENT_SECRET"
  }'

import requests

response = requests.post(
    "https://api.streamkap.com/auth/access-token",
    json={
        "client_id": "YOUR_CLIENT_ID",
        "secret": "YOUR_CLIENT_SECRET"
    }
)

data = response.json()
access_token = data["accessToken"]
refresh_token = data["refreshToken"]
print(f"Access token expires: {data['expires']}")

const response = await fetch("https://api.streamkap.com/auth/access-token", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    client_id: "YOUR_CLIENT_ID",
    secret: "YOUR_CLIENT_SECRET",
  }),
});

const data = await response.json();
const accessToken = data.accessToken;
const refreshToken = data.refreshToken;

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
)

func main() {
    body, _ := json.Marshal(map[string]string{
        "client_id": "YOUR_CLIENT_ID",
        "secret":    "YOUR_CLIENT_SECRET",
    })

    resp, err := http.Post(
        "https://api.streamkap.com/auth/access-token",
        "application/json",
        bytes.NewBuffer(body),
    )
    if err != nil {
        panic(err)
    }
    defer resp.Body.Close()

    var data map[string]interface{}
    json.NewDecoder(resp.Body).Decode(&data)
    fmt.Println("Access token:", data["accessToken"])
}

The response includes:

Field	Description
`accessToken`	Bearer token to use in API requests
`refreshToken`	Token to obtain a new access token
`expires`	Expiration timestamp of the access token
`expiresIn`	Seconds until the access token expires

Make Authenticated Requests

Include the access token in the Authorization header for all API calls:

curl -X GET "https://api.streamkap.com/sources" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

import requests

headers = {"Authorization": "Bearer YOUR_ACCESS_TOKEN"}
response = requests.get("https://api.streamkap.com/sources", headers=headers)
sources = response.json()

const response = await fetch("https://api.streamkap.com/sources", {
  headers: { Authorization: "Bearer YOUR_ACCESS_TOKEN" },
});
const sources = await response.json();

req, _ := http.NewRequest("GET", "https://api.streamkap.com/sources", nil)
req.Header.Set("Authorization", "Bearer YOUR_ACCESS_TOKEN")
resp, err := http.DefaultClient.Do(req)

Refresh an Access Token

When your access token expires, use the refresh token to obtain a new one without re-authenticating with your client credentials:

curl -X POST "https://api.streamkap.com/auth/access-token/refresh" \
  -H "Content-Type: application/json" \
  -d '{
    "refresh_token": "YOUR_REFRESH_TOKEN"
  }'

Common Workflows

1. Create a Complete Pipeline

Build an end-to-end CDC pipeline by creating a source, destination, and then linking them in a pipeline.

Create a source connector

Create a source by specifying the connector type and its configuration. The connector field identifies the source type (e.g., postgresql, mysql, mongodb), and config contains connector-specific settings.

curl -X POST "https://api.streamkap.com/sources" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-postgresql-source",
    "connector": "postgresql",
    "config": {
      "database.hostname": "db.example.com",
      "database.port": "5432",
      "database.user": "cdc_user",
      "database.password": "YOUR_DB_PASSWORD",
      "database.dbname": "my_database",
      "table.include.list": "public.orders,public.customers"
    }
  }'

Save the returned id value – you will need it when creating the pipeline.

Create a destination connector

Create a destination where your CDC data will be delivered. As with sources, the connector field identifies the destination type and config holds its settings.

curl -X POST "https://api.streamkap.com/destinations" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-snowflake-destination",
    "connector": "snowflake",
    "config": {
      "snowflake.url.name": "your-account.snowflakecomputing.com",
      "snowflake.user.name": "STREAMKAP_USER",
      "snowflake.private.key": "YOUR_PRIVATE_KEY",
      "snowflake.database.name": "MY_DATABASE",
      "snowflake.schema.name": "PUBLIC"
    }
  }'

Save the returned id value for the next step.

Create a pipeline

Link the source and destination together by referencing their IDs.

curl -X POST "https://api.streamkap.com/pipelines" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-cdc-pipeline",
    "source": {
      "id": "SOURCE_ID_HERE",
      "connector": "postgresql",
      "name": "my-postgresql-source"
    },
    "destination": {
      "id": "DESTINATION_ID_HERE",
      "connector": "snowflake",
      "name": "my-snowflake-destination"
    },
    "snapshot_new_tables": true,
    "tags": []
  }'

The pipeline starts automatically after creation. The source will begin an initial snapshot of the configured tables.

2. Monitor Pipeline Health

Check the status and performance of your running pipelines.

Get pipeline details

Retrieve the current state of a pipeline:

curl -X GET "https://api.streamkap.com/pipelines/PIPELINE_ID" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Check pipeline metrics

Get latency, throughput, and lag metrics for a pipeline:

curl -X GET "https://api.streamkap.com/pipelines/PIPELINE_ID/metrics" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Key metrics to watch:

latency – End-to-end delay from source change to destination delivery
recordsLag – Number of records waiting to be processed

Monitor consumer group lag

List consumer groups and inspect their lag to identify bottlenecks:

# List all consumer groups
curl -X GET "https://api.streamkap.com/consumer-groups" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

# Get details for a specific consumer group
curl -X GET "https://api.streamkap.com/consumer-groups/GROUP_ID" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Check source metrics and snapshot status

Monitor source connector health and snapshot progress:

curl -X GET "https://api.streamkap.com/sources/SOURCE_ID/metrics" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

The response includes snapshot-related fields such as snapshotStatus, snapshotState, SnapshotCompleted, and SnapshotRunning.

3. Trigger a Snapshot

Trigger an incremental snapshot to re-read data from specific tables without resetting the entire source connector.

Execute an incremental snapshot

Specify the source ID and optionally the topic names (tables) to snapshot. If topic_names is omitted, all tables configured on the source will be snapshotted.

curl -X POST "https://api.streamkap.com/sources/execute_incremental_snapshot" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "SOURCE_ID_HERE",
    "topic_names": ["my_database.public.orders"]
  }'

You can also include optional additional_conditions to filter which rows are snapshotted:

curl -X POST "https://api.streamkap.com/sources/execute_incremental_snapshot" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "SOURCE_ID_HERE",
    "topic_names": ["my_database.public.orders"],
    "additional_conditions": [
      {
        "data_collection": "public.orders",
        "filter": "created_at > '\''2025-01-01'\''"
      }
    ]
  }'

Monitor snapshot progress

Check the source metrics to track the snapshot:

curl -X GET "https://api.streamkap.com/sources/SOURCE_ID/metrics" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Look for SnapshotRunning (number of tables currently being snapshotted) and SnapshotCompleted (number of tables that have finished).

Stop a snapshot (if needed)

To cancel a running incremental snapshot:

curl -X POST "https://api.streamkap.com/sources/stop_incremental_snapshot" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "SOURCE_ID_HERE"
  }'

4. Add Tables to an Existing Source

Add new tables to a running source without modifying the original table list, using the table.include.list.user.defined configuration parameter.

Get the current source configuration

curl -X GET "https://api.streamkap.com/sources/SOURCE_ID" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Note the current name, connector, and config values from the response.

Update the source with new tables

Use PUT /sources/{source_id} to update the source configuration. Add the new tables via the table.include.list.user.defined parameter while keeping all existing configuration values.

curl -X PUT "https://api.streamkap.com/sources/SOURCE_ID" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-postgresql-source",
    "connector": "postgresql",
    "config": {
      "database.hostname": "db.example.com",
      "database.port": "5432",
      "database.user": "cdc_user",
      "database.password": "YOUR_DB_PASSWORD",
      "database.dbname": "my_database",
      "table.include.list": "public.orders,public.customers",
      "table.include.list.user.defined": "public.products,public.inventory"
    }
  }'

API Response Patterns

Standard Responses

All API responses return JSON. Successful responses typically include the resource object directly:

{
  "id": "abc123",
  "name": "my-postgresql-source",
  "connector": "postgresql",
  "config": { ... },
  "created_at": "2025-06-15T10:30:00Z"
}

Paginated Responses

List endpoints support pagination via page and page_size query parameters:

# Get page 2 with 20 items per page
curl -X GET "https://api.streamkap.com/sources?page=2&page_size=20" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Parameter	Default	Description
`page`	`1`	Page number (1-indexed)
`page_size`	`10`	Number of items per page
`sort`	–	Sort field (e.g., `name`, `created_at`)
`sort_dir`	`asc`	Sort direction: `asc` or `desc`

Error Responses

Validation errors return a 422 status code with details about the issue:

{
  "detail": [
    {
      "loc": ["body", "name"],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

Other common HTTP status codes:

Status	Meaning
`200`	Success
`401`	Unauthorized – invalid or expired token
`404`	Resource not found
`422`	Validation error – check the `detail` field

Rate Limits and Best Practices

Use pagination for list endpoints. Avoid fetching all resources in a single request by using page and page_size parameters.
Cache responses where appropriate. Source and destination configurations change infrequently, so you can safely cache them for short periods.
Handle retries with exponential backoff. If a request fails with a 5xx status code, wait before retrying. Start with a 1-second delay and double it on each retry, up to a maximum of 30 seconds.
Refresh tokens proactively. Use the refresh token endpoint before your access token expires rather than waiting for a 401 error.
Minimize unnecessary polling. When monitoring pipeline health, use a reasonable interval (e.g., every 30–60 seconds) rather than polling continuously.

API Tokens – Create and manage API credentials
API Reference – Full interactive endpoint documentation
Terraform Provider – Manage infrastructure as code
Terraform Resources – Source, destination, and pipeline resource examples
Snapshots – Learn how initial and incremental snapshots work