OCI Kafka MCP Server

An AI-native control interface for OCI Streaming with Apache Kafka, built on the Model Context Protocol (MCP) 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 (recommended) or pip

Install

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

Run with local Kafka (development, Podman)

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

Set environment variables up front (optional)
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:

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:

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, Cursor, and MCPHost. See the Oracle MCP client configuration guide 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:

{
  "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):

{
  "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):

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

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

# 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 for the full security architecture document, including threat model, dependency audit, and deployment architecture.

License

Apache-2.0