Topics, messages and schemas

Every Kafka cluster’s page has a Manage Topics view, with three tabs:

Tab What you can do
Configuration List topics; create, change and delete them. Partitions can only grow
Messages Read messages on a topic, oldest or newest first, ten at a time. Reading never affects the applications consuming the topic
Schemas Browse the cluster’s Schema Registry: subjects, their versions and each schema. Read-only

Confluent Platform on VMs shows Configuration only.

Creating, changing and deleting topics is shown in every interface, including Terraform, on each platform’s page, for example Apache Kafka. Message contents are never written to logs.

The API examples below share this setup:

export STREAMTIME_TOKEN=<your-access-token>
export API=https://api.example.com/organizations/<org>

import os
from streamtime_sdk.client import AuthenticatedClient

client = AuthenticatedClient(
    base_url="https://api.example.com",
    token=os.environ["STREAMTIME_TOKEN"],
)
ORG = "<org>"

Uses the generated client in streamtime-sdk/typescript/client.

import { client } from "./client/client.gen";

client.setConfig({
  baseUrl: "https://api.example.com",
  headers: { Authorization: `Bearer ${process.env.STREAMTIME_TOKEN}` },
});
const org = "<org>";

package main

import (
    "bytes"
    "crypto/rand"
    "encoding/hex"
    "encoding/json"
    "net/http"
    "os"
)

const api = "https://api.example.com/organizations/<org>"

// newKey returns a random idempotency key.
func newKey() string {
    b := make([]byte, 16)
    rand.Read(b)
    return hex.EncodeToString(b)
}

// call sends one request with the access token and a JSON body (or none).
// An optional idempotency key is sent as the Idempotency-Key header.
func call(method, path string, body any, idempotencyKey ...string) (*http.Response, error) {
    var buf bytes.Buffer
    if body != nil {
        if err := json.NewEncoder(&buf).Encode(body); err != nil {
            return nil, err
        }
    }
    req, err := http.NewRequest(method, api+path, &buf)
    if err != nil {
        return nil, err
    }
    req.Header.Set("Authorization", "Bearer "+os.Getenv("STREAMTIME_TOKEN"))
    req.Header.Set("Content-Type", "application/json")
    if len(idempotencyKey) > 0 {
        req.Header.Set("Idempotency-Key", idempotencyKey[0])
    }
    return http.DefaultClient.Do(req)
}

Browse messages

  1. Open the cluster and select Manage Topics, then the Messages tab.
  2. Choose the topic, and Oldest first or Newest first.
  3. Each row shows the partition, offset, timestamp, key and value. Load more to page through.
curl "$API/kafka/<cluster>/topics/payments/messages/?position=latest" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Next page: pass the cursor the response returned.
curl "$API/kafka/<cluster>/topics/payments/messages/?cursor=<cursor>" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.kafka_topics import list_kafka_topic_messages
from streamtime_sdk.models import ListKafkaTopicMessagesPosition

page = list_kafka_topic_messages.sync(
    ORG, "<cluster>", "payments",
    client=client, position=ListKafkaTopicMessagesPosition.LATEST,
)

import { listKafkaTopicMessages } from "./client";

const { data: page } = await listKafkaTopicMessages({
  path: { org_identifier: org, kafka_id: "<cluster>", topic: "payments" },
  query: { position: "latest" },
});

resp, err := call("GET", "/kafka/<cluster>/topics/payments/messages/?position=latest", nil)
if err != nil {
    return err
}
defer resp.Body.Close()

Not available in the CLI yet.

Reading messages isn’t available in MCP. MCP can read a topic’s configuration: “What is the retention of the payments topic on the orders cluster?”

Browse schemas

  1. Open the cluster and select Manage Topics, then the Schemas tab.
  2. Choose a subject and a Version to see its schema.
curl "$API/kafka/<cluster>/schemas/subjects/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl "$API/kafka/<cluster>/schemas/subjects/payments-value/versions/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl "$API/kafka/<cluster>/schemas/subjects/payments-value/versions/latest/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.kafka_schemas import (
    get_kafka_schema_subject_version,
    list_kafka_schema_subjects,
)

subjects = list_kafka_schema_subjects.sync(ORG, "<cluster>", client=client)
schema = get_kafka_schema_subject_version.sync(
    ORG, "<cluster>", "payments-value", "latest", client=client
)

import { getKafkaSchemaSubjectVersion, listKafkaSchemaSubjects } from "./client";

const cluster = { org_identifier: org, kafka_id: "<cluster>" };
const { data: subjects } = await listKafkaSchemaSubjects({ path: cluster });
const { data: schema } = await getKafkaSchemaSubjectVersion({
  path: { ...cluster, subject: "payments-value", version: "latest" },
});

resp, err := call("GET", "/kafka/<cluster>/schemas/subjects/", nil)
if err != nil {
    return err
}
defer resp.Body.Close()

Not available in the CLI yet.

Not available in MCP.