Create Knowledge Base

curl --request POST \
  --url https://api.streamkap.com/knowledge-bases/config \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --data '
{
  "name": "<string>",
  "source": {
    "topic": "<string>",
    "topicPattern": "<string>",
    "inputSerialization": "JsonConfluent",
    "fields": [
      "<string>"
    ],
    "textFields": [
      "<string>"
    ],
    "metadataFields": [
      "<string>"
    ]
  },
  "embedding": {
    "embeddingConnectionId": "<string>",
    "model": "<string>",
    "dimensions": 32768,
    "batchSize": 1024
  },
  "vectorStore": {
    "vectorStoreConnectionId": "<string>",
    "indexName": "<string>",
    "namespace": "<string>",
    "metric": "cosine"
  },
  "description": "<string>",
  "text": {
    "textTemplate": ""
  },
  "processing": {
    "parallelism": 1,
    "checkpointIntervalMin": 5,
    "chunkSize": 512,
    "chunkOverlap": 50
  }
}
'

import requests

url = "https://api.streamkap.com/knowledge-bases/config"

payload = {
    "name": "<string>",
    "source": {
        "topic": "<string>",
        "topicPattern": "<string>",
        "inputSerialization": "JsonConfluent",
        "fields": ["<string>"],
        "textFields": ["<string>"],
        "metadataFields": ["<string>"]
    },
    "embedding": {
        "embeddingConnectionId": "<string>",
        "model": "<string>",
        "dimensions": 32768,
        "batchSize": 1024
    },
    "vectorStore": {
        "vectorStoreConnectionId": "<string>",
        "indexName": "<string>",
        "namespace": "<string>",
        "metric": "cosine"
    },
    "description": "<string>",
    "text": { "textTemplate": "" },
    "processing": {
        "parallelism": 1,
        "checkpointIntervalMin": 5,
        "chunkSize": 512,
        "chunkOverlap": 50
    }
}
headers = {
    "Authorization": "Bearer <token>",
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)

print(response.text)

const options = {
  method: 'POST',
  headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
  body: JSON.stringify({
    name: '<string>',
    source: {
      topic: '<string>',
      topicPattern: '<string>',
      inputSerialization: 'JsonConfluent',
      fields: ['<string>'],
      textFields: ['<string>'],
      metadataFields: ['<string>']
    },
    embedding: {
      embeddingConnectionId: '<string>',
      model: '<string>',
      dimensions: 32768,
      batchSize: 1024
    },
    vectorStore: {
      vectorStoreConnectionId: '<string>',
      indexName: '<string>',
      namespace: '<string>',
      metric: 'cosine'
    },
    description: '<string>',
    text: {textTemplate: ''},
    processing: {parallelism: 1, checkpointIntervalMin: 5, chunkSize: 512, chunkOverlap: 50}
  })
};

fetch('https://api.streamkap.com/knowledge-bases/config', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.streamkap.com/knowledge-bases/config",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "POST",
  CURLOPT_POSTFIELDS => json_encode([
    'name' => '<string>',
    'source' => [
        'topic' => '<string>',
        'topicPattern' => '<string>',
        'inputSerialization' => 'JsonConfluent',
        'fields' => [
                '<string>'
        ],
        'textFields' => [
                '<string>'
        ],
        'metadataFields' => [
                '<string>'
        ]
    ],
    'embedding' => [
        'embeddingConnectionId' => '<string>',
        'model' => '<string>',
        'dimensions' => 32768,
        'batchSize' => 1024
    ],
    'vectorStore' => [
        'vectorStoreConnectionId' => '<string>',
        'indexName' => '<string>',
        'namespace' => '<string>',
        'metric' => 'cosine'
    ],
    'description' => '<string>',
    'text' => [
        'textTemplate' => ''
    ],
    'processing' => [
        'parallelism' => 1,
        'checkpointIntervalMin' => 5,
        'chunkSize' => 512,
        'chunkOverlap' => 50
    ]
  ]),
  CURLOPT_HTTPHEADER => [
    "Authorization: Bearer <token>",
    "Content-Type: application/json"
  ],
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"strings"
	"net/http"
	"io"
)

func main() {

	url := "https://api.streamkap.com/knowledge-bases/config"

	payload := strings.NewReader("{\n  \"name\": \"<string>\",\n  \"source\": {\n    \"topic\": \"<string>\",\n    \"topicPattern\": \"<string>\",\n    \"inputSerialization\": \"JsonConfluent\",\n    \"fields\": [\n      \"<string>\"\n    ],\n    \"textFields\": [\n      \"<string>\"\n    ],\n    \"metadataFields\": [\n      \"<string>\"\n    ]\n  },\n  \"embedding\": {\n    \"embeddingConnectionId\": \"<string>\",\n    \"model\": \"<string>\",\n    \"dimensions\": 32768,\n    \"batchSize\": 1024\n  },\n  \"vectorStore\": {\n    \"vectorStoreConnectionId\": \"<string>\",\n    \"indexName\": \"<string>\",\n    \"namespace\": \"<string>\",\n    \"metric\": \"cosine\"\n  },\n  \"description\": \"<string>\",\n  \"text\": {\n    \"textTemplate\": \"\"\n  },\n  \"processing\": {\n    \"parallelism\": 1,\n    \"checkpointIntervalMin\": 5,\n    \"chunkSize\": 512,\n    \"chunkOverlap\": 50\n  }\n}")

	req, _ := http.NewRequest("POST", url, payload)

	req.Header.Add("Authorization", "Bearer <token>")
	req.Header.Add("Content-Type", "application/json")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

HttpResponse<String> response = Unirest.post("https://api.streamkap.com/knowledge-bases/config")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"name\": \"<string>\",\n  \"source\": {\n    \"topic\": \"<string>\",\n    \"topicPattern\": \"<string>\",\n    \"inputSerialization\": \"JsonConfluent\",\n    \"fields\": [\n      \"<string>\"\n    ],\n    \"textFields\": [\n      \"<string>\"\n    ],\n    \"metadataFields\": [\n      \"<string>\"\n    ]\n  },\n  \"embedding\": {\n    \"embeddingConnectionId\": \"<string>\",\n    \"model\": \"<string>\",\n    \"dimensions\": 32768,\n    \"batchSize\": 1024\n  },\n  \"vectorStore\": {\n    \"vectorStoreConnectionId\": \"<string>\",\n    \"indexName\": \"<string>\",\n    \"namespace\": \"<string>\",\n    \"metric\": \"cosine\"\n  },\n  \"description\": \"<string>\",\n  \"text\": {\n    \"textTemplate\": \"\"\n  },\n  \"processing\": {\n    \"parallelism\": 1,\n    \"checkpointIntervalMin\": 5,\n    \"chunkSize\": 512,\n    \"chunkOverlap\": 50\n  }\n}")
  .asString();

require 'uri'
require 'net/http'

url = URI("https://api.streamkap.com/knowledge-bases/config")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Post.new(url)
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n  \"name\": \"<string>\",\n  \"source\": {\n    \"topic\": \"<string>\",\n    \"topicPattern\": \"<string>\",\n    \"inputSerialization\": \"JsonConfluent\",\n    \"fields\": [\n      \"<string>\"\n    ],\n    \"textFields\": [\n      \"<string>\"\n    ],\n    \"metadataFields\": [\n      \"<string>\"\n    ]\n  },\n  \"embedding\": {\n    \"embeddingConnectionId\": \"<string>\",\n    \"model\": \"<string>\",\n    \"dimensions\": 32768,\n    \"batchSize\": 1024\n  },\n  \"vectorStore\": {\n    \"vectorStoreConnectionId\": \"<string>\",\n    \"indexName\": \"<string>\",\n    \"namespace\": \"<string>\",\n    \"metric\": \"cosine\"\n  },\n  \"description\": \"<string>\",\n  \"text\": {\n    \"textTemplate\": \"\"\n  },\n  \"processing\": {\n    \"parallelism\": 1,\n    \"checkpointIntervalMin\": 5,\n    \"chunkSize\": 512,\n    \"chunkOverlap\": 50\n  }\n}"

response = http.request(request)
puts response.read_body

{
  "_id": "<string>",
  "name": "<string>",
  "flink_job_name": "<string>",
  "flink_job_id": "<string>",
  "parallelism": 1,
  "error_message": "<string>",
  "agent_config": {},
  "created_by": "<string>",
  "created_timestamp": "2023-11-07T05:31:56Z",
  "updated_timestamp": "2023-11-07T05:31:56Z"
}

{
  "detail": [
    {
      "loc": [
        "<string>"
      ],
      "msg": "<string>",
      "type": "<string>",
      "input": "<unknown>",
      "ctx": {}
    }
  ]
}

POST

knowledge-bases

config

Create Knowledge Base

curl --request POST \
  --url https://api.streamkap.com/knowledge-bases/config \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --data '
{
  "name": "<string>",
  "source": {
    "topic": "<string>",
    "topicPattern": "<string>",
    "inputSerialization": "JsonConfluent",
    "fields": [
      "<string>"
    ],
    "textFields": [
      "<string>"
    ],
    "metadataFields": [
      "<string>"
    ]
  },
  "embedding": {
    "embeddingConnectionId": "<string>",
    "model": "<string>",
    "dimensions": 32768,
    "batchSize": 1024
  },
  "vectorStore": {
    "vectorStoreConnectionId": "<string>",
    "indexName": "<string>",
    "namespace": "<string>",
    "metric": "cosine"
  },
  "description": "<string>",
  "text": {
    "textTemplate": ""
  },
  "processing": {
    "parallelism": 1,
    "checkpointIntervalMin": 5,
    "chunkSize": 512,
    "chunkOverlap": 50
  }
}
'

import requests

url = "https://api.streamkap.com/knowledge-bases/config"

payload = {
    "name": "<string>",
    "source": {
        "topic": "<string>",
        "topicPattern": "<string>",
        "inputSerialization": "JsonConfluent",
        "fields": ["<string>"],
        "textFields": ["<string>"],
        "metadataFields": ["<string>"]
    },
    "embedding": {
        "embeddingConnectionId": "<string>",
        "model": "<string>",
        "dimensions": 32768,
        "batchSize": 1024
    },
    "vectorStore": {
        "vectorStoreConnectionId": "<string>",
        "indexName": "<string>",
        "namespace": "<string>",
        "metric": "cosine"
    },
    "description": "<string>",
    "text": { "textTemplate": "" },
    "processing": {
        "parallelism": 1,
        "checkpointIntervalMin": 5,
        "chunkSize": 512,
        "chunkOverlap": 50
    }
}
headers = {
    "Authorization": "Bearer <token>",
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)

print(response.text)

const options = {
  method: 'POST',
  headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
  body: JSON.stringify({
    name: '<string>',
    source: {
      topic: '<string>',
      topicPattern: '<string>',
      inputSerialization: 'JsonConfluent',
      fields: ['<string>'],
      textFields: ['<string>'],
      metadataFields: ['<string>']
    },
    embedding: {
      embeddingConnectionId: '<string>',
      model: '<string>',
      dimensions: 32768,
      batchSize: 1024
    },
    vectorStore: {
      vectorStoreConnectionId: '<string>',
      indexName: '<string>',
      namespace: '<string>',
      metric: 'cosine'
    },
    description: '<string>',
    text: {textTemplate: ''},
    processing: {parallelism: 1, checkpointIntervalMin: 5, chunkSize: 512, chunkOverlap: 50}
  })
};

fetch('https://api.streamkap.com/knowledge-bases/config', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.streamkap.com/knowledge-bases/config",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "POST",
  CURLOPT_POSTFIELDS => json_encode([
    'name' => '<string>',
    'source' => [
        'topic' => '<string>',
        'topicPattern' => '<string>',
        'inputSerialization' => 'JsonConfluent',
        'fields' => [
                '<string>'
        ],
        'textFields' => [
                '<string>'
        ],
        'metadataFields' => [
                '<string>'
        ]
    ],
    'embedding' => [
        'embeddingConnectionId' => '<string>',
        'model' => '<string>',
        'dimensions' => 32768,
        'batchSize' => 1024
    ],
    'vectorStore' => [
        'vectorStoreConnectionId' => '<string>',
        'indexName' => '<string>',
        'namespace' => '<string>',
        'metric' => 'cosine'
    ],
    'description' => '<string>',
    'text' => [
        'textTemplate' => ''
    ],
    'processing' => [
        'parallelism' => 1,
        'checkpointIntervalMin' => 5,
        'chunkSize' => 512,
        'chunkOverlap' => 50
    ]
  ]),
  CURLOPT_HTTPHEADER => [
    "Authorization: Bearer <token>",
    "Content-Type: application/json"
  ],
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"strings"
	"net/http"
	"io"
)

func main() {

	url := "https://api.streamkap.com/knowledge-bases/config"

	payload := strings.NewReader("{\n  \"name\": \"<string>\",\n  \"source\": {\n    \"topic\": \"<string>\",\n    \"topicPattern\": \"<string>\",\n    \"inputSerialization\": \"JsonConfluent\",\n    \"fields\": [\n      \"<string>\"\n    ],\n    \"textFields\": [\n      \"<string>\"\n    ],\n    \"metadataFields\": [\n      \"<string>\"\n    ]\n  },\n  \"embedding\": {\n    \"embeddingConnectionId\": \"<string>\",\n    \"model\": \"<string>\",\n    \"dimensions\": 32768,\n    \"batchSize\": 1024\n  },\n  \"vectorStore\": {\n    \"vectorStoreConnectionId\": \"<string>\",\n    \"indexName\": \"<string>\",\n    \"namespace\": \"<string>\",\n    \"metric\": \"cosine\"\n  },\n  \"description\": \"<string>\",\n  \"text\": {\n    \"textTemplate\": \"\"\n  },\n  \"processing\": {\n    \"parallelism\": 1,\n    \"checkpointIntervalMin\": 5,\n    \"chunkSize\": 512,\n    \"chunkOverlap\": 50\n  }\n}")

	req, _ := http.NewRequest("POST", url, payload)

	req.Header.Add("Authorization", "Bearer <token>")
	req.Header.Add("Content-Type", "application/json")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

HttpResponse<String> response = Unirest.post("https://api.streamkap.com/knowledge-bases/config")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"name\": \"<string>\",\n  \"source\": {\n    \"topic\": \"<string>\",\n    \"topicPattern\": \"<string>\",\n    \"inputSerialization\": \"JsonConfluent\",\n    \"fields\": [\n      \"<string>\"\n    ],\n    \"textFields\": [\n      \"<string>\"\n    ],\n    \"metadataFields\": [\n      \"<string>\"\n    ]\n  },\n  \"embedding\": {\n    \"embeddingConnectionId\": \"<string>\",\n    \"model\": \"<string>\",\n    \"dimensions\": 32768,\n    \"batchSize\": 1024\n  },\n  \"vectorStore\": {\n    \"vectorStoreConnectionId\": \"<string>\",\n    \"indexName\": \"<string>\",\n    \"namespace\": \"<string>\",\n    \"metric\": \"cosine\"\n  },\n  \"description\": \"<string>\",\n  \"text\": {\n    \"textTemplate\": \"\"\n  },\n  \"processing\": {\n    \"parallelism\": 1,\n    \"checkpointIntervalMin\": 5,\n    \"chunkSize\": 512,\n    \"chunkOverlap\": 50\n  }\n}")
  .asString();

require 'uri'
require 'net/http'

url = URI("https://api.streamkap.com/knowledge-bases/config")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Post.new(url)
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n  \"name\": \"<string>\",\n  \"source\": {\n    \"topic\": \"<string>\",\n    \"topicPattern\": \"<string>\",\n    \"inputSerialization\": \"JsonConfluent\",\n    \"fields\": [\n      \"<string>\"\n    ],\n    \"textFields\": [\n      \"<string>\"\n    ],\n    \"metadataFields\": [\n      \"<string>\"\n    ]\n  },\n  \"embedding\": {\n    \"embeddingConnectionId\": \"<string>\",\n    \"model\": \"<string>\",\n    \"dimensions\": 32768,\n    \"batchSize\": 1024\n  },\n  \"vectorStore\": {\n    \"vectorStoreConnectionId\": \"<string>\",\n    \"indexName\": \"<string>\",\n    \"namespace\": \"<string>\",\n    \"metric\": \"cosine\"\n  },\n  \"description\": \"<string>\",\n  \"text\": {\n    \"textTemplate\": \"\"\n  },\n  \"processing\": {\n    \"parallelism\": 1,\n    \"checkpointIntervalMin\": 5,\n    \"chunkSize\": 512,\n    \"chunkOverlap\": 50\n  }\n}"

response = http.request(request)
puts response.read_body

{
  "_id": "<string>",
  "name": "<string>",
  "flink_job_name": "<string>",
  "flink_job_id": "<string>",
  "parallelism": 1,
  "error_message": "<string>",
  "agent_config": {},
  "created_by": "<string>",
  "created_timestamp": "2023-11-07T05:31:56Z",
  "updated_timestamp": "2023-11-07T05:31:56Z"
}

{
  "detail": [
    {
      "loc": [
        "<string>"
      ],
      "msg": "<string>",
      "type": "<string>",
      "input": "<unknown>",
      "ctx": {}
    }
  ]
}

Authorizations

Authorization

string

header

required

Bearer authentication header of the form Bearer <token>, where <token> is your auth token.

Query Parameters

deploy

boolean

default:true

Body

application/json

Request body for POST /knowledge-bases/config — create a knowledge base pipeline.

name

string

required

Knowledge base display name

Required string length: 1 - 200

source

KBSourceConfig · object

required

Kafka source configuration for the knowledge base pipeline.

Exactly one of topic or topicPattern must be set:

topic — exact Kafka topic name (the FE wizard's single-topic picker emits this). The service layer composes ^re.escape(topic)$ for the Java --topicMatcherRegex CLI arg, so the Java side keeps the regex-only contract.
topicPattern — escape hatch for tenants who genuinely need a multi-topic regex (e.g. fan-in of a sharded topic family). Not exposed by the FE wizard today — programmatic API callers only.

The XOR rule is enforced by _topic_xor below: empty payload (neither set) and over-specified payload (both set) both 422.

Show child attributes

embedding

KBEmbeddingConfig · object

required

Embedding-side configuration for a knowledge base.

Carries a reference to a saved AgentLlmConnection plus optional per-KB overrides of the connection's defaults. Credentials, provider, and the default model live on the connection — there is no inline apiKey / baseUrl path on a KB. The deploy-time resolver (resolve_kb_connections) reads the referenced connection's embedding defaults block, applies any per-KB overrides, and stamps the resulting bundle into the Flink CLI JSON.

The referenced connection must carry 'embedding' in its capabilities list and have a non-empty embedding.model — the validator inside resolve_saved_embedding_credentials raises 422 at deploy time otherwise so the FE error surfaces the cause without waiting for a Java runtime 4xx.

Show child attributes

vectorStore

KBVectorStoreConfig · object

required

Vector store configuration for the knowledge base.

References a lightweight AgentVectorStoreConnection by id. The BE resolves the connection's credentials (apiKey + endpoint) at deploy time. There is no inline-credentials path — KBs must reference a saved connection (the FE picker is the only authoring surface).

Show child attributes

description

string | null

Human-readable context

Maximum string length: 500

text

KBTextConfig · object

Text template configuration

Show child attributes

processing

KBProcessingConfig · object

Processing configuration for the knowledge base pipeline.

Show child attributes

Response

Successful Response

Response model for knowledge base entities. Mirrors FlinkJobResponse shape.

_id

string

required

name

string

required

job_type

enum<string>

required

Available options:

pyflink,

jar,

agent_config,

knowledge_base

status

enum<string>

required

Available options:

CREATED,

DEPLOYING,

RUNNING,

CANCELLING,

CANCELLED,

FAILED,

FINISHED

flink_job_name

string

required

desired_status

enum<string> | null

Available options:

CREATED,

DEPLOYING,

RUNNING,

CANCELLING,

CANCELLED,

FAILED,

FINISHED

flink_job_id

string | null

parallelism

integer

default:1

error_message

string | null

agent_config

Agent Config · object | null

KB config (secrets masked)

created_by

string | null

created_timestamp

string<date-time> | null

updated_timestamp

string<date-time> | null

⌘I

Authentication

Topics

Logs

Transforms

Services

Alerts

Usage

other

Dashboard

Destinations

Sources

Pipelines

Kafka Access

Project Keys

Consumer Groups

Schema Registry

Tags

Cluster Scaling

Agents

Knowledge Bases

Flink Jobs

Agents Observability

Create Knowledge Base

Authorizations

Query Parameters

Body

Response