# OCI Kafka MCP Server

An AI-native control interface for **OCI Streaming with Apache Kafka**, built on the [Model Context Protocol (MCP)](https://modelcontextprotocol.io) specification.

This MCP server enables LLM agents (Claude, GPT, etc.) to securely manage Kafka clusters through structured tool execution — with built-in safety guardrails, audit logging, and enterprise-grade security.

## Features

- **42 structured tools** for cluster, topic, consumer, observability, AI diagnostics, OCI metadata, cluster lifecycle, cluster configuration, and work request operations
- **Read-only by default** — write tools require explicit `--allow-writes` flag
- **Policy guard** — every tool is risk-classified (LOW/MEDIUM/HIGH); destructive operations require confirmation
- **AI diagnostic tools** — orchestrate multiple Kafka operations to produce scaling recommendations and lag root cause analyses
- **Circuit breaker** — prevents cascading failures when Kafka is unavailable
- **Structured audit logging** — every tool execution logged as JSON with timestamp, input hash, and duration
- **SASL/SCRAM-SHA-512 + TLS** — enterprise security from day one
- **Private networking** — designed for OCI private endpoints

## Quick Start

### Prerequisites

- Python 3.11+
- [uv](https://docs.astral.sh/uv/) (recommended) or pip

### Install

```bash
git clone <repo-url>
cd oci-kafka-mcp-server
uv sync
```

### Run with local Kafka (development, Podman)

```bash
# Start a local Kafka broker
podman compose -f docker/docker-compose.yaml up -d

# Run the MCP server (read-only mode)
uv run oci-kafka-mcp

# Run with write tools enabled
uv run oci-kafka-mcp --allow-writes

# Stop local Kafka
podman compose -f docker/docker-compose.yaml down
```

### Configure for OCI Streaming

You can configure OCI Kafka in either of these ways:

1. **Set environment variables up front** (optional)
2. **Leave variables unset** and let the MCP server request the required values at runtime, then call `oci_kafka_configure_connection`

If you want to pre-configure with environment variables:

```bash
export KAFKA_BOOTSTRAP_SERVERS="bootstrap-clstr-XXXXX.kafka.us-chicago-1.oci.oraclecloud.com:9092"
export KAFKA_SECURITY_PROTOCOL="SASL_SSL"
export KAFKA_SASL_MECHANISM="SCRAM-SHA-512"
export KAFKA_SASL_USERNAME="your-username"
export KAFKA_SASL_PASSWORD="your-password"
export KAFKA_SSL_CA_LOCATION="/path/to/ca.pem"

uv run oci-kafka-mcp
```

Or use the OCI template file:

```bash
cp .env.oci.example .env.oci
# edit .env.oci with your cluster values
source .env.oci
uv run oci-kafka-mcp
```

> Note: `KAFKA_*` variables are **not mandatory** at server startup. If not set, tools will guide the agent/user to provide connection details and use `oci_kafka_configure_connection` before data-plane operations.

### Use with an MCP Client

This server works with any MCP-compatible client. Oracle recommends [Cline](https://github.com/cline/cline), [Cursor](https://www.cursor.com/), and [MCPHost](https://github.com/oracle/mcp). See the [Oracle MCP client configuration guide](https://github.com/oracle/mcp/tree/main?tab=readme-ov-file#client-configuration) for details.

The `env` block below is optional — if omitted, the server will prompt the agent to call `oci_kafka_configure_connection` with your cluster details at runtime.

#### Cline (VS Code extension)

Add to your Cline MCP settings:

```json
{
  "mcpServers": {
    "oci-kafka": {
      "type": "stdio",
      "command": "/path/to/oci-kafka-mcp-server/.venv/bin/oci-kafka-mcp",
      "args": ["--allow-writes"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "your-bootstrap:9092",
        "KAFKA_SECURITY_PROTOCOL": "SASL_SSL",
        "KAFKA_SASL_MECHANISM": "SCRAM-SHA-512",
        "KAFKA_SASL_USERNAME": "your-username",
        "KAFKA_SASL_PASSWORD": "your-password"
      }
    }
  }
}
```

#### Cursor

Add to `.cursor/mcp.json` (project-level) or `~/.cursor/mcp.json` (global):

```json
{
  "mcpServers": {
    "oci-kafka": {
      "type": "stdio",
      "command": "/path/to/oci-kafka-mcp-server/.venv/bin/oci-kafka-mcp",
      "args": ["--allow-writes"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "your-bootstrap:9092",
        "KAFKA_SECURITY_PROTOCOL": "SASL_SSL",
        "KAFKA_SASL_MECHANISM": "SCRAM-SHA-512",
        "KAFKA_SASL_USERNAME": "your-username",
        "KAFKA_SASL_PASSWORD": "your-password"
      }
    }
  }
}
```

#### MCPHost

Add to your MCPHost configuration file (e.g., `~/.mcphost.json`):

```json
{
  "mcpServers": {
    "oci-kafka": {
      "type": "stdio",
      "command": "/path/to/oci-kafka-mcp-server/.venv/bin/oci-kafka-mcp",
      "args": ["--allow-writes"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "your-bootstrap:9092",
        "KAFKA_SECURITY_PROTOCOL": "SASL_SSL",
        "KAFKA_SASL_MECHANISM": "SCRAM-SHA-512",
        "KAFKA_SASL_USERNAME": "your-username",
        "KAFKA_SASL_PASSWORD": "your-password"
      }
    }
  }
}
```

Then start MCPHost with:

```bash
mcphost -m ollama:<model> --config ~/.mcphost.json
```

## Available Tools (42)

### Connection Management

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_configure_connection` | Set or update Kafka cluster connection details at runtime (no restart needed) | LOW |
| `oci_kafka_get_connection_info` | Show current connection config with masked password | LOW |

### Cluster Operations

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_get_cluster_health` | Broker status, controller ID, topic count | LOW |
| `oci_kafka_get_cluster_config` | Broker-level Kafka configuration settings | LOW |

### Topic Operations

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_list_topics` | List all topics | LOW |
| `oci_kafka_describe_topic` | Partition details, leaders, replicas, ISR, topic config | LOW |
| `oci_kafka_create_topic` | Create a topic with partitions and replication factor | MEDIUM |
| `oci_kafka_update_topic_config` | Update topic configuration (retention, compaction, etc.) | MEDIUM |
| `oci_kafka_delete_topic` | Delete a topic (requires confirmation) | HIGH |

### Consumer Operations

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_list_consumer_groups` | List all consumer groups | LOW |
| `oci_kafka_describe_consumer_group` | Group state, members, coordinator, partition assignments | LOW |
| `oci_kafka_get_consumer_lag` | Per-partition lag, committed offsets, end offsets | LOW |
| `oci_kafka_reset_consumer_offset` | Reset offsets to earliest/latest/specific offset (requires confirmation) | HIGH |
| `oci_kafka_delete_consumer_group` | Delete a consumer group (requires confirmation) | HIGH |

### Observability

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_get_partition_skew` | Detect partition leader imbalance across brokers | LOW |
| `oci_kafka_detect_under_replicated_partitions` | Find partitions where ISR count < replica count | LOW |

### AI Diagnostics

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_recommend_scaling` | Orchestrates health, skew, and replication data into scaling recommendations | LOW |
| `oci_kafka_analyze_lag_root_cause` | Correlates consumer state, lag, and topology into root cause analysis | LOW |

### OCI Control Plane Metadata

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_list_oci_clusters` | List all Kafka clusters in an OCI compartment (auto-discovers compartment) | LOW |
| `oci_kafka_get_oci_cluster_info` | Cluster OCID, lifecycle state, broker shape, bootstrap URLs, tags | LOW |

### Cluster Lifecycle (OCI Control Plane)

Async operations — returns a work request OCID; use `oci_kafka_get_work_request` to poll for completion.

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_create_cluster` | Provision a new OCI Kafka cluster (requires confirmation) | HIGH |
| `oci_kafka_update_cluster` | Update cluster display name, tags, or applied configuration | MEDIUM |
| `oci_kafka_scale_cluster` | Scale broker count for an existing cluster (requires confirmation) | HIGH |
| `oci_kafka_delete_cluster` | Permanently delete a cluster and all its data (requires confirmation) | HIGH |
| `oci_kafka_change_cluster_compartment` | Move a cluster to a different OCI compartment (requires confirmation) | HIGH |
| `oci_kafka_enable_superuser` | Grant full administrative access to the cluster's superuser | MEDIUM |
| `oci_kafka_disable_superuser` | Revoke superuser access to restore least-privilege | MEDIUM |

### Cluster Configuration (OCI Control Plane)

Named, versioned sets of Kafka broker settings that can be applied to one or more clusters.

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_list_cluster_configs` | List all cluster configurations in a compartment | LOW |
| `oci_kafka_get_oci_cluster_config` | Get a cluster configuration and its latest version | LOW |
| `oci_kafka_create_cluster_config` | Create a new named cluster configuration | MEDIUM |
| `oci_kafka_update_cluster_config` | Update a config's display name or tags | MEDIUM |
| `oci_kafka_delete_cluster_config` | Delete a configuration and all its versions (requires confirmation) | HIGH |
| `oci_kafka_change_cluster_config_compartment` | Move a configuration to a different compartment | MEDIUM |
| `oci_kafka_list_cluster_config_versions` | List all versions of a cluster configuration | LOW |
| `oci_kafka_get_cluster_config_version` | Get a specific version of a cluster configuration | LOW |
| `oci_kafka_delete_cluster_config_version` | Delete a specific configuration version | MEDIUM |

### Work Requests & Node Shapes (OCI Control Plane)

Track asynchronous OCI operations returned by cluster lifecycle and configuration tools.

| Tool | Description | Risk |
|------|-------------|------|
| `oci_kafka_get_work_request` | Poll status and progress of an async OCI operation | LOW |
| `oci_kafka_list_work_requests` | List work requests by compartment or resource OCID | LOW |
| `oci_kafka_get_work_request_errors` | Get error details from a failed work request | LOW |
| `oci_kafka_get_work_request_logs` | Get timestamped log entries from a work request | LOW |
| `oci_kafka_cancel_work_request` | Cancel an in-progress work request | MEDIUM |
| `oci_kafka_list_node_shapes` | List available broker node shapes for cluster provisioning | LOW |

## Safety Model

| Risk Level | Behavior | Examples |
|------------|----------|----------|
| **LOW** | Always allowed | Health checks, list/describe operations |
| **MEDIUM** | Requires `--allow-writes` | Create topic, update config |
| **HIGH** | Requires `--allow-writes` + confirmation | Delete topic, reset offsets, cluster lifecycle |

## Development

```bash
# Run tests (92 tests, all unit — no Kafka broker needed)
uv run pytest

# Run tests with coverage
uv run pytest --cov=oci_kafka_mcp --cov-report=term-missing

# Lint
uv run ruff check src/ tests/

# Format
uv run ruff format src/ tests/

# Type check
uv run mypy src/
```

## Architecture

See [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) for the full security architecture document, including threat model, dependency audit, and deployment architecture.

## License

Apache-2.0