create_topic_with_schema

Create Kafka topics with configurable formats and schemas for structured data streaming. Define partitions, replication, and data formats like AVRO or JSON.

Instructions

Creates a new Kafka topic with optional format and schema configuration.

Args: environment: The environment name. name: Topic name. partitions: Number of partitions (default: 1). replication: Replication factor (default: 1). configs: Topic configurations. key_format: Key format (AVRO, JSON, CSV, XML, INT, LONG, STRING, BYTES, etc.). key_schema: Key schema (required for AVRO, JSON, CSV, XML). value_format: Value format. value_schema: Value schema.

Returns: Creation result.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`environment`	Yes
`name`	Yes
`partitions`	No
`replication`	No
`configs`	No
`key_format`	No
`key_schema`	No
`value_format`	No
`value_schema`	No

Implementation Reference

src/lenses_mcp/tools/topics.py:98-154 (handler)

The handler function decorated with @mcp.tool(), implementing the logic to create a Kafka topic with schema support via HTTP API request to Lenses.

@mcp.tool()
async def create_topic_with_schema(
    environment: str,
    name: str,
    partitions: int = 1,
    replication: int = 1,
    configs: Optional[Dict[str, str]] = None,
    key_format: Optional[str] = None,
    key_schema: Optional[str] = None,
    value_format: Optional[str] = None,
    value_schema: Optional[str] = None
) -> Dict[str, Any]:
    """
    Creates a new Kafka topic with optional format and schema configuration.
    
    Args:
        environment: The environment name.
        name: Topic name.
        partitions: Number of partitions (default: 1).
        replication: Replication factor (default: 1).
        configs: Topic configurations.
        key_format: Key format (AVRO, JSON, CSV, XML, INT, LONG, STRING, BYTES, etc.).
        key_schema: Key schema (required for AVRO, JSON, CSV, XML).
        value_format: Value format.
        value_schema: Value schema.
    
    Returns:
        Creation result.
    """
    payload = {
        "name": name,
        "partitions": partitions,
        "replication": replication
    }
    
    if configs:
        payload["configs"] = configs
    else:
        payload["configs"] = {}
    
    if key_format or value_format:
        format_config = {}
        if key_format:
            format_config["key"] = {"format": key_format}
            if key_schema:
                format_config["key"]["schema"] = key_schema
        if value_format:
            format_config["value"] = {"format": value_format}
            if value_schema:
                format_config["value"]["schema"] = value_schema
        payload["format"] = format_config
    
    endpoint = f"/api/v1/environments/{environment}/proxy/api/v1/kafka/topic"
    try:
        return await api_client._make_request("POST", endpoint, payload)
    except Exception as e:
        raise ToolError(f"Topic creation failed: {e}")

src/lenses_mcp/server.py:33-33 (registration)
Invocation of register_topics(mcp) which defines and registers the create_topic_with_schema tool among others.
```
register_topics(mcp)
```
src/lenses_mcp/tools/topics.py:10-10 (registration)
Function that contains the definition of all topic tools including create_topic_with_schema, which are registered when this function is called.
```
def register_topics(mcp: FastMCP):
```

Lenses MCP Server