create_context
Create a new schema context for Kafka Schema Registry to organize and manage schema definitions effectively. Simplify schema governance and ensure compatibility across messaging systems.
Instructions
Create a new schema context.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| context | Yes | ||
| registry | No |
Implementation Reference
- core_registry_tools.py:1547-1620 (handler)The core handler function that implements the 'create_context' tool logic. Sends POST to /contexts/{context} in Kafka Schema Registry to create the context, handles registry modes, viewonly checks, error handling, and adds MCP-standard metadata and HATEOAS links.@structured_output("create_context", fallback_on_error=True) def create_context_tool( context: str, registry_manager, registry_mode: str, registry: Optional[str] = None, auth=None, headers=None, schema_registry_url: str = "", ) -> Dict[str, str]: """ Create a new schema context. Args: context: The context name to create registry: Optional registry name (ignored in single-registry mode) Returns: Success message with structured validation and resource links """ # Check viewonly mode viewonly_check = _check_viewonly_mode(registry_manager, registry) if viewonly_check: return validate_registry_response(viewonly_check, registry_mode) try: if registry_mode == "single": # Single-registry mode: use secure session approach client = registry_manager.get_registry() if client is None: return create_error_response( "No default registry configured", error_code="REGISTRY_NOT_FOUND", registry_mode="single", ) response = client.session.post( f"{client.config.url}/contexts/{context}", auth=client.auth, headers=client.headers ) response.raise_for_status() result = create_success_response(f"Context '{context}' created successfully", registry_mode="single") # Add resource links registry_name = _get_registry_name(registry_mode, registry) result = add_links_to_response(result, "context", registry_name, context=context) return result else: # Multi-registry mode: use client approach client = registry_manager.get_registry(registry) if client is None: return create_error_response( f"Registry '{registry}' not found", error_code="REGISTRY_NOT_FOUND", registry_mode="multi", ) response = client.session.post( f"{client.config.url}/contexts/{context}", auth=client.auth, headers=client.headers ) response.raise_for_status() result = create_success_response( f"Context '{context}' created successfully", data={"registry": client.config.name}, registry_mode="multi", ) # Add resource links result = add_links_to_response(result, "context", client.config.name, context=context) return result except Exception as e: return create_error_response(str(e), error_code="CONTEXT_CREATE_FAILED", registry_mode=registry_mode)
- schema_definitions.py:480-493 (schema)JSON Schema definition for the output of the create_context tool (CONTEXT_OPERATION_SCHEMA), used for structured validation.CONTEXT_OPERATION_SCHEMA = { "type": "object", "properties": { "message": {"type": "string", "description": "Operation result message"}, "context": {"type": "string", "description": "Context name"}, "registry": { "type": "string", "description": "Registry name (multi-registry mode)", }, **METADATA_FIELDS, }, "required": ["message"], "additionalProperties": True, }
- schema_definitions.py:970-972 (registration)Registration of the create_context tool's output schema in the master TOOL_OUTPUT_SCHEMAS dictionary, mapping tool name to its output schema for MCP compliance."list_contexts": LIST_CONTEXTS_SCHEMA, "create_context": CONTEXT_OPERATION_SCHEMA, "delete_context": CONTEXT_OPERATION_SCHEMA,