---
title: "API Quickstart"
description: "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.

:::info
For full endpoint details, request/response schemas, and an interactive playground, see the [API Reference](/api-reference/authentication/get-access-token) tab.
:::

## Authentication

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

1. **Create API credentials** (Client ID and Secret) in the Streamkap UI -- see [API Tokens](/api-tokens) for instructions.
2. **Exchange credentials for an access token** by calling the `/auth/access-token` endpoint.
3. **Use the access token** as a Bearer token in the `Authorization` header for all subsequent requests.

### Get an Access Token

```bash cURL
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"
  }'
```

```python Python
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']}")
```

```javascript JavaScript
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;
```

```go Go
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:

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

```python Python
import requests

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

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

```go Go
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:

```bash
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.

1. **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.

    ```bash
    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.

2. **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.

    ```bash
    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.

3. **Create a pipeline**

    Link the source and destination together by referencing their IDs.

    ```bash
    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.

:::note
Connector configurations vary by type. Refer to the source and destination documentation pages for the full list of configuration options for each connector.
:::

:::warning
**This workflow does not apply to [API sources](/api-sources)**. You don't create a pipeline for them — see [Send an API Source to a Destination](#5-send-an-api-source-to-a-destination) below.
:::

### 2. Monitor Pipeline Health

Check the status and performance of your running pipelines.

1. **Get pipeline details**

    Retrieve the current state of a pipeline:

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

2. **Check pipeline metrics**

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

    ```bash
    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

3. **Monitor consumer group lag**

    List consumer groups and inspect their lag to identify bottlenecks:

    ```bash
    # 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"
    ```

4. **Check source metrics and snapshot status**

    Monitor source connector health and snapshot progress:

    ```bash
    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.

1. **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.

    ```bash
    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:

    ```bash
    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'\''"
          }
        ]
      }'
    ```

2. **Monitor snapshot progress**

    Check the source metrics to track the snapshot:

    ```bash
    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).

3. **Stop a snapshot (if needed)**

    To cancel a running incremental snapshot:

    ```bash
    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.

1. **Get the current source configuration**

    ```bash
    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.

2. **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.

    ```bash
    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"
        }
      }'
    ```

:::tip
For a Terraform-based approach to adding tables, see [Adding Tables to an Existing Source](/terraform-resources#adding-tables-to-an-existing-source).
:::

### 5. Send an API Source to a Destination

You don't create a pipeline for an [API source](/api-sources): `POST /pipelines` rejects one with a `400`. Instead you attach individual **topics** to a destination, and Streamkap creates and maintains the underlying pipeline for you.

The source itself is created with the ordinary `POST /sources`; only the linking step differs.

1. **Create the API source**

    ```bash
    curl -X POST "https://api.streamkap.com/sources" \
      -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
      -H "Content-Type: application/json" \
      -d '{
        "name": "my-hubspot-source",
        "connector": "hubspot",
        "config": {
          "token": "YOUR_HUBSPOT_PRIVATE_APP_TOKEN",
          "resources": ["contacts", "deals"]
        }
      }'
    ```

    Each selected resource produces one topic, named `source_{source_id}.{connector}.{resource}` — for example `source_123.hubspot.contacts`.

2. **Send a topic to a destination**

    `PUT` is idempotent: sending the same topic to the same destination twice is not an error.

    ```bash
    curl -X PUT "https://api.streamkap.com/topics/TOPIC_ID/destinations/DESTINATION_ID" \
      -H "Authorization: Bearer YOUR_ACCESS_TOKEN"
    ```

    The response is the managed binding: its `binding_id` (the auto-managed pipeline), the source and destination ids, and the binding's full `topic_ids` set.

    A destination needs an environment tag (such as Production) to receive topics; if it has none, Streamkap uses your `production` environment tag, and if that doesn't exist either, add one on the destination's page or under **Tags**.

3. **Check where a topic is being sent**

    ```bash
    curl -X GET "https://api.streamkap.com/topics/TOPIC_ID/destinations" \
      -H "Authorization: Bearer YOUR_ACCESS_TOKEN"
    ```

4. **Stop sending a topic to a destination**

    ```bash
    curl -X DELETE "https://api.streamkap.com/topics/TOPIC_ID/destinations/DESTINATION_ID" \
      -H "Authorization: Bearer YOUR_ACCESS_TOKEN"
    ```

    Detaching the last topic removes the binding entirely.

The full set of topic-destination endpoints:

| Method | Path | Purpose |
| --- | --- | --- |
| `GET` | `/topics/{topic_id}/destinations` | List the destinations a topic is sent to |
| `GET` | `/topics/{topic_id}/destinations/{destination_id}` | Get one destination a topic is sent to |
| `PUT` | `/topics/{topic_id}/destinations/{destination_id}` | Send a topic to a destination (idempotent) |
| `PUT` | `/topics/{topic_id}/destinations` | Replace a topic's full destination set |
| `DELETE` | `/topics/{topic_id}/destinations/{destination_id}` | Stop sending a topic to a destination |

`PUT /topics/{topic_id}/destinations` takes the topic's complete destination set, keyed by destination id. Destinations you omit are detached:

```bash
curl -X PUT "https://api.streamkap.com/topics/TOPIC_ID/destinations" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "destinations": {
      "DESTINATION_ID_A": {},
      "DESTINATION_ID_B": {}
    }
  }'
```

:::note
These endpoints work on API-source topics only. A topic from a CDC source returns a `400` — build a pipeline for those, as in [Create a Complete Pipeline](#1-create-a-complete-pipeline).
:::

:::warning
The pipelines these bindings create are **auto-managed**. Edit or delete them through the topic-destination endpoints above, not through `/pipelines/{pipeline_id}` — that returns a `409`.
:::

---

## API Response Patterns

### Standard Responses

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

```json
{
  "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:

```bash
# 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:

```json
{
  "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  |

:::info
Detailed request/response schemas for every endpoint are available in the [API Reference](/api-reference/authentication/get-access-token).
:::

---

## Rate Limits and Best Practices

:::note
API rate limits are managed by Streamkap and are designed to accommodate normal usage patterns. If you encounter `429 Too Many Requests` responses or believe your use case requires higher limits, contact [Streamkap support](mailto:support@streamkap.com) with details about your integration and expected request volume.
:::

:::info
**API versioning**

The Streamkap API is currently unversioned. All endpoints use a single base URL (`https://api.streamkap.com`). If breaking changes are introduced in the future, they will be communicated in advance through release notes and direct customer notification, with a deprecation period to allow migration.
:::

- **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.

---

## SDKs and Tools

### Terraform Provider

The [Streamkap Terraform Provider](/streamkap-provider-for-terraform) enables infrastructure-as-code management of sources, destinations, pipelines, and transforms. See the full [Terraform Configuration guide](/terraform-configuration) to get started.

### OpenAPI Specification

The Streamkap API is documented with an OpenAPI 3.1 specification. You can use the spec to generate client libraries for any language using tools like [OpenAPI Generator](https://openapi-generator.tech/) or [Speakeasy](https://www.speakeasy.com/).

The spec is accessible in the [API Reference](/api-reference/authentication/get-access-token) tab and powers the interactive endpoint playground.

---

## Next Steps

- [API Tokens](/api-tokens) -- Create and manage API credentials
- [API Reference](/api-reference/authentication/get-access-token) -- Full interactive endpoint documentation
- [Terraform Provider](/streamkap-provider-for-terraform) -- Manage infrastructure as code
- [Terraform Resources](/terraform-resources) -- Source, destination, and pipeline resource examples
- [Snapshots](/snapshots) -- Learn how initial and incremental snapshots work