> ## Documentation Index
> Fetch the complete documentation index at: https://docs.pinkfish.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# kafka

> Apache Kafka (Amazon MSK) integration server

**Server path:** `/kafka` | **Type:** Application | **PCID required:** Yes

## Tools

| Tool                                                              | Description                                                                                            |
| ----------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| [`kafka_consume_messages`](#kafka_consume_messages)               | Consume (poll) messages from a Kafka topic using an ephemeral consumer group. Does not commit offsets. |
| [`kafka_create_topic`](#kafka_create_topic)                       | Create a new Kafka topic                                                                               |
| [`kafka_delete_topic`](#kafka_delete_topic)                       | Delete a Kafka topic                                                                                   |
| [`kafka_describe_cluster`](#kafka_describe_cluster)               | Get Kafka cluster information including brokers and controller                                         |
| [`kafka_describe_consumer_group`](#kafka_describe_consumer_group) | Get detailed information about a Kafka consumer group including members and state                      |
| [`kafka_describe_topic`](#kafka_describe_topic)                   | Get detailed information about a Kafka topic including partitions                                      |
| [`kafka_list_consumer_groups`](#kafka_list_consumer_groups)       | List all consumer groups in the Kafka cluster                                                          |
| [`kafka_list_topics`](#kafka_list_topics)                         | List all topics in the Kafka cluster                                                                   |
| [`kafka_produce_messages`](#kafka_produce_messages)               | Produce (publish) messages to a Kafka topic                                                            |

***

## kafka\_consume\_messages

Consume (poll) messages from a Kafka topic using an ephemeral consumer group. Does not commit offsets.

**Parameters:**

| Parameter        | Type    | Required | Default | Description                                                                                         |
| ---------------- | ------- | -------- | ------- | --------------------------------------------------------------------------------------------------- |
| `topic`          | string  | Yes      | —       | Topic name to consume messages from                                                                 |
| `max_messages`   | number  | No       | `10`    | Maximum number of messages to return (default: 10, max: 100)                                        |
| `timeout_ms`     | number  | No       | `5000`  | Poll timeout in milliseconds (default: 5000, max: 30000)                                            |
| `from_beginning` | boolean | No       | —       | If true, start from the earliest available offset; otherwise start from the latest (default: false) |

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      },
      "topic": {
        "type": "string",
        "description": "Topic name to consume messages from"
      },
      "max_messages": {
        "type": "number",
        "default": 10,
        "description": "Maximum number of messages to return (default: 10, max: 100)"
      },
      "timeout_ms": {
        "type": "number",
        "default": 5000,
        "description": "Poll timeout in milliseconds (default: 5000, max: 30000)"
      },
      "from_beginning": {
        "type": "boolean",
        "description": "If true, start from the earliest available offset; otherwise start from the latest (default: false)"
      }
    },
    "required": [
      "PCID",
      "topic"
    ]
  }
  ```
</Expandable>

***

## kafka\_create\_topic

Create a new Kafka topic

**Parameters:**

| Parameter            | Type   | Required | Default | Description                                                                |
| -------------------- | ------ | -------- | ------- | -------------------------------------------------------------------------- |
| `topic`              | string | Yes      | —       | Name of the topic to create                                                |
| `num_partitions`     | number | No       | `1`     | Number of partitions (default: 1)                                          |
| `replication_factor` | number | No       | `1`     | Replication factor (default: 1)                                            |
| `configs`            | object | No       | `{}`    | Optional topic configuration overrides (e.g. retention.ms, cleanup.policy) |

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      },
      "topic": {
        "type": "string",
        "description": "Name of the topic to create"
      },
      "num_partitions": {
        "type": "number",
        "default": 1,
        "description": "Number of partitions (default: 1)"
      },
      "replication_factor": {
        "type": "number",
        "default": 1,
        "description": "Replication factor (default: 1)"
      },
      "configs": {
        "type": "object",
        "additionalProperties": true,
        "default": {},
        "description": "Optional topic configuration overrides (e.g. retention.ms, cleanup.policy)"
      }
    },
    "required": [
      "PCID",
      "topic"
    ]
  }
  ```
</Expandable>

***

## kafka\_delete\_topic

Delete a Kafka topic

**Parameters:**

| Parameter | Type   | Required | Default | Description          |
| --------- | ------ | -------- | ------- | -------------------- |
| `topic`   | string | Yes      | —       | Topic name to delete |

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      },
      "topic": {
        "type": "string",
        "description": "Topic name to delete"
      }
    },
    "required": [
      "PCID",
      "topic"
    ]
  }
  ```
</Expandable>

***

## kafka\_describe\_cluster

Get Kafka cluster information including brokers and controller

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      }
    },
    "required": [
      "PCID"
    ]
  }
  ```
</Expandable>

***

## kafka\_describe\_consumer\_group

Get detailed information about a Kafka consumer group including members and state

**Parameters:**

| Parameter  | Type   | Required | Default | Description                   |
| ---------- | ------ | -------- | ------- | ----------------------------- |
| `group_id` | string | Yes      | —       | Consumer group ID to describe |

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      },
      "group_id": {
        "type": "string",
        "description": "Consumer group ID to describe"
      }
    },
    "required": [
      "PCID",
      "group_id"
    ]
  }
  ```
</Expandable>

***

## kafka\_describe\_topic

Get detailed information about a Kafka topic including partitions

**Parameters:**

| Parameter | Type   | Required | Default | Description            |
| --------- | ------ | -------- | ------- | ---------------------- |
| `topic`   | string | Yes      | —       | Topic name to describe |

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      },
      "topic": {
        "type": "string",
        "description": "Topic name to describe"
      }
    },
    "required": [
      "PCID",
      "topic"
    ]
  }
  ```
</Expandable>

***

## kafka\_list\_consumer\_groups

List all consumer groups in the Kafka cluster

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      }
    },
    "required": [
      "PCID"
    ]
  }
  ```
</Expandable>

***

## kafka\_list\_topics

List all topics in the Kafka cluster

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      }
    },
    "required": [
      "PCID"
    ]
  }
  ```
</Expandable>

***

## kafka\_produce\_messages

Produce (publish) messages to a Kafka topic

**Parameters:**

| Parameter  | Type      | Required | Default | Description                       |
| ---------- | --------- | -------- | ------- | --------------------------------- |
| `topic`    | string    | Yes      | —       | Topic name to publish messages to |
| `messages` | object\[] | Yes      | —       | Array of messages to produce      |

<Expandable title="inputSchema">
  ```json theme={null}
  {
    "type": "object",
    "properties": {
      "PCID": {
        "type": "string",
        "description": "Pink Connect ID"
      },
      "topic": {
        "type": "string",
        "description": "Topic name to publish messages to"
      },
      "messages": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "key": {
              "type": "string",
              "description": "Optional message key for partitioning"
            },
            "value": {
              "oneOf": [
                {
                  "type": "string"
                },
                {
                  "type": "object",
                  "additionalProperties": true
                },
                {
                  "type": "null"
                }
              ],
              "description": "Message value — string, JSON object, or null (tombstone for compacted topics)"
            },
            "headers": {
              "type": "object",
              "additionalProperties": true,
              "description": "Optional message headers as key-value pairs"
            }
          }
        },
        "description": "Array of messages to produce"
      }
    },
    "required": [
      "PCID",
      "topic",
      "messages"
    ]
  }
  ```
</Expandable>