MCP Kafka Schema Reg

Instructions

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,