create-schema
Generate and manage schemas for Apache Pinot using a structured JSON input, with options to override existing schemas or force creation when necessary.
Instructions
Create a new schema
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| force | No | ||
| override | No | ||
| schemaJson | Yes |
Implementation Reference
- mcp_pinot/server.py:133-144 (handler)MCP tool handler for 'create_schema'. This function is decorated with @mcp.tool decorator which registers it as an MCP tool. It delegates to the pinot_client.create_schema method.
@mcp.tool def create_schema(schemaJson: str, override: bool = True, force: bool = False) -> str: """Create a new schema""" try: results = pinot_client.create_schema( schemaJson, override, force, ) return json.dumps(results, indent=2) except Exception as e: return f"Error: {str(e)}" - mcp_pinot/pinot_client.py:386-413 (helper)Helper method in PinotClient class that performs the HTTP POST request to the Pinot controller endpoint to create a new schema.
def create_schema( self, schemaJson: str, override: bool = True, force: bool = False, ) -> dict[str, Any]: url = f"{self.config.controller_url}/{PinotEndpoints.SCHEMAS}" params = {"override": str(override).lower(), "force": str(force).lower()} headers = self._create_auth_headers() headers["Content-Type"] = "application/json" response = requests.post( url, headers=headers, params=params, data=schemaJson, timeout=(self.config.connection_timeout, self.config.request_timeout), verify=True, ) response.raise_for_status() try: return response.json() except requests.exceptions.JSONDecodeError: return { "status": "success", "message": "Schema creation request processed.", "response_body": response.text, }